counterweight
280c1e5687
- Add get_available_slots() and _expand_availability_to_slots() to ExchangeService - Update routes/exchange.py to use ExchangeService.get_available_slots() - Remove all business logic from get_available_slots endpoint - Add AvailabilityRepository to ExchangeService dependencies - Add Availability and BookableSlot imports to ExchangeService - Fix import path for validate_date_in_range (use date_validation module) - Remove unused user_repo variable and import from routes/invites.py - Fix mypy error in ValidationError by adding proper type annotation
116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
"""Invite routes for public check, user invites, and admin management."""
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from auth import require_permission
|
|
from database import get_db
|
|
from mappers import InviteMapper
|
|
from models import Permission, User
|
|
from schemas import (
|
|
AdminUserResponse,
|
|
InviteCheckResponse,
|
|
InviteCreate,
|
|
InviteResponse,
|
|
PaginatedInviteRecords,
|
|
UserInviteResponse,
|
|
)
|
|
from services.invite import InviteService
|
|
|
|
router = APIRouter(prefix="/api/invites", tags=["invites"])
|
|
admin_router = APIRouter(prefix="/api/admin", tags=["admin"])
|
|
|
|
|
|
@router.get("/{identifier}/check", response_model=InviteCheckResponse)
|
|
async def check_invite(
|
|
identifier: str,
|
|
db: AsyncSession = Depends(get_db),
|
|
) -> InviteCheckResponse:
|
|
"""Check if an invite is valid and can be used for signup."""
|
|
service = InviteService(db)
|
|
return await service.check_invite_validity(identifier)
|
|
|
|
|
|
@router.get("", response_model=list[UserInviteResponse])
|
|
async def get_my_invites(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(require_permission(Permission.VIEW_OWN_INVITES)),
|
|
) -> list[UserInviteResponse]:
|
|
"""Get all invites owned by the current user."""
|
|
service = InviteService(db)
|
|
invites = await service.get_user_invites(current_user.id)
|
|
|
|
return [
|
|
UserInviteResponse(
|
|
id=invite.id,
|
|
identifier=invite.identifier,
|
|
status=invite.status.value,
|
|
used_by_email=invite.used_by.email if invite.used_by else None,
|
|
created_at=invite.created_at,
|
|
spent_at=invite.spent_at,
|
|
)
|
|
for invite in invites
|
|
]
|
|
|
|
|
|
@admin_router.get("/users", response_model=list[AdminUserResponse])
|
|
async def list_users_for_admin(
|
|
db: AsyncSession = Depends(get_db),
|
|
_current_user: User = Depends(require_permission(Permission.MANAGE_INVITES)),
|
|
) -> list[AdminUserResponse]:
|
|
"""List all users for admin dropdowns (invite creation, etc.)."""
|
|
# Note: UserRepository doesn't have list_all yet
|
|
# For now, keeping direct query for this specific use case
|
|
result = await db.execute(select(User.id, User.email).order_by(User.email))
|
|
users = result.all()
|
|
return [AdminUserResponse(id=u.id, email=u.email) for u in users]
|
|
|
|
|
|
@admin_router.post("/invites", response_model=InviteResponse)
|
|
async def create_invite(
|
|
data: InviteCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
_current_user: User = Depends(require_permission(Permission.MANAGE_INVITES)),
|
|
) -> InviteResponse:
|
|
"""Create a new invite for a specified godfather user."""
|
|
service = InviteService(db)
|
|
invite = await service.create_invite(data.godfather_id)
|
|
return InviteMapper.to_response(invite)
|
|
|
|
|
|
@admin_router.get("/invites", response_model=PaginatedInviteRecords)
|
|
async def list_all_invites(
|
|
page: int = Query(1, ge=1),
|
|
per_page: int = Query(10, ge=1, le=100),
|
|
status_filter: str | None = Query(
|
|
None, alias="status", description="Filter by status: ready, spent, revoked"
|
|
),
|
|
godfather_id: int | None = Query(None, description="Filter by godfather user ID"),
|
|
db: AsyncSession = Depends(get_db),
|
|
_current_user: User = Depends(require_permission(Permission.MANAGE_INVITES)),
|
|
) -> PaginatedInviteRecords:
|
|
"""List all invites with optional filtering and pagination."""
|
|
service = InviteService(db)
|
|
return await service.list_invites(
|
|
page=page,
|
|
per_page=per_page,
|
|
status_filter=status_filter,
|
|
godfather_id=godfather_id,
|
|
)
|
|
|
|
|
|
@admin_router.post("/invites/{invite_id}/revoke", response_model=InviteResponse)
|
|
async def revoke_invite(
|
|
invite_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
_current_user: User = Depends(require_permission(Permission.MANAGE_INVITES)),
|
|
) -> InviteResponse:
|
|
"""Revoke an invite. Only READY invites can be revoked."""
|
|
service = InviteService(db)
|
|
invite = await service.revoke_invite(invite_id)
|
|
return InviteMapper.to_response(invite)
|
|
|
|
|
|
# All routers from this module for easy registration
|
|
routers = [router, admin_router]
|