All checks were successful
CI/CD Pipeline / VM Test - backend-integration (push) Successful in 12s
CI/CD Pipeline / VM Test - full-stack (push) Successful in 8s
CI/CD Pipeline / VM Test - performance (push) Successful in 8s
CI/CD Pipeline / VM Test - security (push) Successful in 8s
CI/CD Pipeline / Backend Linting (push) Successful in 4s
CI/CD Pipeline / Frontend Linting (push) Successful in 30s
CI/CD Pipeline / Nix Flake Check (push) Successful in 43s
CI/CD Pipeline / VM Test - backend-integration (pull_request) Successful in 4s
CI/CD Pipeline / VM Test - full-stack (pull_request) Successful in 2s
CI/CD Pipeline / VM Test - performance (pull_request) Successful in 2s
CI/CD Pipeline / VM Test - security (pull_request) Successful in 2s
CI/CD Pipeline / Backend Linting (pull_request) Successful in 2s
CI/CD Pipeline / Frontend Linting (pull_request) Successful in 17s
CI/CD Pipeline / Nix Flake Check (pull_request) Successful in 38s
CI/CD Pipeline / CI Summary (push) Successful in 1s
CI/CD Pipeline / CI Summary (pull_request) Successful in 1s
217 lines
5.8 KiB
Python
217 lines
5.8 KiB
Python
"""Group management API endpoints."""
|
|
|
|
from typing import Annotated
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.boards.repository import BoardRepository
|
|
from app.boards.schemas import GroupCreate, GroupResponse, GroupUpdate
|
|
from app.core.deps import get_current_user, get_db_sync
|
|
from app.database.models.user import User
|
|
|
|
router = APIRouter(prefix="/boards/{board_id}/groups", tags=["groups"])
|
|
|
|
|
|
@router.post("", response_model=GroupResponse, status_code=status.HTTP_201_CREATED)
|
|
def create_group(
|
|
board_id: UUID,
|
|
group_data: GroupCreate,
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
db: Annotated[Session, Depends(get_db_sync)],
|
|
):
|
|
"""
|
|
Create a new group on a board.
|
|
|
|
Assigns the specified images to the group.
|
|
"""
|
|
repo = BoardRepository(db)
|
|
|
|
# Verify board ownership
|
|
board = repo.get_board_by_id(board_id, current_user.id)
|
|
if not board:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found",
|
|
)
|
|
|
|
# Create group
|
|
group = repo.create_group(
|
|
board_id=board_id,
|
|
name=group_data.name,
|
|
color=group_data.color,
|
|
annotation=group_data.annotation,
|
|
image_ids=group_data.image_ids,
|
|
)
|
|
|
|
# Calculate member count
|
|
response = GroupResponse.model_validate(group)
|
|
response.member_count = len(group_data.image_ids)
|
|
|
|
return response
|
|
|
|
|
|
@router.get("", response_model=list[GroupResponse])
|
|
def list_groups(
|
|
board_id: UUID,
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
db: Annotated[Session, Depends(get_db_sync)],
|
|
):
|
|
"""
|
|
List all groups on a board.
|
|
|
|
Returns groups with member counts.
|
|
"""
|
|
repo = BoardRepository(db)
|
|
|
|
# Verify board ownership
|
|
board = repo.get_board_by_id(board_id, current_user.id)
|
|
if not board:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found",
|
|
)
|
|
|
|
# Get groups
|
|
groups = repo.get_board_groups(board_id)
|
|
|
|
# Convert to response with member counts
|
|
from sqlalchemy import func, select
|
|
|
|
from app.database.models.board_image import BoardImage
|
|
|
|
responses = []
|
|
for group in groups:
|
|
# Count members
|
|
count_stmt = select(func.count(BoardImage.id)).where(BoardImage.group_id == group.id)
|
|
member_count = db.execute(count_stmt).scalar_one()
|
|
|
|
response = GroupResponse.model_validate(group)
|
|
response.member_count = member_count
|
|
responses.append(response)
|
|
|
|
return responses
|
|
|
|
|
|
@router.get("/{group_id}", response_model=GroupResponse)
|
|
def get_group(
|
|
board_id: UUID,
|
|
group_id: UUID,
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
db: Annotated[Session, Depends(get_db_sync)],
|
|
):
|
|
"""
|
|
Get group details by ID.
|
|
"""
|
|
repo = BoardRepository(db)
|
|
|
|
# Verify board ownership
|
|
board = repo.get_board_by_id(board_id, current_user.id)
|
|
if not board:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found",
|
|
)
|
|
|
|
# Get group
|
|
group = repo.get_group_by_id(group_id, board_id)
|
|
if not group:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Group not found",
|
|
)
|
|
|
|
# Count members
|
|
from sqlalchemy import func, select
|
|
|
|
from app.database.models.board_image import BoardImage
|
|
|
|
count_stmt = select(func.count(BoardImage.id)).where(BoardImage.group_id == group.id)
|
|
member_count = db.execute(count_stmt).scalar_one()
|
|
|
|
response = GroupResponse.model_validate(group)
|
|
response.member_count = member_count
|
|
|
|
return response
|
|
|
|
|
|
@router.patch("/{group_id}", response_model=GroupResponse)
|
|
def update_group(
|
|
board_id: UUID,
|
|
group_id: UUID,
|
|
group_data: GroupUpdate,
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
db: Annotated[Session, Depends(get_db_sync)],
|
|
):
|
|
"""
|
|
Update group metadata (name, color, annotation).
|
|
"""
|
|
repo = BoardRepository(db)
|
|
|
|
# Verify board ownership
|
|
board = repo.get_board_by_id(board_id, current_user.id)
|
|
if not board:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found",
|
|
)
|
|
|
|
# Update group
|
|
group = repo.update_group(
|
|
group_id=group_id,
|
|
board_id=board_id,
|
|
name=group_data.name,
|
|
color=group_data.color,
|
|
annotation=group_data.annotation,
|
|
)
|
|
|
|
if not group:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Group not found",
|
|
)
|
|
|
|
# Count members
|
|
from sqlalchemy import func, select
|
|
|
|
from app.database.models.board_image import BoardImage
|
|
|
|
count_stmt = select(func.count(BoardImage.id)).where(BoardImage.group_id == group.id)
|
|
member_count = db.execute(count_stmt).scalar_one()
|
|
|
|
response = GroupResponse.model_validate(group)
|
|
response.member_count = member_count
|
|
|
|
return response
|
|
|
|
|
|
@router.delete("/{group_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_group(
|
|
board_id: UUID,
|
|
group_id: UUID,
|
|
current_user: Annotated[User, Depends(get_current_user)],
|
|
db: Annotated[Session, Depends(get_db_sync)],
|
|
):
|
|
"""
|
|
Delete a group (ungroups all images).
|
|
"""
|
|
repo = BoardRepository(db)
|
|
|
|
# Verify board ownership
|
|
board = repo.get_board_by_id(board_id, current_user.id)
|
|
if not board:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Board not found",
|
|
)
|
|
|
|
# Delete group
|
|
success = repo.delete_group(group_id, board_id)
|
|
|
|
if not success:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Group not found",
|
|
)
|