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

from uuid import UUID

from fastapi import HTTPException
from sqlalchemy import exists, select
from sqlalchemy.orm import Session
from sqlalchemy.sql import func

from app.models import Project


def get_project_organization_id(session: Session, project_id: UUID) -> str:
    """Fetches only Project.organization_id -- avoids loading the full
    Project row (and its relationships) just to check existence/ownership."""
    org_id = session.execute(
        select(Project.organization_id).where(Project.id == project_id)
    ).scalar_one_or_none()
    if org_id is None:
        raise HTTPException(status_code=404, detail="Project not found")
    return org_id


def assert_project_exists(session: Session, project_id: UUID) -> None:
    if not session.execute(select(exists().where(Project.id == project_id))).scalar():
        raise HTTPException(status_code=404, detail="Project not found")


def get_project_or_404(session: Session, project_id: UUID) -> Project:
    p = session.get(Project, project_id)
    if p is None:
        raise HTTPException(status_code=404, detail="Project not found")
    return p


async def verify_id_list(
    session: Session,
    id_list: list[UUID],
    project_id: UUID,
    ObjectClass,
    error_message: str = "Invalid id list",
) -> None:
    """Verfiy the list of uuids exists and are from the right project

    ---
    Parameters
    - session : sqlachlemy Async session to use to verify slot validity
    - id_list : list of id to check
    - project_id :
    - ObjectClass : the ORM class of the object to check the ids from. Need to have id and project_id property

    ---
    Raise
    HTTPException(400, error_message) - if all the ids doesn't exists or are not associated with the right project
    """
    statement = select(func.count(ObjectClass.id)).where(
        ObjectClass.id.in_(id_list) & (ObjectClass.project_id == project_id)
    )
    results = session.execute(statement)
    if results.scalar() != len(id_list):
        raise HTTPException(status_code=400, detail=error_message)


def update_object_from_payload(obj, payload: dict):
    """Update the ORM model object from a pydantic payload dictionary"""
    for attr_name, value in payload.items():
        setattr(obj, attr_name, value)