| from uuid import UUID |
|
|
| from fastapi import APIRouter, Depends, HTTPException |
| from sqlalchemy.orm import Session |
|
|
| from app.core.dependencies import get_current_user |
| from app.database.session import get_db |
| from app.models.user import User |
| from app.models.workspace import WorkspaceStatus |
| from app.schemas.workspace import ( |
| WorkspaceCreate, |
| WorkspaceResponse, |
| ) |
| from app.services.workspace_service import WorkspaceService |
|
|
|
|
| router = APIRouter( |
| prefix="/workspaces", |
| tags=["Workspaces"], |
| ) |
|
|
|
|
| @router.post( |
| "", |
| response_model=WorkspaceResponse, |
| ) |
| def create_workspace( |
| workspace: WorkspaceCreate, |
| current_user: User = Depends(get_current_user), |
| db: Session = Depends(get_db), |
| ): |
| service = WorkspaceService(db) |
|
|
| return service.create_workspace( |
| workspace, |
| current_user.id, |
| ) |
|
|
|
|
| @router.get( |
| "", |
| response_model=list[WorkspaceResponse], |
| ) |
| def list_workspaces( |
| current_user: User = Depends(get_current_user), |
| db: Session = Depends(get_db), |
| ): |
| return WorkspaceService(db).list_workspaces( |
| current_user.id |
| ) |
|
|
|
|
| @router.delete("/{workspace_id}", status_code=204) |
| def delete_workspace( |
| workspace_id: UUID, |
| current_user: User = Depends(get_current_user), |
| db: Session = Depends(get_db), |
| ): |
| """Soft-delete a workspace. It will no longer appear in listings.""" |
| service = WorkspaceService(db) |
| workspace = service.get_workspace(workspace_id) |
|
|
| if workspace is None: |
| raise HTTPException(status_code=404, detail="Workspace not found.") |
| if workspace.created_by != current_user.id: |
| raise HTTPException(status_code=403, detail="You do not have access to this workspace.") |
| if workspace.status == WorkspaceStatus.DELETED: |
| raise HTTPException(status_code=404, detail="Workspace not found.") |
|
|
| from app.repositories.workspace_repository import WorkspaceRepository |
| repo = WorkspaceRepository(db) |
| repo.soft_delete(workspace) |
|
|
|
|
| @router.post("/{workspace_id}/archive", response_model=WorkspaceResponse) |
| def archive_workspace( |
| workspace_id: UUID, |
| current_user: User = Depends(get_current_user), |
| db: Session = Depends(get_db), |
| ): |
| """Archive a workspace. It will no longer appear in active listings.""" |
| service = WorkspaceService(db) |
| workspace = service.get_workspace(workspace_id) |
|
|
| if workspace is None: |
| raise HTTPException(status_code=404, detail="Workspace not found.") |
| if workspace.created_by != current_user.id: |
| raise HTTPException(status_code=403, detail="You do not have access to this workspace.") |
|
|
| from app.repositories.workspace_repository import WorkspaceRepository |
| repo = WorkspaceRepository(db) |
| return repo.archive(workspace) |
|
|