Source code for ingot.responses

"""HTTP-404 helpers for row-lookup and row-count verification."""

from __future__ import annotations

from typing import TYPE_CHECKING, Any

from fastapi import HTTPException, status

if TYPE_CHECKING:
    from sqlalchemy.ext.asyncio import AsyncSession




async def get_object_from_query_or_404(
    db: AsyncSession,
    stmt: Any,
    *,
    detail: str = "Not found",
) -> Any:
    """Execute *stmt* and return the first row, or raise HTTP 404.

    Args:
        db: The async database session.
        stmt: A SQLAlchemy selectable statement.
        detail: The error message for the 404 response.

    Returns:
        The first row from the result set.

    Raises:
        HTTPException: With status 404 when no row is found.

    """
    result = await db.execute(stmt)
    row = result.scalars().one_or_none()

    if row is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=detail,
        )

    return row





def assert_rowcount(
    result: Any,
    *,
    expected: int = 1,
    status_code: int = status.HTTP_404_NOT_FOUND,
    detail: str = "Not found",
) -> None:
    """Raise HTTPException when *result* did not affect *expected* rows.

    Args:
        result: SQLAlchemy ``CursorResult`` from an ``execute()`` call.
        expected: Number of rows that must have been affected.
        status_code: HTTP status code for the raised exception.
        detail: The error message for the response.

    Raises:
        HTTPException: When ``result.rowcount`` does not equal *expected*.

    """
    if result.rowcount != expected:
        raise HTTPException(status_code=status_code, detail=detail)