228 lines
9.2 KiB
Python
228 lines
9.2 KiB
Python
# app/api/v1/endpoints/groups.py
|
|
import logging
|
|
from typing import List
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.database import get_db
|
|
from app.auth import current_active_user
|
|
from app.models import User as UserModel, UserRoleEnum # Import model and enum
|
|
from app.schemas.group import GroupCreate, GroupPublic
|
|
from app.schemas.invite import InviteCodePublic
|
|
from app.schemas.message import Message # For simple responses
|
|
from app.schemas.list import ListPublic
|
|
from app.crud import group as crud_group
|
|
from app.crud import invite as crud_invite
|
|
from app.crud import list as crud_list
|
|
from app.core.exceptions import (
|
|
GroupNotFoundError,
|
|
GroupPermissionError,
|
|
GroupMembershipError,
|
|
GroupOperationError,
|
|
GroupValidationError,
|
|
InviteCreationError
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
router = APIRouter()
|
|
|
|
@router.post(
|
|
"", # Route relative to prefix "/groups"
|
|
response_model=GroupPublic,
|
|
status_code=status.HTTP_201_CREATED,
|
|
summary="Create New Group",
|
|
tags=["Groups"]
|
|
)
|
|
async def create_group(
|
|
group_in: GroupCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: UserModel = Depends(current_active_user),
|
|
):
|
|
"""Creates a new group, adding the creator as the owner."""
|
|
logger.info(f"User {current_user.email} creating group: {group_in.name}")
|
|
created_group = await crud_group.create_group(db=db, group_in=group_in, creator_id=current_user.id)
|
|
# Load members explicitly if needed for the response (optional here)
|
|
# created_group = await crud_group.get_group_by_id(db, created_group.id)
|
|
return created_group
|
|
|
|
|
|
@router.get(
|
|
"", # Route relative to prefix "/groups"
|
|
response_model=List[GroupPublic],
|
|
summary="List User's Groups",
|
|
tags=["Groups"]
|
|
)
|
|
async def read_user_groups(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: UserModel = Depends(current_active_user),
|
|
):
|
|
"""Retrieves all groups the current user is a member of."""
|
|
logger.info(f"Fetching groups for user: {current_user.email}")
|
|
groups = await crud_group.get_user_groups(db=db, user_id=current_user.id)
|
|
return groups
|
|
|
|
|
|
@router.get(
|
|
"/{group_id}",
|
|
response_model=GroupPublic,
|
|
summary="Get Group Details",
|
|
tags=["Groups"]
|
|
)
|
|
async def read_group(
|
|
group_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: UserModel = Depends(current_active_user),
|
|
):
|
|
"""Retrieves details for a specific group, including members, if the user is part of it."""
|
|
logger.info(f"User {current_user.email} requesting details for group ID: {group_id}")
|
|
# Check if user is a member first
|
|
is_member = await crud_group.is_user_member(db=db, group_id=group_id, user_id=current_user.id)
|
|
if not is_member:
|
|
logger.warning(f"Access denied: User {current_user.email} not member of group {group_id}")
|
|
raise GroupMembershipError(group_id, "view group details")
|
|
|
|
group = await crud_group.get_group_by_id(db=db, group_id=group_id)
|
|
if not group:
|
|
logger.error(f"Group {group_id} requested by member {current_user.email} not found (data inconsistency?)")
|
|
raise GroupNotFoundError(group_id)
|
|
|
|
return group
|
|
|
|
|
|
@router.post(
|
|
"/{group_id}/invites",
|
|
response_model=InviteCodePublic,
|
|
summary="Create Group Invite",
|
|
tags=["Groups", "Invites"]
|
|
)
|
|
async def create_group_invite(
|
|
group_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: UserModel = Depends(current_active_user),
|
|
):
|
|
"""Generates a new invite code for the group. Requires owner/admin role (MVP: owner only)."""
|
|
logger.info(f"User {current_user.email} attempting to create invite for group {group_id}")
|
|
user_role = await crud_group.get_user_role_in_group(db, group_id=group_id, user_id=current_user.id)
|
|
|
|
# --- Permission Check (MVP: Owner only) ---
|
|
if user_role != UserRoleEnum.owner:
|
|
logger.warning(f"Permission denied: User {current_user.email} (role: {user_role}) cannot create invite for group {group_id}")
|
|
raise GroupPermissionError(group_id, "create invites")
|
|
|
|
# Check if group exists (implicitly done by role check, but good practice)
|
|
group = await crud_group.get_group_by_id(db, group_id)
|
|
if not group:
|
|
raise GroupNotFoundError(group_id)
|
|
|
|
invite = await crud_invite.create_invite(db=db, group_id=group_id, creator_id=current_user.id)
|
|
if not invite:
|
|
logger.error(f"Failed to generate unique invite code for group {group_id}")
|
|
raise InviteCreationError(group_id)
|
|
|
|
logger.info(f"User {current_user.email} created invite code for group {group_id}")
|
|
return invite
|
|
|
|
@router.delete(
|
|
"/{group_id}/leave",
|
|
response_model=Message,
|
|
summary="Leave Group",
|
|
tags=["Groups"]
|
|
)
|
|
async def leave_group(
|
|
group_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: UserModel = Depends(current_active_user),
|
|
):
|
|
"""Removes the current user from the specified group."""
|
|
logger.info(f"User {current_user.email} attempting to leave group {group_id}")
|
|
user_role = await crud_group.get_user_role_in_group(db, group_id=group_id, user_id=current_user.id)
|
|
|
|
if user_role is None:
|
|
raise GroupMembershipError(group_id, "leave (you are not a member)")
|
|
|
|
# --- MVP: Prevent owner leaving if they are the last member/owner ---
|
|
if user_role == UserRoleEnum.owner:
|
|
member_count = await crud_group.get_group_member_count(db, group_id)
|
|
# More robust check: count owners. For now, just check member count.
|
|
if member_count <= 1:
|
|
logger.warning(f"Owner {current_user.email} attempted to leave group {group_id} as last member.")
|
|
raise GroupValidationError("Owner cannot leave the group as the last member. Delete the group or transfer ownership.")
|
|
|
|
# Proceed with removal
|
|
deleted = await crud_group.remove_user_from_group(db, group_id=group_id, user_id=current_user.id)
|
|
|
|
if not deleted:
|
|
# Should not happen if role check passed, but handle defensively
|
|
logger.error(f"Failed to remove user {current_user.email} from group {group_id} despite being a member.")
|
|
raise GroupOperationError("Failed to leave group")
|
|
|
|
logger.info(f"User {current_user.email} successfully left group {group_id}")
|
|
return Message(detail="Successfully left the group")
|
|
|
|
# --- Optional: Remove Member Endpoint ---
|
|
@router.delete(
|
|
"/{group_id}/members/{user_id_to_remove}",
|
|
response_model=Message,
|
|
summary="Remove Member From Group (Owner Only)",
|
|
tags=["Groups"]
|
|
)
|
|
async def remove_group_member(
|
|
group_id: int,
|
|
user_id_to_remove: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: UserModel = Depends(current_active_user),
|
|
):
|
|
"""Removes a specified user from the group. Requires current user to be owner."""
|
|
logger.info(f"Owner {current_user.email} attempting to remove user {user_id_to_remove} from group {group_id}")
|
|
owner_role = await crud_group.get_user_role_in_group(db, group_id=group_id, user_id=current_user.id)
|
|
|
|
# --- Permission Check ---
|
|
if owner_role != UserRoleEnum.owner:
|
|
logger.warning(f"Permission denied: User {current_user.email} (role: {owner_role}) cannot remove members from group {group_id}")
|
|
raise GroupPermissionError(group_id, "remove members")
|
|
|
|
# Prevent owner removing themselves via this endpoint
|
|
if current_user.id == user_id_to_remove:
|
|
raise GroupValidationError("Owner cannot remove themselves using this endpoint. Use 'Leave Group' instead.")
|
|
|
|
# Check if target user is actually in the group
|
|
target_role = await crud_group.get_user_role_in_group(db, group_id=group_id, user_id=user_id_to_remove)
|
|
if target_role is None:
|
|
raise GroupMembershipError(group_id, "remove this user (they are not a member)")
|
|
|
|
# Proceed with removal
|
|
deleted = await crud_group.remove_user_from_group(db, group_id=group_id, user_id=user_id_to_remove)
|
|
|
|
if not deleted:
|
|
logger.error(f"Owner {current_user.email} failed to remove user {user_id_to_remove} from group {group_id}.")
|
|
raise GroupOperationError("Failed to remove member")
|
|
|
|
logger.info(f"Owner {current_user.email} successfully removed user {user_id_to_remove} from group {group_id}")
|
|
return Message(detail="Successfully removed member from the group")
|
|
|
|
@router.get(
|
|
"/{group_id}/lists",
|
|
response_model=List[ListPublic],
|
|
summary="Get Group Lists",
|
|
tags=["Groups", "Lists"]
|
|
)
|
|
async def read_group_lists(
|
|
group_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: UserModel = Depends(current_active_user),
|
|
):
|
|
"""Retrieves all lists belonging to a specific group, if the user is a member."""
|
|
logger.info(f"User {current_user.email} requesting lists for group ID: {group_id}")
|
|
|
|
# Check if user is a member first
|
|
is_member = await crud_group.is_user_member(db=db, group_id=group_id, user_id=current_user.id)
|
|
if not is_member:
|
|
logger.warning(f"Access denied: User {current_user.email} not member of group {group_id}")
|
|
raise GroupMembershipError(group_id, "view group lists")
|
|
|
|
# Get all lists for the user and filter by group_id
|
|
lists = await crud_list.get_lists_for_user(db=db, user_id=current_user.id)
|
|
group_lists = [list for list in lists if list.group_id == group_id]
|
|
|
|
return group_lists |