Compact FastAPI application¶

This application owns one AsyncMongoClient and one CacheManager per lifespan. GET reads use a cached collection; PATCH writes and reads its response directly in a causally consistent session with majority concerns. /metrics/cache returns the public cache snapshot. The manager closes before its client.

Run the self-check from the repository checkout:

uv run examples/fastapi_minimal_example.py

Use a disposable MongoDB 8.0+ replica set. The self-check resets and cleans client_query_cache_example_fastapi_minimal; the application lifespan itself never resets data. See example prerequisites for connection settings. The self-check uses FastAPI's in-process client and exposes no HTTP listener. Supply your own authentication before serving these endpoints; the advanced catalogue demonstrates trusted identity and tenant separation.

Application¶

The application and verification share one canonical source. Imports are shown separately from the application below.

import os
from contextlib import asynccontextmanager
from dataclasses import dataclass
from functools import partial
from time import monotonic, sleep
from typing import TYPE_CHECKING, Annotated, TypedDict

from fastapi import FastAPI, HTTPException, Request
from fastapi.testclient import TestClient
from pydantic import BaseModel, ConfigDict, Field
from pymongo import AsyncMongoClient, MongoClient, ReadPreference
from pymongo.read_concern import ReadConcern
from pymongo.write_concern import WriteConcern

from client_query_cache.asynchronous import (
    BypassReason,
    CacheCoreConfig,
    CacheManager,
    CacheSnapshot,
    StreamHealthStatus,
)

if TYPE_CHECKING:
    from collections.abc import AsyncGenerator, Awaitable, Callable

    from pymongo.asynchronous.collection import AsyncCollection

    from client_query_cache.asynchronous import CachedCollection

DATABASE_NAME = "client_query_cache_example_fastapi_minimal"
MONGODB_URI = os.getenv(
    "MONGODB_URI", "mongodb://localhost:27017/?directConnection=true"
)
NonEmptyStr = Annotated[str, Field(min_length=1)]


class ProductDocument(TypedDict):
    _id: NonEmptyStr
    description: NonEmptyStr


class DescriptionUpdate(BaseModel):
    model_config = ConfigDict(extra="forbid")
    description: Annotated[str, Field(min_length=1, max_length=200)]


@dataclass
class Resources:
    client: AsyncMongoClient[ProductDocument]
    manager: CacheManager[ProductDocument]
    raw: AsyncCollection[ProductDocument]
    cached: CachedCollection[ProductDocument]


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
    async with (
        AsyncMongoClient[ProductDocument](MONGODB_URI) as client,
        CacheManager(
            client, cache_config=CacheCoreConfig(max_entry_bytes=1024)
        ) as manager,
    ):
        raw = client[DATABASE_NAME].get_collection(
            "products",
            read_preference=ReadPreference.PRIMARY,
            read_concern=ReadConcern("majority"),
            write_concern=WriteConcern(w="majority"),
        )
        app.state.resources = Resources(
            client, manager, raw, manager.get_cached_collection(raw)
        )
        yield


app = FastAPI(lifespan=lifespan)


def resources(request: Request) -> Resources:
    owned: Resources = request.app.state.resources
    return owned


@app.get("/products/{product_id}")
async def read_product(product_id: NonEmptyStr, request: Request) -> ProductDocument:
    product = await resources(request).cached.find_one({"_id": product_id})
    if product is None:
        raise HTTPException(status_code=404, detail="Product not found")
    return product


@app.patch("/products/{product_id}")
async def update_product(
    product_id: NonEmptyStr,
    update: DescriptionUpdate,
    request: Request,
) -> ProductDocument:
    owned = resources(request)
    async with owned.client.start_session(causal_consistency=True) as session:
        await owned.raw.update_one(
            {"_id": product_id},
            {"$set": {"description": update.description}},
            session=session,
        )
        product = await owned.raw.find_one({"_id": product_id}, session=session)
    if product is None:
        raise HTTPException(status_code=404, detail="Product not found")
    return product


@app.get("/metrics/cache")
def cache_metrics(request: Request) -> CacheSnapshot:
    return resources(request).manager.snapshot()

The 1 KiB entry limit makes an oversized read easy to observe in the eligibility exercise. Choose your application's budget using capacity guidance.

Read and write observations¶

The self-check verifies a miss followed by a hit, then updates the document through PATCH. The PATCH response contains the updated description without using the cache. It prints subsequent GET observations until an updated cache hit is observed, and checks the stream's invalidation count increased.

One possible ordering is:

GET: original description, cache hit
PATCH: write acknowledged, direct response contains updated description
GET: original description, cache hit while the event is in flight
stream processes the event and invalidates the stored result
GET: updated description, cache miss
GET: updated description, cache hit

The event can arrive before the first GET after PATCH, so that GET can already return the new value. The program accepts both orderings; it does not force a stale read or print an invented event arrival time. Its five-second polling deadline is a self-check limit, not a freshness guarantee. Direct reads do not refresh cached content; see consistency.

For deployment flags, pagination, and authorization, use the catalogue reference. For exporting counters to your telemetry system, use monitoring.