vpsx-fast/app/utils/__init__.py

from typing import Annotated

from fastapi import Path, HTTPException
from sqlalchemy import select


def make_primary_annotation(name=str):
    return Annotated[str, Path(description=f'{name}, either id (int) or code')]


def get_single_record(session, cls, name: str, code: str):
    result = session.get(cls, code)
    if result is None:
        result = session.scalar(select(cls).where(cls.code == code))
    if result is None:
        raise HTTPException(
            status_code=404, 
            detail=f"{name} {code!r} not found.")
    return result


def set_item_status(session, current_user, item, status: str):
    item.status_id = status
    item._user__id = current_user.id
    try:
        session.commit()
    except Exception as exc:
        raise HTTPException(status_code=422,
                            detail=[dict(msg=', '.join(exc.args),
                                         type="Database Error")])
    session.refresh(item)
    return item


def update_item(session, current_user, item, data):
    for k, v in data.model_dump(exclude_unset=True).items():
        if k == 'code':
            r = session.scalar(select(item.__class__).where(item.__class__.code == v))
            if r and r != item:
                raise HTTPException(status_code=422,
                                    detail=[dict(msg=f"{item.__class__.__name__} {v!r} already exists",
                                                 type="Integrity Error")])
        setattr(item, k, v)
    item._user__id = current_user.id
    try:
        session.commit()
    except Exception as exc:
        raise HTTPException(status_code=422,
                            detail=[dict(msg=', '.join(exc.args),
                                         type="Database Error")])
    session.refresh(item)
    return item


def create_item(session, cls, name, current_user, data):
    item = cls(**data.model_dump())
    if session.scalar(select(cls).where(cls.code == item.code)):
        raise HTTPException(status_code=422,
                            detail=[dict(msg=f"{name} {item.code!r} already exists",
                                         type="Integrity Error")])
    item._user__id = current_user.id
    session.add(item)
    try:
        session.commit()
    except Exception as exc:
        raise HTTPException(status_code=422,
                            detail=[dict(msg=', '.join(exc.args),
                                         type="Database Error")])
    session.refresh(item)
    return item