fastbff 0.1.0

Simple back-end for front-end using Pydantic. Declarative data composition with typed transformers, dependency injection, and automatic N+1 avoidance.

fastbff

Simple back-end for front-end using Pydantic. Declarative data composition with typed transformers, dependency injection, and automatic N+1 avoidance. Suitable for modular monolithic systems.

Features

• Declarative data composition — describe the shape of a response once on a Pydantic model; fetching happens automatically.
• One-call orchestration — app.executor.render(Model, rows) runs Plan + Fetch + Merge for a whole page in a single line.
• Typed queries — Query[T] carries its own return type, or register a plain function with a typed signature; both forms cache identically.
• Automatic N+1 avoidance — transformers declare a BatchArg[T] and the framework plans a single bulk fetch per page instead of one call per row.
• Two-level cache — call-level (identical query args) plus entity-level (overlapping ID sets are merged into one fetch with only the missing ids).
• Dependency injection — built on FastAPI's Depends; the same QueryExecutor / repository / session is shared across every transformer in a request scope.
• Routers — register handlers locally on a QueryRouter and merge them into a FastBFF app with app.include_router(router), mirroring FastAPI's APIRouter.

Install

pip install fastbff

Runtime deps: pydantic>=2, fastapi>=0.100. Python 3.12+ (uses PEP 695 generics).

Quickstart

from dataclasses import dataclass
from typing import Annotated

from fastapi import Depends
from pydantic import BaseModel

from fastbff import (
    FastBFF,
    BatchArg,
    Query,
    QueryExecutor,
    build_transform_annotated,
)

# --- Domain -----------------------------------------------------------------

@dataclass(frozen=True)
class User:
    id: int
    name: str

# --- App --------------------------------------------------------------------

app = FastBFF()

# --- Bulk query -------------------------------------------------------------

class FetchUsers(Query[dict[int, User]]):
    ids: frozenset[int]

@app.queries
def fetch_users(args: FetchUsers) -> dict[int, User]:
    return {i: User(id=i, name=f'u{i}') for i in args.ids}

# --- Transformer + Response model ------------------------------------------

@app.transformer
def transform_owner(
    owner_id: int,
    batch: BatchArg[int],
    query_executor: Annotated[QueryExecutor, Depends(QueryExecutor)],
) -> User | None:
    users = query_executor.fetch(FetchUsers(ids=batch.ids))
    return users.get(owner_id)

OwnerTransformerAnnotated = build_transform_annotated(transform_owner)

class TeamDTO(BaseModel):
    id: int
    owner: OwnerTransformerAnnotated

# --- Handler ---------------------------------------------------------------

@app.injector.entrypoint
def render_teams_page() -> list[TeamDTO]:
    rows = [
        {'id': 1, 'owner': 10},
        {'id': 2, 'owner': 20},
        {'id': 3, 'owner': 10},  # duplicate id → still just one DB call
    ]
    return app.executor.render(TeamDTO, rows)

A single page of N rows issues one fetch_users(...) call — regardless of N, and regardless of how many duplicate ids the rows contain.

Two-phase execution (under the hood)

app.executor.render(Model, rows) does two things:

Phase 1 — Plan    populate_context_with_batch(Model, rows)
                  → walks rows, collects every unique id for every BatchArg field
                    into {batch_key: set[ids]}

Phase 2 — Merge   Model.model_validate(row, context=ctx) for each row
                  → each @transformer runs with dependencies injected; the first
                    row's executor.fetch(...) issues one bulk call covering the
                    whole page, subsequent rows hit the entity-level cache

You can run either phase manually if you need to — see populate_context_with_batch, get_model_batches, and executor.fetch / executor.call.

Core concepts

Query[T] + @queries

A Query[T] subclass is a typed request object whose return type T is recovered from Pydantic's own generic metadata.

class FetchUsers(Query[dict[int, User]]):
    ids: frozenset[int]

@app.queries
def fetch_users(args: FetchUsers) -> dict[int, User]:
    ...

Return-type mismatches raise QueryRegistrationError at registration time, not at runtime.

Function-signature queries

If you don't want a Query[T] subclass per call, register a plain typed function and dispatch via app.executor.call:

@app.queries
def fetch_users(ids: frozenset[int]) -> dict[int, User]:
    ...

users = app.executor.call(fetch_users, ids=frozenset({1, 2, 3}))

The same call-level + entity-level caches apply.

QueryExecutor.fetch / QueryExecutor.call

Per-request dispatcher with two caching layers:

• Call-level — identical query args return the cached result.
• Entity-level — for dict[K, V]-returning queries whose request has an Iterable field, overlapping ID sets are merged. A second call with ids {2, 3, 4} after the first with {1, 2, 3} only fetches {4}. Absent ids (returned {} from the backend) are remembered too, so asking again doesn't hit the backend.

Absence is cached per-executor (per-request). With FastAPI integration (below) each request gets a fresh QueryExecutor automatically.

@transformer + build_transform_annotated

A transformer is a plain function with a return type annotation. @app.transformer registers it and returns the function unchanged — directly callable in tests. Use build_transform_annotated(func) to build a Pydantic-ready Annotated[ReturnType, TransformerFieldInfo] alias; bind it to a PascalCase <Name>TransformerAnnotated name and use it directly as a field type:

@app.transformer
def transform_owner(
    owner_id: int,
    query_executor: Annotated[QueryExecutor, Depends(QueryExecutor)],
) -> User | None:
    ...

OwnerTransformerAnnotated = build_transform_annotated(transform_owner)

class TeamDTO(BaseModel):
    owner: OwnerTransformerAnnotated

The return type baked into the alias is exactly the function's declared return type (including Optional, list[...], etc.) — reuse the alias on as many models as you like:

class CommentDTO(BaseModel):
    author: OwnerTransformerAnnotated

For unit testing, recover the DI-wrapped underlying callable with transformer_callable:

from fastbff import transformer_callable

call = transformer_callable(transform_owner)
assert call(owner_id=1, query_executor=fake) == User(id=1, name='…')

BatchArg[T]

Declaring a BatchArg[T] parameter on a transformer opts into bulk fetching. The parameter carries the full set of ids for this field on the current page, collected by Phase 1 of executor.render(...):

@app.transformer
def transform_owner(
    owner_id: int,
    batch: BatchArg[int],            # all ids for this field on the current page
    query_executor: Annotated[QueryExecutor, Depends(QueryExecutor)],
) -> User | None:
    users = query_executor.fetch(FetchUsers(ids=batch.ids))
    return users.get(owner_id)

The first row's executor.fetch(FetchUsers(ids=batch.ids)) issues the bulk call; subsequent rows hit the query executor's entity-level cache. One DB call per page, regardless of row count.

Dependency injection

InjectorRegistry wraps FastAPI's Depends. Registration decorators (@app.queries, @app.transformer) automatically wrap your callables so FastAPI-style dependencies resolve at call time:

@app.queries
def fetch_users(args: FetchUsers, session: DBSession) -> dict[int, User]:
    # `session: DBSession` is Annotated[Session, Depends(get_session)] elsewhere
    ...

@app.injector.entrypoint
def handler() -> ...:
    # `entrypoint` opens a fresh dependency scope for this call
    ...

app.bind(InterfaceOrAnnotatedAlias, factory) registers a provider — both a bare class and its Annotated[Class, Depends(Class)] alias resolve to the same override entry, so pass whichever is convenient. Works equally well with test doubles:

app.bind(QueryExecutor, lambda: shared_executor)
app.bind(SomeService, lambda: FakeService())

QueryRouter + app.include_router

For multi-module apps, register handlers locally on a QueryRouter and attach the whole bundle to a FastBFF app at composition time — exactly like FastAPI's APIRouter:

from fastbff import FastBFF, QueryRouter

# users/handlers.py
router = QueryRouter()

@router.queries
def fetch_users(args: FetchUsers) -> dict[int, User]: ...

@router.transformer
def transform_owner(
    owner_id: int,
    batch: BatchArg[int],
    query_executor: Annotated[QueryExecutor, Depends(QueryExecutor)],
) -> User | None: ...

# main.py
app = FastBFF()
app.include_router(router)

include_router merges the router's queries into the app's registry and rewires the router's DI plumbing to share the app's. Field annotations built via build_transform_annotated continue to work — no rebuilding required.

Duplicate registrations (same Query subclass or same function on both router and app) raise QueryRegistrationError at include time so collisions surface during composition, not at runtime.

FastAPI integration

QueryExecutor is request-scoped naturally: annotate handler parameters as Annotated[QueryExecutor, Depends(QueryExecutor)] and FastAPI's own Depends(...) pipeline will resolve a fresh instance per request. A complete route:

from collections.abc import Iterator
from typing import Annotated

from fastapi import Depends, FastAPI
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker

from fastbff import (
    FastBFF, BatchArg, Query, QueryExecutor, build_transform_annotated,
)

# --- SQLAlchemy wiring -----------------------------------------------------

engine = create_engine('postgresql+psycopg://localhost/app')
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

def get_db_session() -> Iterator[Session]:
    with SessionLocal() as session:
        yield session

DBSession = Annotated[Session, Depends(get_db_session)]

# --- App + route -----------------------------------------------------------

app = FastBFF()
fastapi_app = FastAPI()

class FetchUsers(Query[dict[int, User]]):
    ids: frozenset[int]

@app.queries
def fetch_users(args: FetchUsers, session: DBSession) -> dict[int, User]:
    stmt = select(UserRow).where(UserRow.id.in_(args.ids))
    rows = session.execute(stmt).scalars().all()
    return {row.id: User(id=row.id, name=row.name) for row in rows}

@app.transformer
def transform_owner(
    owner_id: int,
    batch: BatchArg[int],
    query_executor: Annotated[QueryExecutor, Depends(QueryExecutor)],
) -> User | None:
    return query_executor.fetch(FetchUsers(ids=batch.ids)).get(owner_id)

OwnerTransformerAnnotated = build_transform_annotated(transform_owner)

class TeamDTO(BaseModel):
    id: int
    owner: OwnerTransformerAnnotated

@fastapi_app.get('/teams', response_model=list[TeamDTO])
def list_teams(
    query_executor: Annotated[QueryExecutor, Depends(QueryExecutor)],
    session: DBSession,
) -> list[TeamDTO]:
    rows = session.execute(select(TeamRow)).mappings().all()
    return query_executor.render(TeamDTO, rows)

The DBSession alias is a plain FastAPI Depends(...) — fastbff's @app.queries and @app.transformer decorators wrap your callable with the injector, so FastAPI-style Depends parameters resolve at call time exactly as they would in a FastAPI route handler. The same Session instance is reused across every query/transformer in a single request.

Spell out the Annotated[QueryExecutor, Depends(QueryExecutor)] form at every use site — FastAPI walks the Annotated metadata and resolves a fresh QueryExecutor per request (per-request cache, per-request absence tracking). Override providers in tests via FastAPI's standard fastapi_app.dependency_overrides, or the injector's own app.bind(...).

Testing with QueryExecutorMock

from fastbff import QueryExecutorMock

mock = QueryExecutorMock(queries_registry=app.queries)
mock.stub_query(FetchUsers, {10: User(id=10, name='u10')})

assert mock.fetch(FetchUsers(ids=frozenset({10}))) == {10: User(id=10, name='u10')}
mock.reset_mock()  # clear stubs; subsequent fetch() calls hit real @queries handlers

Errors

All errors raised by the library subclass FastBFFError. Common ones:

• QueryRegistrationError — bad @queries declaration (missing return type, mismatch), or duplicate registration when including a router.
• TransformerRegistrationError — bad @transformer declaration, or build_transform_annotated called on an unregistered function.
• QueryNotRegisteredError — fetch/call against an unregistered handler.
• BatchContextMissingError — transformer with BatchArg invoked without context (forgot populate_context_with_batch or executor.render).
• DependencyResolutionError — one or more Depends(...) parameters failed to resolve.
• DependencyOverrideError — DependenciesSetup.override(...) targeted an unregistered interface.

Development

This project uses uv for dependency management, ruff for lint + format, ty for type checking, and pre-commit to run them on every commit.

uv sync                         # install project + dev deps into .venv
uv run pytest                   # run the test suite
uv run ruff check . --fix       # lint + autofix
uv run ruff format .            # format
uv run ty check src             # type check
uv run pre-commit install       # install git hooks
uv run pre-commit run --all-files

Tests are colocated with the modules they exercise, using the _test.py suffix (e.g. src/fastbff/query_executor/query_executor_test.py). The cross-cutting three-phase integration test lives at integration_test.py in the project root.

License

MIT