Read-only document API

When a service only reads an aggregate — a projection, a lookup, a reporting view owned by someone else — give its spec write=None. No command port is registered, so the API can only query, and the wiring stays minimal.

The runnable version lives at examples/recipes/read_only/ — just run brings up ephemeral Postgres, seeds a couple of rows, and serves the read API.

A read-only spec¶

write=None is the whole opt-in. The read model still inherits id, rev, and timestamps from ReadDocument:

class ArticleRead(ReadDocument):  # inherits id, rev, created_at, last_update_at
    title: str
    body: str


ARTICLE_SPEC = DocumentSpec(name="articles", read=ArticleRead, write=None)

Wire it read-only¶

Register the document under ro_documents (not rw_documents) with a PostgresReadOnlyDocumentConfig — it carries only the read relation, no write tables or bookkeeping, and no transaction route is needed:

# Read-only config: just the read relation — no write side, no bookkeeping.
ARTICLE_PG = PostgresReadOnlyDocumentConfig(read=("public", "articles"))


def build_runtime(pg: PostgresClient, *, dsn: str) -> ExecutionRuntime:
    deps = DepsRegistry.from_modules(
        PostgresDepsModule(client=pg, ro_documents={"articles": ARTICLE_PG}),
    )
    lifecycle = LifecyclePlan.from_modules(
        PostgresLifecycleModule(client=pg, dsn=dsn, config=PostgresConfig()),
    )
    return ExecutionRuntime(deps=deps.freeze(), lifecycle=lifecycle.freeze())

Data gets in elsewhere

A read-only doc has no command port — ctx.document.command(spec) won't resolve. The writer is another service (or a migration); the example seeds rows directly through the client just to be self-contained.

Query routes¶

Resolve ctx.document.query(spec) and call its read methods:

@asynccontextmanager
async def lifespan(app: FastAPI):
    pg = PostgresClient()
    dsn = os.environ.get("POSTGRES_DSN", "postgresql://forze:forze@localhost:5432/forze")
    _rt.set_once(build_runtime(pg, dsn=dsn))
    async with _rt.get().scope():
        await pg.execute(SCHEMA)
        await seed(pg)
        yield


app = FastAPI(title="Articles API (read-only)", lifespan=lifespan)
register_exception_handlers(app)


@app.get("/articles/{article_id}")
async def get_article(article_id: UUID) -> ArticleRead:
    # `get` raises not_found (→ 404) on a miss; use `find` for the None variant.
    return await ctx().document.query(ARTICLE_SPEC).get(article_id)


@app.get("/articles")
async def list_articles(limit: int = 20, offset: int = 0) -> list[ArticleRead]:
    page = await ctx().document.query(ARTICLE_SPEC).find_many(
        pagination={"limit": limit, "offset": offset}
    )
    return list(page.hits)

The query port gives you the full read surface — pick by how you want misses handled:

Method	Returns	On miss
`get(id)`	the document	raises `not_found` → 404
`get_many(ids)`	a list	raises `not_found` if any is missing
`find(filters)`	the document or `None`	returns `None`
`find_many(filters, pagination, sorts)`	a `CountlessPage` (`.hits`)	empty page
`find_page(...)`	a `Page` (adds `.count`)	empty page
`count(filters)`	an `int`	`0`

Filters and sorts use the query DSL; pagination is {"limit": …, "offset": …}.