agent-platform/services/session-service/app/application/services.py

from app.db.models import Message, RunRequest
from app.db.models import Session as SessionModel
from app.domain.repositories import MessageRepository, RunRequestRepository, SessionRepository
from app.schemas.message import MessageCreateRequest
from app.schemas.run_request import RunRequestCreateRequest
from app.schemas.session import SessionCreateRequest


class SessionApplicationService:
    def __init__(
        self,
        session_repository: SessionRepository,
        message_repository: MessageRepository,
        run_request_repository: RunRequestRepository) -> None:
        self.session_repository = session_repository
        self.message_repository = message_repository
        self.run_request_repository = run_request_repository

    def create_session(self, payload: SessionCreateRequest) -> SessionModel:
        return self.session_repository.create(
            app_id=payload.app_id,
            user_id=payload.user_id,
            channel_type=payload.channel_type,
            title=payload.title)

    def list_sessions(self, app_id: str | None = None) -> list[SessionModel]:
        return self.session_repository.list_by_scope(app_id=app_id)

    def create_message(self, payload: MessageCreateRequest) -> Message:
        return self.message_repository.create(
            session_id=payload.session_id,
            turn_id=payload.turn_id,
            role=payload.role,
            content_type=payload.content_type,
            content_text=payload.content_text,
            content_json=payload.content_json)

    def list_messages(self, session_id: str) -> list[Message]:
        return self.message_repository.list_by_session(session_id=session_id)

    def create_run_request(self, payload: RunRequestCreateRequest) -> RunRequest:
        return self.run_request_repository.create(
            session_id=payload.session_id,
            app_config_id=payload.app_config_id,
            workflow_config_id=payload.workflow_config_id,
            trigger_type=payload.trigger_type,
            request_payload_json=payload.request_payload_json,
            request_status=payload.request_status)

    def list_run_requests(self, session_id: str) -> list[RunRequest]:
        return self.run_request_repository.list_by_session(session_id=session_id)