FastAPI¶

Wire a storix session into FastAPI with dependency injection: create one session for the app's lifetime, inject it into handlers, and stream uploads straight to storage.

The pattern¶

One session, shared. Build it in the lifespan and close it on shutdown, so setup is paid once and every request reuses it.
Inject with Depends. A tiny get_fs dependency hands the session to any route, which keeps handlers testable: override it with a MemoryBackend session in tests, no disk and no cloud.
Stream, do not buffer. echo takes an async iterator, so an upload flows to storage a chunk at a time and memory stays flat regardless of file size.
Hand back links, not bytes. url() returns a short-lived presigned link (a data: URL on backends without native presigning, via DataUrlLayer).

The full example¶

"""Use storix inside FastAPI with dependency injection.

One session is created for the app's lifetime and injected into route
handlers. Uploads are streamed to storage in chunks, so a large file never
lands in memory whole.

Run:  uv run --with "fastapi[standard]" fastapi dev samples/recipes/fastapi_di.py
"""

from __future__ import annotations

from contextlib import asynccontextmanager
from typing import Annotated

from fastapi import Depends, FastAPI, Request, UploadFile

from storix.aio import DataUrlLayer, Storix, get_storage


@asynccontextmanager
async def lifespan(app: FastAPI):
    # One session for the whole app. get_storage reads STORIX_* from the
    # environment, so the provider (local, azure, ...) is configured, not
    # hard-coded. DataUrlLayer backfills url() so it works on any backend.
    app.state.fs = get_storage().with_layer_missing(DataUrlLayer)
    try:
        yield
    finally:
        await app.state.fs.close()


app = FastAPI(lifespan=lifespan)


def get_fs(request: Request) -> Storix:
    """The shared storix session, injected into route handlers."""
    return request.app.state.fs


FS = Annotated[Storix, Depends(get_fs)]


@app.post('/files/{name}')
async def upload(name: str, file: UploadFile, fs: FS) -> dict[str, str]:
    async def chunks():
        while data := await file.read(1 << 20):  # 1 MiB at a time
            yield data

    await fs.echo(chunks(), f'/uploads/{name}')  # streamed, bounded memory
    return {'stored': name}


@app.get('/files/{name}')
async def link(name: str, fs: FS) -> dict[str, str]:
    # Hand back a short-lived link instead of proxying the bytes.
    return {'url': await fs.url(f'/uploads/{name}', expires_in=600)}

Run it:

uv run --with "fastapi[standard]" fastapi dev samples/recipes/fastapi_di.py

Then POST /files/report.csv with a file body, and GET /files/report.csv to get a link back.

Testing the routes

Because the session is a dependency, a test overrides get_fs to return a Storix(MemoryBackend()). The routes run against in-memory storage with no disk, no cloud, and no mocking.

This example is a real, syntax-checked file in the repository (samples/recipes/fastapi_di.py), embedded here so the page never drifts from working code.