agent-platform/services/session-service/app/schemas/run_request.py

from datetime import datetime
from typing import TYPE_CHECKING

from pydantic import BaseModel, Field
from core_domain import InitialNodeContract, RunBootstrapContract, RunCreateContract
from core_shared import JSONValue

if TYPE_CHECKING:
    from app.db.models import RunRequest


class InitialNodeRequest(InitialNodeContract):
    pass


class RunRequestCreateRequest(BaseModel):
    tenant_id: str
    session_id: str
    app_version_id: str
    workflow_version_id: str
    trigger_type: str = "chat"
    request_payload_json: dict[str, JSONValue] = Field(default_factory=dict)
    request_status: str = "accepted"


class RunRequestResponse(BaseModel):
    id: str
    tenant_id: str
    session_id: str
    app_version_id: str
    workflow_version_id: str
    trigger_type: str
    request_payload_json: dict[str, JSONValue] | None = None
    request_status: str
    created_time: datetime

    @classmethod
    def from_entity(cls, entity: "RunRequest") -> "RunRequestResponse":
        return cls.model_validate(entity, from_attributes=True)


class DispatchRunRequest(BaseModel):
    tenant_id: str
    session_id: str
    app_id: str
    app_version_id: str
    workflow_id: str
    workflow_version_id: str
    trigger_type: str = "chat"
    priority: int = 0
    request_payload_json: dict[str, JSONValue] = Field(default_factory=dict)
    initial_node: InitialNodeRequest | None = None


class DispatchRunResponse(BaseModel):
    run_request: RunRequestResponse
    runtime_response: RunBootstrapContract

    @classmethod
    def from_parts(
        cls,
        *,
        run_request: "RunRequest",
        runtime_response: RunBootstrapContract,
    ) -> "DispatchRunResponse":
        return cls(
            run_request=RunRequestResponse.from_entity(run_request),
            runtime_response=runtime_response,
        )