FastAPI with shared caching and Granian¶
Run multiple actual HTTP workers that reuse product documents and one MongoDB change stream. One parent starts SharedCacheServer; Granian passes its connection to spawned application loaders. Each worker's FastAPI lifespan creates an AsyncMongoClient and attaches an asyncio SharedCacheManager. The owner stays outside the workers.
The product read retains both its _id and tenant_id predicate across workers. The demonstration uses the fixed tenant demo; an application supplies its own trusted tenant context. See manager selection for the full comparison.
Run and verify¶
From the repository checkout, with the example prerequisites:
The self-check starts Granian with two spawned workers on a free localhost port and uses real HTTP connections. It resets and cleans client_query_cache_example_fastapi_shared; use a disposable deployment meeting the MongoDB requirements. It fails unless:
- Two different worker PIDs answer HTTP requests.
- Worker A warms a product and records a cache hit.
- Worker B's first product read records a shared hit and zero MongoDB
findcommands. - A raw PATCH returns the updated document, invalidation forces a refetch, and both workers subsequently hit the updated value.
- Both workers close their managers before their clients, and the parent closes the cache server after worker shutdown.
The response includes worker-local cumulative counters. The optional worker_pid query parameter rejects a request handled by another worker before accessing storage, so verification can target each process without warming the second worker accidentally. This is demonstration instrumentation, not a routing mechanism for an application. Snapshot differences only identify individual requests in this isolated, sequential self-check.
The program prints evidence resembling:
shared hit: worker 123 warmed; worker 124 read with zero MongoDB finds
invalidation: both workers hit the updated document after a raw PATCH
worker 123: shared manager closed before client
worker 124: shared manager closed before client
parent: shared cache server closed after workers
Startup and admission can take several reads. The check allows preceding cached values after PATCH while waiting for updated hits; its deadlines bound the demonstration, not production freshness. There is no TTL or manual cache clear in the check. See consistency.
Serve the application¶
uv run examples/fastapi_shared_example.py --serve --workers 4 --port 8000
curl http://127.0.0.1:8000/health
curl http://127.0.0.1:8000/products/book
Serve mode leaves existing data intact. Insert a product into client_query_cache_example_fastapi_shared.products using your normal MongoDB tools, for example {_id: "book", description: "A product description", tenant_id: "demo"}; a missing product returns 404. Update it with:
curl -X PATCH http://127.0.0.1:8000/products/book \
-H 'Content-Type: application/json' \
-d '{"description":"An updated description"}'
PATCH uses PyMongo's find_one_and_update(..., return_document=AFTER) with majority write concern. Its response comes from the write, while subsequent GETs can lag until invalidation arrives. Stop with Ctrl+C for graceful cleanup.
This localhost demonstration has no authentication. Add trusted authentication and restrict diagnostics before exposing it. All shared workers must have the same database read visibility. Never remove tenant or permission predicates to make a query eligible for sharing.
Application lifecycle¶
The example pins Granian 2.8.4 and uses its programmatic serve(target_loader=..., wrap_loader=False) API. A picklable functools.partial carries only the connection; it creates the FastAPI app inside each worker. The script selects the spawn process method and runs the ASGI lifespan in each worker. The ordinary granian module:app --workers N command alone cannot distribute this parent's live connection; run the provided launcher.
def create_app(connection: SharedCacheConnection) -> FastAPI:
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
reads = DatabaseReads()
async with (
AsyncMongoClient[Product](
MONGODB_URI,
event_listeners=[reads],
serverSelectionTimeoutMS=5000,
timeoutMS=5000,
) as client,
SharedCacheManager(client, connection) as manager,
):
raw = client[DATABASE_NAME].get_collection(
"products",
read_concern=ReadConcern("majority"),
write_concern=WriteConcern(w="majority"),
)
app.state.resources = Resources(
raw, manager.get_cached_collection(raw), manager, reads
)
print(f"worker {os.getpid()}: shared manager started", flush=True)
yield
print(f"worker {os.getpid()}: shared manager closed before client", flush=True)
app = FastAPI(lifespan=lifespan)
app.get("/health")(health)
app.get("/products/{product_id}")(read_product)
app.patch("/products/{product_id}")(update_product)
return app
def health() -> WorkerIdentity:
return WorkerIdentity(pid=os.getpid())
async def read_product(
product_id: NonEmptyStr,
request: Request,
worker_pid: PositiveInt | None = None,
) -> Observation:
# Targeting is only for the demonstration: a rejected worker performs no read.
if worker_pid is not None and worker_pid != os.getpid():
raise HTTPException(status_code=409, detail="Try another worker")
owned = resources(request)
product = await owned.cached.find_one({"_id": product_id, "tenant_id": "demo"})
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
counts = owned.manager.snapshot()
return Observation(
pid=os.getpid(),
product=product,
hits=counts.hits,
misses=counts.misses,
bypasses=counts.bypasses,
database_reads=owned.reads.count,
)
async def update_product(
product_id: NonEmptyStr,
update: DescriptionUpdate,
request: Request,
) -> Product:
product = await resources(request).raw.find_one_and_update(
{"_id": product_id, "tenant_id": "demo"},
{"$set": {"description": update.description}},
return_document=ReturnDocument.AFTER,
)
if product is None:
raise HTTPException(status_code=404, detail="Product not found")
return product
def serve(workers: PositiveInt, port: PositiveInt) -> None:
with SharedCacheServer(
MONGODB_URI,
(DATABASE_NAME,),
client_options={"serverSelectionTimeoutMS": 5000, "timeoutMS": 5000},
) as cache_server:
Granian(
"shared-products",
interface=Interfaces.ASGI,
workers=workers,
address="127.0.0.1",
port=port,
log_level=LogLevels.error,
workers_kill_timeout=10,
).serve(
target_loader=partial(create_app, cache_server.connection),
wrap_loader=False,
)
print("parent: shared cache server closed after workers", flush=True)
The complete source includes imports, response types, command counting and the HTTP self-check. It is exercised by just examples. The parent and each worker use bounded MongoDB operation timeouts; configure those for your own service and shutdown policy. See shared deployment for capacity, outage behavior and security.