xiaoxia
5dbde7e176
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Waiting to run
Tests / test (pull_request) Waiting to run
Tests / lint (pull_request) Waiting to run
CI/CD Pipeline / Validate Code Quality And Tests (push) Has been cancelled
Deploy / Deploy Staging (push) Has been cancelled
Deploy / Build Production Runtime Images (push) Has been cancelled
Deploy / Deploy Production (push) Has been cancelled
Deploy / Production Browser E2E (push) Has been cancelled
78 lines
2.9 KiB
Python
78 lines
2.9 KiB
Python
from typing import Any
|
|
|
|
from app.auth import AuthenticatedUser, get_current_user
|
|
from app.dependencies import get_project_repository, get_workspace_member_repository
|
|
from app.schemas.project import (
|
|
CreateProjectRequest,
|
|
ListProjectsResponse,
|
|
ProjectResponse,
|
|
)
|
|
from app.api.routes.permissions import require_workspace_member
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
|
|
from packages.application import (
|
|
CreateProjectCommand,
|
|
CreateProjectUseCase,
|
|
GetProjectUseCase,
|
|
ListProjectsUseCase,
|
|
)
|
|
from packages.ports.workspace_member_repository import WorkspaceMemberRepository
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def _to_project_response(item) -> ProjectResponse:
|
|
return ProjectResponse(
|
|
id=item.id,
|
|
workspace_id=item.workspace_id,
|
|
name=item.name,
|
|
description=item.description,
|
|
)
|
|
|
|
|
|
@router.get("/{project_id}", response_model=ProjectResponse)
|
|
def get_project(
|
|
project_id: str,
|
|
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
|
project_repository: Any = Depends(get_project_repository),
|
|
workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
|
|
) -> ProjectResponse:
|
|
use_case = GetProjectUseCase(project_repository)
|
|
project = use_case.execute(project_id)
|
|
if project is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
|
|
require_workspace_member(project.workspace_id, authenticated_user, workspace_member_repository)
|
|
return _to_project_response(project)
|
|
|
|
|
|
@router.get("", response_model=ListProjectsResponse)
|
|
def list_projects(
|
|
workspace_id: str,
|
|
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
|
project_repository: Any = Depends(get_project_repository),
|
|
workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
|
|
) -> ListProjectsResponse:
|
|
require_workspace_member(workspace_id, authenticated_user, workspace_member_repository)
|
|
use_case = ListProjectsUseCase(project_repository)
|
|
projects = use_case.execute(workspace_id)
|
|
return ListProjectsResponse(items=[_to_project_response(item) for item in projects])
|
|
|
|
|
|
@router.post("", response_model=ProjectResponse)
|
|
def create_project(
|
|
request: CreateProjectRequest,
|
|
authenticated_user: AuthenticatedUser = Depends(get_current_user),
|
|
project_repository: Any = Depends(get_project_repository),
|
|
workspace_member_repository: WorkspaceMemberRepository = Depends(get_workspace_member_repository),
|
|
) -> ProjectResponse:
|
|
require_workspace_member(request.workspace_id, authenticated_user, workspace_member_repository)
|
|
use_case = CreateProjectUseCase(project_repository)
|
|
project = use_case.execute(
|
|
CreateProjectCommand(
|
|
workspace_id=request.workspace_id,
|
|
name=request.name,
|
|
description=request.description,
|
|
)
|
|
)
|
|
return _to_project_response(project)
|