135 lines
4.9 KiB
Python
Executable File
135 lines
4.9 KiB
Python
Executable File
"""Band membership and invite endpoints."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from rehearsalhub.db.engine import get_session
|
|
from rehearsalhub.db.models import Member
|
|
from rehearsalhub.dependencies import get_current_member
|
|
from rehearsalhub.repositories.band import BandRepository
|
|
from rehearsalhub.schemas.invite import BandInviteRead, BandMemberRead
|
|
from rehearsalhub.services.band import BandService
|
|
|
|
router = APIRouter(tags=["members"])
|
|
|
|
|
|
@router.get("/bands/{band_id}/members", response_model=list[BandMemberRead])
|
|
async def list_members(
|
|
band_id: uuid.UUID,
|
|
session: AsyncSession = Depends(get_session),
|
|
current_member: Member = Depends(get_current_member),
|
|
):
|
|
svc = BandService(session)
|
|
try:
|
|
await svc.assert_membership(band_id, current_member.id)
|
|
except PermissionError:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not a member")
|
|
|
|
band = await svc.get_band_with_members(band_id)
|
|
if band is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Band not found")
|
|
|
|
return [
|
|
BandMemberRead(
|
|
id=bm.member.id,
|
|
display_name=bm.member.display_name,
|
|
email=bm.member.email,
|
|
role=bm.role,
|
|
joined_at=bm.joined_at,
|
|
)
|
|
for bm in band.memberships
|
|
]
|
|
|
|
|
|
@router.post("/bands/{band_id}/invites", response_model=BandInviteRead, status_code=status.HTTP_201_CREATED)
|
|
async def create_invite(
|
|
band_id: uuid.UUID,
|
|
session: AsyncSession = Depends(get_session),
|
|
current_member: Member = Depends(get_current_member),
|
|
):
|
|
svc = BandService(session)
|
|
try:
|
|
await svc.assert_admin(band_id, current_member.id)
|
|
except PermissionError:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin role required")
|
|
|
|
repo = BandRepository(session)
|
|
invite = await repo.create_invite(band_id, current_member.id)
|
|
return BandInviteRead.model_validate(invite)
|
|
|
|
|
|
@router.delete("/bands/{band_id}/members/{member_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
async def remove_member(
|
|
band_id: uuid.UUID,
|
|
member_id: uuid.UUID,
|
|
session: AsyncSession = Depends(get_session),
|
|
current_member: Member = Depends(get_current_member),
|
|
):
|
|
svc = BandService(session)
|
|
try:
|
|
await svc.assert_admin(band_id, current_member.id)
|
|
except PermissionError:
|
|
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Admin role required")
|
|
|
|
if member_id == current_member.id:
|
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot remove yourself")
|
|
|
|
repo = BandRepository(session)
|
|
await repo.remove_member(band_id, member_id)
|
|
|
|
|
|
@router.post("/invites/{token}/accept", response_model=BandMemberRead)
|
|
async def accept_invite(
|
|
token: str,
|
|
session: AsyncSession = Depends(get_session),
|
|
current_member: Member = Depends(get_current_member),
|
|
):
|
|
repo = BandRepository(session)
|
|
invite = await repo.get_invite_by_token(token)
|
|
|
|
if invite is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Invite not found")
|
|
if invite.used_at is not None:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Invite already used")
|
|
if invite.expires_at < datetime.now(timezone.utc):
|
|
raise HTTPException(status_code=status.HTTP_410_GONE, detail="Invite expired")
|
|
|
|
# Idempotent — already a member
|
|
existing_role = await repo.get_member_role(invite.band_id, current_member.id)
|
|
if existing_role is not None:
|
|
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Already a member")
|
|
|
|
bm = await repo.add_member(invite.band_id, current_member.id, role=invite.role)
|
|
|
|
# Mark invite as used
|
|
invite.used_at = datetime.now(timezone.utc)
|
|
invite.used_by = current_member.id
|
|
await session.flush()
|
|
|
|
return BandMemberRead(
|
|
id=current_member.id,
|
|
display_name=current_member.display_name,
|
|
email=current_member.email,
|
|
role=bm.role,
|
|
joined_at=bm.joined_at,
|
|
)
|
|
|
|
|
|
@router.get("/invites/{token}", response_model=BandInviteRead)
|
|
async def get_invite(token: str, session: AsyncSession = Depends(get_session)):
|
|
"""Preview invite info (band name etc.) before accepting — no auth required."""
|
|
from sqlalchemy.orm import selectinload
|
|
from sqlalchemy import select
|
|
from rehearsalhub.db.models import BandInvite
|
|
stmt = select(BandInvite).options(selectinload(BandInvite.band)).where(BandInvite.token == token)
|
|
result = await session.execute(stmt)
|
|
invite = result.scalar_one_or_none()
|
|
if invite is None:
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Invite not found")
|
|
return BandInviteRead.model_validate(invite)
|