FastAPI integration

FastAPI integration for alchemiq. Requires the [fastapi] extra.

alchemiq.fastapi.crud_router(repository, *, prefix='', tags=None, read_schema=None, create_schema=None, update_schema=None, endpoints=None, pagination='offset', dependencies=None, filter_allow=None, id_type=<class 'int'>)[source]

Build a FastAPI APIRouter with standard CRUD endpoints for an alchemiq model.

repository may be a Repository instance, subclass, or a bare model class. By default all five endpoints are mounted:

  • GET / - paginated list (offset or cursor)

  • GET /{id} - single read (404 if missing)

  • POST / - create, returns 201 with the created resource

  • PATCH /{id} - partial update (only supplied fields are changed)

  • DELETE /{id} - soft or hard delete, returns 204 with no body

Pass endpoints={"list", "read"} to mount a subset. pagination="cursor" switches the list endpoint to keyset pagination (after/before query params) returning a CursorPage envelope instead of the default offset Page envelope.

Pydantic schemas are auto-derived from the model’s field metadata via read_schema(), create_schema(), and update_schema(); pass explicit schemas to override.

E.g.:

from fastapi import Depends, FastAPI
from alchemiq import Repository
from alchemiq.fastapi import crud_router, install_exception_handlers

app = FastAPI()
install_exception_handlers(app)
app.include_router(crud_router(User, prefix="/users"))

# Subset of endpoints with cursor pagination and auth dependency:
app.include_router(
    crud_router(
        Repository(Article),
        prefix="/articles",
        tags=["articles"],
        dependencies=[Depends(require_token)],
        endpoints={"list", "read", "create"},
        pagination="cursor",
    )
)
Parameters:

  • repository (Any) – a Repository instance, subclass, or bare model class.

  • prefix (str) – URL prefix for all routes (e.g. "/items").

  • tags (list[str | Enum] | None) – OpenAPI tags applied to every route.

  • read_schema (type[BaseModel] | None) – override the auto-derived read (response) Pydantic schema.

  • create_schema (type[BaseModel] | None) – override the auto-derived create (request body) schema.

  • update_schema (type[BaseModel] | None) – override the auto-derived partial-update schema.

  • endpoints (set[str] | None) – subset of {"list", "read", "create", "update", "delete"} to mount; defaults to all five.

  • pagination (Literal['offset', 'cursor']) – "offset" (default) or "cursor" for the list endpoint.

  • dependencies (Sequence[Any] | None) – FastAPI dependencies applied to every route.

  • filter_allow (set[str] | None) – allowlist of field names the q query param may filter on.

  • id_type (type) – Python type used to parse the {id} path parameter (default int).

Returns:

a configured APIRouter ready to pass to app.include_router().

Raises:

ConfigError – if endpoints contains an unknown name.

Return type:

APIRouter

See also

install_exception_handlers() - wire PersistenceError -> HTTP.

alchemiq.fastapi.health_router(*, prefix='/health', timeout=5.0, include_liveness=True)[source]

Build an APIRouter with GET /ready and optionally GET /live endpoints.

GET /ready calls check_health() and returns 200 when all components are healthy or 503 when any component is degraded - suitable for a Kubernetes readinessProbe. GET /live always returns 200 with {"status": "alive"} and has no backend dependencies - suitable for a Kubernetes livenessProbe.

E.g.:

from fastapi import FastAPI
from alchemiq.fastapi import health_router

app = FastAPI()
app.include_router(health_router())
# custom prefix:
app.include_router(health_router(prefix="/probe"))
# readiness only - omit the /live route:
app.include_router(health_router(include_liveness=False))
Parameters:

  • prefix (str) – URL prefix for the health routes (default "/health").

  • timeout (float) – seconds passed to check_health() for backend probes (default 5.0).

  • include_liveness (bool) – when False, the GET /live route is omitted.

Returns:

a configured APIRouter ready to pass to app.include_router().

Return type:

APIRouter

See also

check_health() - the underlying health probe implementation.

alchemiq.fastapi.lifespan(dsn, *, create_all=False, **engine_kwargs)[source]

Build a FastAPI lifespan that owns the alchemiq engine lifecycle.

Calls configure() on startup and dispose() on shutdown. Pass the result directly to FastAPI(lifespan=...).

E.g.:

from fastapi import FastAPI
from alchemiq.fastapi import lifespan

app = FastAPI(lifespan=lifespan("postgresql+asyncpg://user:pw@host/db"))

# create tables on startup (dev/test only):
app = FastAPI(lifespan=lifespan(dsn, create_all=True))
Parameters:

  • dsn (str) – SQLAlchemy async database URL (e.g. "postgresql+asyncpg://user:pw@host/db").

  • create_all (bool) – when True, run CREATE TABLE IF NOT EXISTS for all registered models before the app starts accepting requests.

  • engine_kwargs (Any) – extra keyword arguments forwarded to create_async_engine (e.g. pool_size=10).

Returns:

an async context-manager factory compatible with FastAPI(lifespan=...).

Return type:

Callable[[Any], AbstractAsyncContextManager[None]]

See also

configure(), dispose() - the underlying engine helpers.

alchemiq.fastapi.repository(repo_or_model)[source]

Return a zero-arg callable that yields the resolved repository (use as a DI default).

Parameters:

repo_or_model (Any)

Return type:

Callable[[], Repository]

async alchemiq.fastapi.unit_of_work()[source]

Yield a UnitOfWork scoped to the request (FastAPI/FastStream dependency).

Return type:

AsyncIterator[UnitOfWork]

async alchemiq.fastapi.db_session()[source]

Yield a read-only AsyncSession scoped to the request (FastAPI/FastStream dependency).

Return type:

AsyncIterator[AsyncSession]

alchemiq.fastapi.install_exception_handlers(app)[source]

Register a global handler that converts PersistenceError to JSON responses.

Attaches a single app.add_exception_handler for PersistenceError. Each subtype is mapped to an HTTP status by status_for(); the response body is {"detail": "<exception message>"}.

E.g.:

from fastapi import FastAPI
from alchemiq.fastapi import install_exception_handlers, crud_router

app = FastAPI()
install_exception_handlers(app)
app.include_router(crud_router(User, prefix="/rows"))
Parameters:

app (FastAPI) – the FastAPI application instance.

Return type:

None

Note

InvalidCursorError is not caught by this handler. The list endpoint in crud_router() raises it as an HTTPException(400) directly.

See also

status_for() - the exception->status mapping.

alchemiq.fastapi.http_exception_for(exc)[source]

Convert a PersistenceError to an HTTPException with the mapped status code.

Parameters:

exc (PersistenceError) – a PersistenceError (or subclass) instance.

Returns:

an HTTPException whose status_code is determined by status_for() and whose detail is the string representation of exc.

Return type:

HTTPException

alchemiq.fastapi.status_for(exc)[source]

Return the HTTP status code for a PersistenceError.

The mapping is:

  • NotFoundError -> 404

  • MultipleResultsFound -> 409

  • ConcurrentModificationError -> 409

  • RelationNotLoaded -> 500

  • any other PersistenceError subtype -> 500 (fallback)

InvalidCursorError is not in this map - crud_router() raises it directly as an HTTPException with status 400.

Parameters:

exc (PersistenceError) – a PersistenceError (or subclass) instance.

Returns:

the HTTP status code integer.

Return type:

int

alchemiq.fastapi.read_schema(model)[source]

Return the cached Pydantic read schema for model (all serialisable fields).

The schema is derived from the model’s field metadata and cached on the class. crud_router() uses this automatically; call it directly to build a response schema for a custom endpoint.

Parameters:

model (type) – an alchemiq Model subclass.

Returns:

a BaseModel subclass whose fields match the serialisable columns.

Return type:

type[BaseModel]

alchemiq.fastapi.create_schema(model)[source]

Return a cached Pydantic schema for creating model instances.

Required fields are required; nullable fields are optional with a None default. PK, server-managed timestamps (created_at/updated_at), and deleted_at (on soft-delete models) are excluded so clients cannot supply them.

Parameters:

model (type) – an alchemiq Model subclass.

Returns:

a cached BaseModel subclass named {Model}Create.

Return type:

type[BaseModel]

alchemiq.fastapi.update_schema(model)[source]

Return a cached Pydantic schema for partial updates of model instances.

All writable fields are optional with a None default. crud_router() passes the body through model_dump(exclude_unset=True) so only the fields the client actually supplied are applied.

Parameters:

model (type) – an alchemiq Model subclass.

Returns:

a cached BaseModel subclass named {Model}Update.

Return type:

type[BaseModel]

alchemiq.fastapi.page_schema(read_model)[source]

Return a cached offset-page envelope schema wrapping read_model.

Produced automatically by crud_router() when pagination="offset" (default). The envelope fields are items, total, page, size, pages, has_next, and has_prev.

Parameters:

read_model (type[BaseModel]) – a Pydantic BaseModel (typically from read_schema()).

Returns:

a cached BaseModel subclass named {ReadModel}Page.

Return type:

type[BaseModel]

alchemiq.fastapi.cursor_page_schema(read_model)[source]

Return a cached cursor-page envelope schema wrapping read_model.

Produced automatically by crud_router() when pagination="cursor". The envelope fields are items, next_cursor, prev_cursor, has_next, and has_prev.

Parameters:

read_model (type[BaseModel]) – a Pydantic BaseModel (typically from read_schema()).

Returns:

a cached BaseModel subclass named {ReadModel}CursorPage.

Return type:

type[BaseModel]

alchemiq.fastapi.pk_name(model)[source]

Return the primary-key field name for model.

Parameters:

model (type) – an alchemiq Model subclass.

Returns:

the name of the field whose primary_key config is True.

Raises:

ConfigError – if no primary-key field is declared on model.

Return type:

str