diff --git a/backend/alembic/versions/cd7de470586e_v1_db_structure.py b/backend/alembic/versions/cd7de470586e_v1_db_structure.py index d1eac36f..67488e4b 100644 --- a/backend/alembic/versions/cd7de470586e_v1_db_structure.py +++ b/backend/alembic/versions/cd7de470586e_v1_db_structure.py @@ -94,7 +94,7 @@ def upgrade() -> None: sa.Column("frontend_post_id", sa.String(200), nullable=False), # unique together with api_client_id sa.Column("created_date", sa.DateTime(), nullable=False, server_default=sa.func.current_timestamp()), sa.Column("payload_type", sa.String(200), nullable=False), # deserialization hint & dbg aid - sa.Column("payload", JSONB(astext_type=sa.Text()), nullable=False), + sa.Column("payload", JSONB(astext_type=sa.Text()), nullable=True), sa.PrimaryKeyConstraint("id"), sa.ForeignKeyConstraint(["person_id"], ["person.id"]), sa.ForeignKeyConstraint(["api_client_id"], ["api_client.id"]), diff --git a/backend/app/api/v1/tasks2.py b/backend/app/api/v1/tasks2.py index 23986fef..4a359ab0 100644 --- a/backend/app/api/v1/tasks2.py +++ b/backend/app/api/v1/tasks2.py @@ -4,7 +4,7 @@ from typing import Any from uuid import UUID from app.api import deps -from app.prompt_repository import PromptRepository +from app.prompt_repository import PromptRepository, TaskPayload from app.schemas import protocol as protocol_schema from fastapi import APIRouter, Depends, HTTPException from fastapi.security.api_key import APIKey @@ -116,15 +116,24 @@ def acknowledge_task( """ The frontend acknowledges a task. """ - deps.api_auth(api_key, db) + api_client = deps.api_auth(api_key, db) + pr = PromptRepository(db, api_client, user=None) match (type(response)): case protocol_schema.PostCreatedTaskResponse: logger.info(f"Frontend acknowledged {task_id=} and created {response.post_id=}.") - # here we would store the post id in the database for the task + + if response.status == "success": + # here we store the post id in the database for the task + pr.bind_frontend_post_id(task_id=task_id, post_id=response.post_id) + case protocol_schema.RatingCreatedTaskResponse: logger.info(f"Frontend acknowledged {task_id=} for {response.post_id=}.") - # here we would store the rating id in the database for the task + + if response.status == "success": + # here we would store the rating id in the database for the task + pr.bind_frontend_post_id(task_id=task_id, post_id=response.post_id) + case _: raise HTTPException( status_code=HTTP_400_BAD_REQUEST, @@ -144,14 +153,22 @@ def post_interaction( """ The frontend reports an interaction. """ - deps.api_auth(api_key, db) + api_client = deps.api_auth(api_key, db) + pr = PromptRepository(db, api_client, user=interaction.user) match (type(interaction)): case protocol_schema.TextReplyToPost: logger.info( f"Frontend reports text reply to {interaction.post_id=} with {interaction.text=} by {interaction.user=}." ) - # here we would store the text reply in the database + + work_package = pr.fetch_workpackage_by_postid(interaction.post_id) + work_payload: TaskPayload = work_package.payload.payload + logger.info(f"found task work package in db: {work_payload}") + + # here we store the text reply in the database + pr.store_text_reply(interaction) + return protocol_schema.TaskDone( reply_to_post_id=interaction.user_post_id, addressed_user=interaction.user, diff --git a/backend/app/models/post.py b/backend/app/models/post.py index fb6d5160..d1569a67 100644 --- a/backend/app/models/post.py +++ b/backend/app/models/post.py @@ -30,4 +30,4 @@ class Post(SQLModel, table=True): sa_column=sa.Column(sa.DateTime(), nullable=False, server_default=sa.func.current_timestamp()) ) payload_type: str = Field(nullable=False, max_length=200) - payload: PayloadContainer = Field(sa_column=sa.Column(payload_column_type(PayloadContainer), nullable=False)) + payload: PayloadContainer = Field(sa_column=sa.Column(payload_column_type(PayloadContainer), nullable=True)) diff --git a/backend/app/prompt_repository.py b/backend/app/prompt_repository.py index 539ef1ab..d2145e2e 100644 --- a/backend/app/prompt_repository.py +++ b/backend/app/prompt_repository.py @@ -1,9 +1,9 @@ # -*- coding: utf-8 -*- +from datetime import datetime from typing import Literal, Optional -from uuid import UUID +from uuid import UUID, uuid4 -# from app.models import ApiClient, Person, PersonStats, Post, PostReaction, WorkPackage -from app.models import ApiClient, Person, WorkPackage +from app.models import ApiClient, Person, Post, WorkPackage from app.models.payload_column_type import PayloadContainer, payload_tpye from app.schemas import protocol as protocol_schema from pydantic import BaseModel @@ -74,6 +74,107 @@ class PromptRepository: self.db.commit() return person + def validate_post_id(self, post_id: str) -> None: + if not isinstance(post_id, str): + raise TypeError("post_id must be string") + if not post_id: + raise ValueError("post_id must not be empty") + + def bind_frontend_post_id(self, task_id: UUID, post_id: str): + self.validate_post_id(post_id) + + # find work package + work_pack: WorkPackage = ( + self.db.query(WorkPackage) + .filter(WorkPackage.id == task_id and WorkPackage.api_client_id == self.api_client.id) + .first() + ) + if work_pack is None: + raise RuntimeError(f"WorkPackage for task {task_id} not found") + if work_pack.expiry_date is not None and datetime.utcnow() > work_pack.expiry_date: + raise RuntimeError("WorkPackage already expired.") + + # ToDo: check race-condition, transaction + + # check if task thread exits + thread_root = ( + self.db.query(Post) + .filter( + Post.workpackage_id == work_pack.id + and Post.frontend_post_id == post_id + and Post.parent_id is None + and self.api_client == self.api_client + ) + .one_or_none() + ) + if thread_root is None: + thread_id = uuid4() + thread_root = Post( + id=thread_id, + thread_id=thread_id, + role="system", + person_id=work_pack.person_id, + workpackage_id=work_pack.id, + frontend_post_id=post_id, + api_client_id=self.api_client.id, + payload_type="bind", + ) + self.db.add(thread_root) + self.db.commit() + self.db.refresh(thread_root) + return thread_root + + def fetch_workpackage_by_postid(self, post_id: str) -> WorkPackage: + self.validate_post_id(post_id) + post: Post = ( + self.db.query(Post) + .filter(Post.api_client_id == self.api_client.id and Post.frontend_post_id == post_id) + .one_or_none() + ) + if post is None: + raise RuntimeError(f"Post with post_id {post_id} not found.") + + work_pack = self.db.query(WorkPackage).filter(WorkPackage.id == post.workpackage_id).one() + return work_pack + + def store_text_reply(self, reply: protocol_schema.TextReplyToPost) -> Post: + self.validate_post_id(reply.post_id) + self.validate_post_id(reply.user_post_id) + + # find post with post-id + parent_post: Post = ( + self.db.query(Post) + .filter( + Post.api_client_id == self.api_client.id + and Post.frontend_post_id == reply.post_id + and Post.person_id == self.person_id + ) + .one_or_none() + ) + if parent_post is None: + raise RuntimeError(f"Post for post_id {reply.post_id} not found.") + + # create reply post + user_post_id = uuid4() + # ToDo: role user or agent? + user_post = Post( + id=user_post_id, + parent_id=parent_post.id, + thread_id=parent_post.thread_id, + workpackage_id=parent_post.workpackage_id, + person_id=self.person_id, + role="unknown", + frontend_post_id=reply.user_post_id, + api_client_id=self.api_client.id, + ) + self.db.add(user_post) + self.db.commit() + self.db.refresh(user_post) + return user_post + + def store_rating(self, rating: protocol_schema.PostRating) -> Post: + pass + def store_task(self, task: protocol_schema.Task) -> WorkPackage: payload: TaskPayload = None match type(task):