Quickstart

In about ten minutes — and without Docker — you'll have a running REST service whose handlers have no idea what's storing the data. Later you point them at Postgres by changing wiring, not code.

What you will build¶

A minimal REST service for a User aggregate:

Storage is in-memory — no Docker, no migrations. The complete, runnable file is examples/quickstart/app.py; the steps below build it up.

Step 1 — Create the project¶

uv init forze-quickstart
cd forze-quickstart
uv add 'forze[fastapi]'

Everything below goes into a single main.py, starting with the imports — the essentials come from the forze and forze_kits top-level packages:

from uuid import UUID

from fastapi import FastAPI
from pydantic import computed_field

from forze import (
    BaseDTO,
    Document,
    DocumentSpec,
    DocumentWriteTypes,
    ReadDocument,
    build_runtime,
)
from forze_fastapi import runtime_lifespan
from forze_fastapi.exceptions import register_exception_handlers
from forze_kits import Paginated, build_document_registry, document_facade
from forze_kits.aggregates.document import DocumentIdDTO, ListRequestDTO
from forze_mock import MockDepsModule

Step 2 — Define the domain models¶

An aggregate needs a domain model, a create command, and a read model. Document gives you id, rev, and timestamps for free.

class User(Document):
    name: str
    email: str | None = None


class CreateUserCmd(BaseDTO):
    name: str
    email: str | None = None


class ReadUser(ReadDocument):
    name: str
    email: str | None = None

    @computed_field
    @property
    def email_provided(self) -> bool:
        return self.email is not None

Step 3 — Declare a specification¶

The specification is the logical name — "users" — that ties the models to their operations and, later, to adapters.

user_spec = DocumentSpec(
    name="users",
    read=ReadUser,
    write=DocumentWriteTypes(domain=User, create_cmd=CreateUserCmd),
)

Step 4 — Build the operation registry¶

build_document_registry assembles the standard CRUD operations; freeze() makes the registry immutable and shareable. The request/response DTOs are derived from the spec, so you don't restate them — pass an explicit DocumentDTOs only to override or to disable an operation.

# DTOs are derived from the spec; pass an explicit DocumentDTOs only to override.
registry = build_document_registry(user_spec).freeze()

Step 5 — Wire the runtime¶

MockDepsModule provides in-memory adapters for every contract. build_runtime assembles an ExecutionRuntime around it. Then document_facade(...) returns a small factory: call users() to get a fully-typed DocumentFacade bound to the runtime's current scope context — fresh each call, never cached across requests.

runtime = build_runtime(MockDepsModule())

# A per-call, fully-typed facade factory bound to the runtime's current context.
users = document_facade(runtime, registry, user_spec)

Step 6 — Attach the routes¶

runtime_lifespan runs the runtime inside the app's lifespan. Each route calls users() for the facade and runs an operation — the handlers never touch HTTP:

app = FastAPI(title="Users API", lifespan=runtime_lifespan(runtime))
register_exception_handlers(app)  # CoreException → HTTP (e.g. not_found → 404)


@app.post("/users")
async def create_user(cmd: CreateUserCmd) -> ReadUser:
    return await users().create(cmd)


@app.get("/users/{user_id}")
async def get_user(user_id: UUID) -> ReadUser:
    return await users().get(DocumentIdDTO(id=user_id))


@app.get("/users")
async def list_users(page: int = 1, size: int = 10) -> Paginated[ReadUser]:
    return await users().list(ListRequestDTO(page=page, size=size))


@app.delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: UUID) -> None:
    await users().kill(DocumentIdDTO(id=user_id))

register_exception_handlers maps a CoreException to a response, so a missing user comes back as a 404. (Routes are hand-wired here to show the moving parts — generated routes can attach the CRUD endpoints from the registry instead.)

Step 7 — Run it¶

uv run uvicorn main:app --reload

Open http://localhost:8000/docs for the interactive explorer, or try it from the shell:

# Create — note the id in the response
curl -s -X POST http://127.0.0.1:8000/users \
  -H 'Content-Type: application/json' \
  -d '{"name": "Ada", "email": "ada@example.com"}'

curl -s http://127.0.0.1:8000/users            # list
curl -s http://127.0.0.1:8000/users/<id>       # get one
curl -s -X DELETE http://127.0.0.1:8000/users/<id>   # delete

What you just did¶

You built a complete service without a single line of HTTP or storage code in your domain:

• A User aggregate with its command and read models — pure Python, no infrastructure.
• A specification and a frozen operation registry — the named operations the service exposes.
• An ExecutionRuntime wired to in-memory adapters, opened for the app's lifetime.
• Routes that resolve operations from the context and return read models.

The only thing tying this to "in-memory" is MockDepsModule in Step 5. Swap it for PostgresDepsModule + RedisDepsModule and the domain, spec, registry, and routes don't change — that's the whole point. The PostgreSQL integration shows the swap.

Where to go next¶