Xiaoxia AI
b5a62ee9a3
- domain: User, Workspace, Project, AssetLibrary, Asset, IngestJob - ports: repository interfaces - application: use cases - adapters: in-memory - API: FastAPI 5 routes - worker: Celery ingest_asset - tests: 4 passing
37 lines
1.1 KiB
Python
37 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
|
|
from packages.domain import Project
|
|
from packages.ports import ProjectRepository
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class CreateProjectCommand:
|
|
workspace_id: str
|
|
name: str
|
|
description: str = ""
|
|
|
|
|
|
class ListProjectsUseCase:
|
|
def __init__(self, project_repository: ProjectRepository):
|
|
self.project_repository = project_repository
|
|
|
|
def execute(self, workspace_id: str) -> list[Project]:
|
|
if not workspace_id.strip():
|
|
raise ValueError("workspace_id 不能为空")
|
|
return self.project_repository.list_by_workspace(workspace_id.strip())
|
|
|
|
|
|
class CreateProjectUseCase:
|
|
def __init__(self, project_repository: ProjectRepository):
|
|
self.project_repository = project_repository
|
|
|
|
def execute(self, command: CreateProjectCommand) -> Project:
|
|
project = Project.create(
|
|
workspace_id=command.workspace_id,
|
|
name=command.name,
|
|
description=command.description,
|
|
)
|
|
return self.project_repository.create(project)
|