Xiaoxia AI
530b31dc10
Deploy / Build Production Runtime Images (push) Successful in 9m19s
Deploy / Deploy Production (push) Failing after 18s
Deploy / Production Browser E2E (push) Has been skipped
CI/CD Pipeline / Validate Code Quality And Tests (push) Has been cancelled
Deploy / Deploy Staging (push) Has been skipped
43 lines
1.4 KiB
Python
43 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from packages.domain import GenerationTask
|
|
from packages.ports.generation_task_repository import GenerationTaskRepository
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class CreateGenerationTaskCommand:
|
|
workspace_id: str
|
|
project_id: str
|
|
asset_library_id: str
|
|
strategy_id: str = ""
|
|
voice_library_id: str = ""
|
|
edit_plan_id: str = ""
|
|
created_by_user_id: str = ""
|
|
|
|
|
|
class CreateGenerationTaskUseCase:
|
|
def __init__(self, generation_task_repository: GenerationTaskRepository):
|
|
self.generation_task_repository = generation_task_repository
|
|
|
|
def execute(self, command: CreateGenerationTaskCommand) -> GenerationTask:
|
|
task = GenerationTask.create(
|
|
workspace_id=command.workspace_id,
|
|
project_id=command.project_id,
|
|
asset_library_id=command.asset_library_id,
|
|
strategy_id=command.strategy_id,
|
|
voice_library_id=command.voice_library_id,
|
|
edit_plan_id=command.edit_plan_id,
|
|
created_by_user_id=command.created_by_user_id,
|
|
)
|
|
return self.generation_task_repository.create(task)
|
|
|
|
|
|
class GetGenerationTaskUseCase:
|
|
def __init__(self, generation_task_repository: GenerationTaskRepository):
|
|
self.generation_task_repository = generation_task_repository
|
|
|
|
def execute(self, task_id: str) -> GenerationTask | None:
|
|
return self.generation_task_repository.get(task_id)
|