| 123456789101112131415161718192021222324252627282930313233343536373839404142 |
- from datetime import datetime
- from typing import TYPE_CHECKING
- from pydantic import BaseModel, Field
- from core_domain import HumanTaskContract, HumanTaskStatus, HumanTaskType
- from core_shared import JSONValue
- if TYPE_CHECKING:
- from app.db.models import HumanTask
- class HumanTaskCreateRequest(BaseModel):
- tenant_id: str
- task_type: HumanTaskType
- title: str
- description: str | None = None
- source_type: str | None = None
- source_id: str | None = None
- run_id: str | None = None
- node_run_id: str | None = None
- requested_by: str | None = None
- assigned_to: str | None = None
- request_payload_json: dict[str, JSONValue] = Field(default_factory=dict)
- due_time: datetime | None = None
- class HumanTaskClaimRequest(BaseModel):
- tenant_id: str
- claimed_by: str
- class HumanTaskCompleteRequest(BaseModel):
- tenant_id: str
- status: HumanTaskStatus
- response_payload_json: dict[str, JSONValue] = Field(default_factory=dict)
- class HumanTaskResponse(HumanTaskContract):
- @classmethod
- def from_entity(cls, entity: "HumanTask") -> "HumanTaskResponse":
- return cls.model_validate(entity, from_attributes=True)
|
