Serve your app on the mock

A frontend team is blocked on a backend that isn't ready. The usual answer is a schema mock: point a tool at the OpenAPI document and let it synthesize responses. That gets you a stateless fake — POST /products returns something, and the list route never shows it.

A forze app has a better option. The in-memory mock is a faithful backend for every port: documents with real filters, sort, cursors and rev, transactions, storage, search, queues. So instead of faking the API, serve the real one on the mock — the same routes, handlers, middleware and identity plane, with MockDepsModule where the backend modules would be.

The property that makes this worth doing: contract drift is structurally impossible. There is no second artifact to keep in sync, because the routes are generated from the same frozen registry production serves.

The swap¶

Everything below is an ordinary app. The only mock-specific line is the deps module:

def build_runtime(identity: LocalIdentityConfig) -> ExecutionRuntime:
    """The only thing a mock server changes: which modules answer the ports.

    `MockDepsModule` registers every port as a *fallback*, so the app's real
    identity wiring composes on top of it in one registry — the local API-key verifier and
    tenant resolver win their routes, the mock keeps the other ~45 planes. Before that,
    this pairing raised at freeze and a mock context could not carry real identity at all.
    """

    deps = DepsRegistry.from_modules(MockDepsModule(state=MockState())).with_deps(
        local_identity_deps(identity, authn_route=AUTHN.name, tenancy_route=AUTHN.name),
    )

    return ExecutionRuntime(deps=deps.freeze())

MockDepsModule registers every port as a fallback (see hybrid contexts), so your real identity wiring composes on top of it in one registry — the local API-key verifier and tenant resolver win their routes, the mock keeps the other planes. The same mechanism lets you keep one real backend and mock the rest: add PostgresDepsModule(...) to the list and documents go to Postgres while search, queues and storage stay in memory.

The app is unchanged¶

The factory is the production one, parameterized by runtime:

registry = build_document_registry(product_spec).freeze()


def build_app(
    runtime: ExecutionRuntime,
    *,
    lifespan: Lifespan | None = None,
) -> FastAPI:
    """The app's own factory — production writes exactly this and passes a real runtime.

    Keeping the factory parameterized by runtime is what makes the swap total: routes,
    middleware, exception handlers and identity ingress are shared verbatim between the
    served mock and the deployed service. *lifespan* defaults to the production one
    (`runtime_lifespan`); the mock server passes a variant that seeds once the scope is
    open, which is the only seam it needs — the routes below never learn about it.
    """

    ctx = runtime.get_context

    router = APIRouter(prefix="/products")
    # Generated from the operation catalog — one route per registered kernel operation,
    # `operation_id` equal to the operation key (`products.get`, `products.list`, …).
    attach_document_routes(
        router,
        registry=registry,
        ns=product_spec.default_namespace,
        ctx_dep=ctx,
        style="rest",
    )

    app = FastAPI(
        title="Products API (mock)",
        lifespan=lifespan or runtime_lifespan(runtime),
    )
    app.include_router(router)
    register_exception_handlers(app)  # CoreException → HTTP: not_found → 404, conflict → 409
    app.add_middleware(InvocationMetadataMiddleware, ctx_dep=ctx)
    app.add_middleware(
        SecurityContextMiddleware,
        ctx_dep=ctx,
        authn=AuthnRequirement(
            ingress=(HeaderApiKeyAuthn(authn_spec=AUTHN, header_name="X-API-Key", required=True),),
        ),
        when_multiple_credentials="first_in_order",
    )

    return app

Routes come from the operation catalog, so operationId is the operation key (products.get, products.list) and the served contract is the app's own. Identity is your wiring too — here local identity reading a JSON key file. That is deliberate: the mock server has no way to mint a principal, so there is no dev-only auth bypass to accidentally ship.

Seed it¶

An API that answers [] is one no frontend can build against:

_CATALOG = (("Espresso", 250), ("Cortado", 320), ("Filter", 280))


async def seed(ctx: ExecutionContext) -> None:
    """Fill the empty store — an API that answers `[]` is one no frontend can build against.

    Seeds go through the **write path**, never into `MockState` directly, so `rev`,
    timestamps, materialized fields and field encryption are produced by the same code
    that serves the reads.
    """

    # The app's own eligibility gate reads a `policy_principal` document, so the dev key
    # needs one or authentication fails with "Principal not found". It has to be this
    # document write: the mock's `PrincipalRegistryPort.ensure_principal` is a no-op that
    # returns a ref without storing anything the real gate can then read.
    await ctx.doc.command(policy_principal_spec).create(
        CreatePolicyPrincipalCmd(kind="user"),
        id=DEV_PRINCIPAL,
    )

    for name, price in _CATALOG:
        await ctx.doc.command(product_spec).create(ProductCreate(name=name, price=price))

Seeds go through the write path, never into MockState directly, so rev, timestamps, materialized fields and field encryption come from the same code that serves the reads.

Three hand-written rows is the right size for a recipe and the wrong size for a real screen. forze_mock.seeding fills specs from a plan instead — generated rows for volume, fixtures for the rows a demo actually shows, and references that point at documents the seed created:

from forze_mock.seeding import SeedPlan, apply_seed, load_fixtures, spec_seed

plan = SeedPlan(
    specs=(
        spec_seed(project_spec, count=5),
        spec_seed(task_spec, count=40, fixtures=load_fixtures("tasks.json")),
    ),
    rng_seed=7,
)
result = await apply_seed(ctx, plan)   # result["tasks"] -> the created ids

A seeded Task.project_id names a seeded project — inferred from the field and spec names, corrected by SeedPlan.links where the names don't line up. The plan is reproducible: one rng_seed fixes the values, and SeedPlan.instant pins the clock the write path mints ids and timestamps from, so two processes running the plan produce byte-identical documents. Seeding needs polyfactory (it ships with the dst extra).

Run it¶

cd examples/recipes/mock_server
just run                    # http://localhost:8000 — no compose file, no containers
just smoke                  # the seeded catalog through the generated list route

Serve it with the CLI, and drive it¶

Composing the runtime by hand is fine for one app. MockApp is the declaration form — it says which real modules to keep, what to seed, and nothing about the app itself:

mock_app = MockApp(
    build_app=build_app,
    # Real modules would go in `modules=`; the mock composes under them as a fallback, so
    # "real Postgres for documents, mock for the rest" is this list and nothing else.
    deps=(local_identity_deps(from_json_path(KEY_FILE), authn_route=AUTHN.name),),
    seed=seed_plan,
)

FORZE_MOCK_SERVER=1 forze mock serve examples.recipes.mock_server.served:mock_app

The gate is not decoration: serve refuses without it, and refuses a composition whose deps contain no fallback-marked mock module — so a real runtime cannot be served here, and can never grow the routes below. Install with the mock-server extra.

The control plane¶

A frontend developer needs to provoke states, not wait for them. /_mock does that, and it sits beside your app rather than inside it, so your own auth does not lock you out of it:

# every products call fails as a real 409, once
curl -X POST localhost:8000/_mock/fault \
     -d '{"route": "products", "op": "create", "kind": "conflict", "times": 1}'

# the expiry screen, without waiting a day
curl -X POST localhost:8000/_mock/time -d '{"action": "advance", "seconds": 86400}'

The fault's kind is a real exc kind, so your own exception handlers turn it into the real status and error envelope — an armed conflict reaches the client exactly as a genuine optimistic-concurrency failure would. A control plane that invented its own error shape would be teaching the frontend a lie.

route is the spec name and op is the port method (create, update, find_page, get, …) — omit op to match every call on that spec, which is usually what you want.

emit is where a notification badge gets built: one signal, at one principal, on demand — which is what a traffic generator cannot give you. The realtime egress plane lives above forze_mock, so the server hands the signal to your on_emit and your own mailbox decides who receives it; see examples/recipes/realtime_sse/served.py. Note that /_mock/reset clears the mock's stores and not state your app holds itself, like that mailbox.

In a container¶

A frontend team should not need a Python toolchain to run your backend:

cd examples/recipes/mock_server
just up            # docker compose up --build, health-checked
just down

The recipe's Dockerfile and compose.yaml build from the repo root and serve $MOCK_APP — point that at your own declaration to serve your app. The image sets FORZE_MOCK_SERVER=1 deliberately: the container is the mock, so running it is the opt-in. The port is published on 127.0.0.1 for the reason above.

Multi-tenant data¶

If your specs are tenant-aware, the served mock partitions them the way a real relation's tenant WHERE clause does. Two API keys on two tenants, one server, disjoint data — the tenant comes from your identity wiring, so nothing about the app changes:

mock = MockDepsModule(routes={"notes": MockRouteConfig(tenant_aware=True)})

An unauthenticated request binds no tenant and is refused rather than shown everything.

What you get that a schema mock cannot give you¶

• create → list → get coherence, so a form, a table and a detail screen all work.
• Pagination cursors that terminate, because they are the real cursors.
• A stale rev that fails the way production fails — revision_mismatch, from the same gateway logic the real adapters run.
• Real error envelopes and status mapping, because register_exception_handlers is yours.

When a schema mock is still the right answer¶

If the consumer has no Python and no access to your app — an external partner, a public demo — export the document and hand it to the mature tooling:

cd examples/recipes/mock_server && just openapi > openapi.json
npx @stoplight/prism-cli mock openapi.json      # or MSW, orval, openapi-typescript

You lose statefulness and get schema-shaped responses; for a consumer who cannot run your app, that is the correct trade.

Keep the data across a restart¶

By default the mock forgets everything on exit. That is right for tests, where fresh state per case is the point, and wrong for the other thing this page gets used for: an MVP somebody is clicking through. Wire a snapshot file and Tuesday's orders are still there on Wednesday.

Four things it is not, stated before the wiring rather than after it:

• Not crash-durable. A kill -9, an OOM kill or a power cut discards everything since the last write. This is a snapshot, not a write-ahead log.
• Not multi-process. One process owns the file; a second is refused, naming the pid of the first.
• Not bounded by the file. The whole dataset is in RAM, exactly as it was before.
• Not a format you own. It is a pickle of MockState's own fields, so upgrading forze invalidates it — and the loader refuses rather than migrating.

None of that matters for the case this is for, and all of it matters the moment somebody mistakes it for a database. The upgrade path is Postgres.

persistence = MockStatePersistence(path=Path(".forze/mvp.state"))

lifecycle = LifecyclePlan.from_steps(
    mock_state_lifecycle_step(state=state, persistence=persistence),
)

state is the same MockState the deps module was built with. Startup loads the file when there is one — a missing file is a first run, not an error — and shutdown writes it.

Every plane the mock implements comes back: documents, counters, outbox and inbox rows, stored objects, identity, durable runs. What deliberately does not is anything whose meaning is local to the process that wrote it. Held locks and in-flight transactions restore as none held, none active, because a lock expiry measured against one process's clock means nothing against another's.

To write periodically as well as at shutdown, set an interval:

MockStatePersistence(path=Path(".forze/mvp.state"), flush_every=timedelta(minutes=5))

Off by default: it needs a background task, which a serverless host cannot keep running between invocations. Turn it on when losing a session to a crash stops being acceptable — for most people, right after the first time it happens.

One pairing is refused outright. Persistence alongside a per-tenant routed state registry would snapshot the unrouted state and silently miss every tenant's data, so it fails at startup rather than writing a file that looks fine and is empty.

Limits¶

Never serve this in production. MockDepsModule keeps everything in memory and enforces none of the durability, isolation or capability limits a real backend does. It is a laptop and CI tool.

And the mock is not the specification — the conformance battery is. A behaviour your frontend depends on should be one the adapter conformance suite pins across real backends, not one you discovered against the in-memory store.

See also¶

• Testing — the mock in unit tests, and hybrid contexts
• Local identity — the API-key file this recipe authenticates with
• CRUD over Postgres — the same shape with a real backend