xiaoxia-saas/packages/ports/generation_task_repository.py

from __future__ import annotations

from typing import Protocol

from packages.domain import GenerationTask


class GenerationTaskRepository(Protocol):
    def create(self, task: GenerationTask) -> GenerationTask: ...

    def get(self, task_id: str) -> GenerationTask | None: ...

    def list_by_project(self, project_id: str) -> list[GenerationTask]: ...

    def list_by_user(self, user_id: str) -> list[GenerationTask]: ...

    def count_by_user(self, user_id: str) -> int: ...

    def count_pending_by_user(self, user_id: str) -> int: ...

    def count_pending_total(self) -> int: ...

    def count_running_by_user(self, user_id: str) -> int: ...

    def count_running_total(self) -> int: ...

    def estimate_avg_duration_seconds(self, limit: int = 20, default_seconds: float = 120.0) -> float: ...

    def list_recent_by_user(self, user_id: str, limit: int = 5) -> list[GenerationTask]: ...

    def list_by_source_edit_plan(self, plan_id: str) -> list[GenerationTask]: ...

    def list_by_user_filtered(
        self,
        user_id: str,
        *,
        status: str | None = None,
        limit: int | None = None,
        offset: int = 0,
    ) -> list[GenerationTask]: ...

    def count_by_user_filtered(
        self,
        user_id: str,
        *,
        status: str | None = None,
    ) -> int: ...

    def list_by_project_filtered(
        self,
        project_id: str,
        *,
        status: str | None = None,
        limit: int | None = None,
        offset: int = 0,
    ) -> list[GenerationTask]: ...

    def count_by_project_filtered(
        self,
        project_id: str,
        *,
        status: str | None = None,
    ) -> int: ...

    def update(self, task: GenerationTask) -> GenerationTask: ...