| 12345678910111213141516171819202122232425262728293031323334353637383940 |
- 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)
|
