bdlg2023-sms-back/app/api/endpoints/sms.py

from typing import Annotated
from uuid import UUID

from fastapi import APIRouter, Depends, HTTPException
from fastapi.params import Query
from sqlalchemy import delete, select
from sqlalchemy.orm import Session

from app.api import deps
from app.api.utils import assert_project_exists_or_404, update_object_from_payload
from app.models import (
    OrgRole,
    Sms,
    User,
)
from app.schemas.requests import SmsCreateRequest, SmsUpdateRequest
from app.schemas.responses import SMSResponse

router = APIRouter(prefix="/project/{project_id}", tags=["project", "sms"])

READ_ROLES = (OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE, OrgRole.RESPO_COMMISSION)
WRITE_ROLES = (OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE, OrgRole.RESPO_COMMISSION)


@router.get("/sms", response_model=list[SMSResponse])
async def list_project_sms(
    project_id: UUID,
    current_user: User = Depends(deps.require_org_role(*READ_ROLES)),
    session: Session = Depends(deps.get_session),
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=20000)] = 20000,
):
    """List sms from project"""
    assert_project_exists_or_404(session, project_id)
    stmt = select(Sms).where(Sms.project_id == project_id)
    stmt = stmt.order_by(Sms.created_at.desc()).offset(skip).limit(limit)

    results = session.execute(stmt)
    return results.scalars().all()


@router.post("/sms", response_model=SMSResponse)
async def create_sms(
    project_id: UUID,
    new_sms: SmsCreateRequest,
    current_user: User = Depends(deps.require_org_role(*WRITE_ROLES)),
    session: Session = Depends(deps.get_session),
):
    """Create a new to the project"""
    assert_project_exists_or_404(session, project_id)
    sms = Sms(project_id=project_id, **new_sms.model_dump())
    session.add(sms)
    session.commit()
    return sms


@router.post("/sms/{sms_id}", response_model=SMSResponse)
async def update_sms(
    project_id: UUID,
    sms_id: UUID,
    new_sms: SmsUpdateRequest,
    current_user: User = Depends(deps.require_org_role(*WRITE_ROLES)),
    session: Session = Depends(deps.get_session),
):
    """Update an sms from the project"""
    sms = session.get(Sms, sms_id)
    if (sms is None) or (sms.project_id != str(project_id)):
        raise HTTPException(status_code=404, detail="Sms not found")
    update_object_from_payload(sms, new_sms.model_dump(exclude_unset=True))
    session.commit()
    session.refresh(sms)
    return sms


@router.delete("/sms/{sms_id}")
async def delete_sms(
    project_id: UUID,
    sms_id: UUID,
    current_user: User = Depends(deps.require_org_role(*WRITE_ROLES)),
    session: Session = Depends(deps.get_session),
):
    """Delete a sms from the project"""
    session.execute(delete(Sms).where((Sms.id == sms_id) & (Sms.project_id == str(project_id))))
    session.commit()