chore(quality): apply ruff autofixes and formatting (8 fixes, 18 files reformatted)

2026-06-10 21:35:12 +00:00
parent 415abcd74e
commit fac6f100bb
22 changed files with 1388 additions and 393 deletions
@@ -1,54 +1,215 @@
-"""Workspace routes."""
-from fastapi import APIRouter, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException, status
 from sqlalchemy.orm import Session
 from typing import List

 from app.database import get_db
-from app.models.workspace import Workspace
-from app.schemas.workspace import WorkspaceCreate, WorkspaceUpdate, WorkspaceResponse
+from app.models.user import User
+from app.models.workspace import Workspace, WorkspaceMember
+from app.schemas.workspace import (
+    WorkspaceCreate,
+    WorkspaceRead,
+    WorkspaceUpdate,
+    WorkspaceMemberCreate,
+    WorkspaceMemberRead,
+    WorkspaceMemberUpdate,
+)
+from app.routers.auth import get_current_user

-router = APIRouter(prefix="/workspaces", tags=["workspaces"])
+router = APIRouter()

-@router.get("", response_model=List[WorkspaceResponse])
-def list_workspaces(db: Session = Depends(get_db)):
-    """List all workspaces."""
-    return db.query(Workspace).all()

-@router.post("", response_model=WorkspaceResponse)
-def create_workspace(workspace: WorkspaceCreate, db: Session = Depends(get_db)):
-    """Create new workspace."""
-    db_workspace = Workspace(**workspace.model_dump())
-    db.add(db_workspace)
-    db.commit()
-    db.refresh(db_workspace)
-    return db_workspace
-
-@router.get("/{workspace_id}", response_model=WorkspaceResponse)
-def get_workspace(workspace_id: int, db: Session = Depends(get_db)):
-    """Get workspace by ID."""
+def check_workspace_access(workspace_id: int, db: Session, user: User) -> Workspace:
+    """Check if user has access to workspace."""
    workspace = db.query(Workspace).filter(Workspace.id == workspace_id).first()
    if not workspace:
        raise HTTPException(status_code=404, detail="Workspace not found")
+    # Owner or member has access
+    if workspace.owner_id != user.id:
+        membership = (
+            db.query(WorkspaceMember)
+            .filter(
+                WorkspaceMember.workspace_id == workspace_id,
+                WorkspaceMember.user_id == user.id,
+            )
+            .first()
+        )
+        if not membership:
+            raise HTTPException(status_code=403, detail="Access denied")
    return workspace

-@router.put("/{workspace_id}", response_model=WorkspaceResponse)
-def update_workspace(workspace_id: int, workspace: WorkspaceUpdate, db: Session = Depends(get_db)):
-    """Update workspace."""
-    db_workspace = db.query(Workspace).filter(Workspace.id == workspace_id).first()
-    if not db_workspace:
-        raise HTTPException(status_code=404, detail="Workspace not found")
-    for key, value in workspace.model_dump(exclude_unset=True).items():
-        setattr(db_workspace, key, value)
-    db.commit()
-    db.refresh(db_workspace)
-    return db_workspace

-@router.delete("/{workspace_id}")
-def delete_workspace(workspace_id: int, db: Session = Depends(get_db)):
-    """Delete workspace."""
-    workspace = db.query(Workspace).filter(Workspace.id == workspace_id).first()
-    if not workspace:
-        raise HTTPException(status_code=404, detail="Workspace not found")
+@router.get("/workspaces", response_model=List[WorkspaceRead])
+def list_workspaces(
+    db: Session = Depends(get_db), current_user: User = Depends(get_current_user)
+):
+    """List user's workspaces (owned + member)."""
+    owned = db.query(Workspace).filter(Workspace.owner_id == current_user.id).all()
+    memberships = (
+        db.query(WorkspaceMember)
+        .filter(WorkspaceMember.user_id == current_user.id)
+        .all()
+    )
+    member_ids = [m.workspace_id for m in memberships]
+    member_workspaces = (
+        db.query(Workspace).filter(Workspace.id.in_(member_ids)).all()
+        if member_ids
+        else []
+    )
+    return list(set(owned + member_workspaces))
+
+
+@router.post(
+    "/workspaces", response_model=WorkspaceRead, status_code=status.HTTP_201_CREATED
+)
+def create_workspace(
+    data: WorkspaceCreate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Create a new workspace."""
+    workspace = Workspace(
+        name=data.name,
+        description=data.description,
+        owner_id=current_user.id,
+        is_private=data.is_private,
+    )
+    db.add(workspace)
+    db.commit()
+    db.refresh(workspace)
+    return workspace
+
+
+@router.get("/workspaces/{workspace_id}", response_model=WorkspaceRead)
+def get_workspace(
+    workspace_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Get workspace details."""
+    return check_workspace_access(workspace_id, db, current_user)
+
+
+@router.put("/workspaces/{workspace_id}", response_model=WorkspaceRead)
+def update_workspace(
+    workspace_id: int,
+    data: WorkspaceUpdate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Update workspace (owner only)."""
+    workspace = check_workspace_access(workspace_id, db, current_user)
+    if workspace.owner_id != current_user.id:
+        raise HTTPException(status_code=403, detail="Owner access required")
+    for key, value in data.model_dump(exclude_unset=True).items():
+        setattr(workspace, key, value)
+    db.commit()
+    db.refresh(workspace)
+    return workspace
+
+
+@router.delete("/workspaces/{workspace_id}")
+def delete_workspace(
+    workspace_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Delete workspace (owner only)."""
+    workspace = check_workspace_access(workspace_id, db, current_user)
+    if workspace.owner_id != current_user.id:
+        raise HTTPException(status_code=403, detail="Owner access required")
    db.delete(workspace)
    db.commit()
    return {"message": "Workspace deleted"}
+
+
+@router.get(
+    "/workspaces/{workspace_id}/members", response_model=List[WorkspaceMemberRead]
+)
+def list_members(
+    workspace_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """List workspace members."""
+    check_workspace_access(workspace_id, db, current_user)
+    return (
+        db.query(WorkspaceMember)
+        .filter(WorkspaceMember.workspace_id == workspace_id)
+        .all()
+    )
+
+
+@router.post(
+    "/workspaces/{workspace_id}/members",
+    response_model=WorkspaceMemberRead,
+    status_code=status.HTTP_201_CREATED,
+)
+def add_member(
+    workspace_id: int,
+    data: WorkspaceMemberCreate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Add member to workspace (owner only)."""
+    workspace = check_workspace_access(workspace_id, db, current_user)
+    if workspace.owner_id != current_user.id:
+        raise HTTPException(status_code=403, detail="Owner access required")
+    membership = WorkspaceMember(
+        workspace_id=workspace_id, user_id=data.user_id, role=data.role
+    )
+    db.add(membership)
+    db.commit()
+    db.refresh(membership)
+    return membership
+
+
+@router.put(
+    "/workspaces/{workspace_id}/members/{user_id}", response_model=WorkspaceMemberRead
+)
+def update_member(
+    workspace_id: int,
+    user_id: int,
+    data: WorkspaceMemberUpdate,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Update member role (owner only)."""
+    check_workspace_access(workspace_id, db, current_user)
+    membership = (
+        db.query(WorkspaceMember)
+        .filter(
+            WorkspaceMember.workspace_id == workspace_id,
+            WorkspaceMember.user_id == user_id,
+        )
+        .first()
+    )
+    if not membership:
+        raise HTTPException(status_code=404, detail="Member not found")
+    membership.role = data.role
+    db.commit()
+    db.refresh(membership)
+    return membership
+
+
+@router.delete("/workspaces/{workspace_id}/members/{user_id}")
+def remove_member(
+    workspace_id: int,
+    user_id: int,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_user),
+):
+    """Remove member from workspace (owner only)."""
+    check_workspace_access(workspace_id, db, current_user)
+    membership = (
+        db.query(WorkspaceMember)
+        .filter(
+            WorkspaceMember.workspace_id == workspace_id,
+            WorkspaceMember.user_id == user_id,
+        )
+        .first()
+    )
+    if not membership:
+        raise HTTPException(status_code=404, detail="Member not found")
+    db.delete(membership)
+    db.commit()
+    return {"message": "Member removed"}