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.
Provision once at startup. Create application-owned directories before the app accepts requests. mkdir(..., parents=True) is idempotent across restarts.
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 when the provider supports them. Azure returns a short-lived presigned link. DataUrlLayer keeps the example portable by returning an inline data: URL on local and memory storage.

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 application memory whole.

Note: Reading `UploadFile` in chunks avoids accumulating large payloads in application
memory. `UploadFile` is backed by Starlette's spooled temporary file for large requests.
For raw unbuffered network body streaming without multipart parsing, `request.stream()`
can be passed directly to `fs.echo()`.

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, memory, azure, ...) is configured, not
    # hard-coded. DataUrlLayer backfills url() so it works on any backend.
    fs = get_storage().with_layer_missing(DataUrlLayer)
    app.state.fs = fs
    try:
        # Provision application-owned directories once, before requests arrive.
        await fs.mkdir('/uploads', parents=True)
        yield
    finally:
        await 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


FsDep = Annotated[Storix, Depends(get_fs)]


@app.post('/files/{name}')
async def upload(name: str, file: UploadFile, fs: FsDep) -> dict[str, str]:
    async def chunks():
        while data := await file.read(1024 * 1024):  # 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: FsDep) -> 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 upload and retrieve a link:

curl -F "[email protected]" http://127.0.0.1:8000/files/report.csv
curl http://127.0.0.1:8000/files/report.csv

Large local files

Azure creates a presigned URL without reading the file. The portable DataUrlLayer fallback must read and base64-encode the complete file, so use it for small local or in-memory files, not large downloads.

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.