from uuid import UUID from fastapi import HTTPException from sqlalchemy import select from sqlalchemy.orm import Session from sqlalchemy.sql import func 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)