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¶

Build the document registry with only the read DTO, then call the typed facade:

# A read-only spec (write=None) derives a read-only DTO mapping.
registry = build_document_registry(article_spec).freeze()


@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)


def articles() -> DocumentFacade[ArticleRead, BaseDTO, BaseDTO]:
    return DocumentFacade(
        ctx=ctx(),
        registry=registry,
        namespace=article_spec.default_namespace,
    )


@app.get("/articles/{article_id}")
async def get_article(article_id: UUID) -> ArticleRead:
    # `get` raises not_found (→ 404) on a miss.
    return await articles().get(DocumentIdDTO(id=article_id))


@app.get("/articles")
async def list_articles(page: int = 1, size: int = 20) -> Paginated[ArticleRead]:
    return await articles().list(ListRequestDTO(page=page, size=size))

The facade gives driving code a typed operation surface without exposing the query port:

Method	Returns	On miss
`get(DocumentIdDTO(...))`	the document	raises `not_found` → 404
`list(ListRequestDTO(...))`	a `Paginated` result	empty page

ListRequestDTO carries page, size, filters, and sorts. Filters and sorts use the query DSL.