fast-storages

Django-style, loosely-coupled async file storage for FastAPI.

Pluggable backends behind one stable Storage contract — swap from local filesystem to S3, Azure Blob, PostgreSQL, or your own custom backend without changing a single line of application code.

Features

• One interface, many backends — save, open, delete, exists, size, url work identically across all backends.
• Streaming-first — save() accepts bytes or AsyncIterable[bytes]; open() returns an AsyncIterator[bytes]. Large files never need to be buffered in memory.
• Named storages — configure multiple backends simultaneously (e.g. "default" → S3, "avatars" → local) just like Django's STORAGES.
• FastAPI dependency injection — Depends(get_storage()) gives you a configured Storage in any route handler.
• Env-driven configuration — every backend ships a pydantic-settings *StorageSettings class that reads from environment variables.
• Custom backends — register your own backend with @register_backend("name") or use a dotted import path.

Supported Backends

Installation

# Core (includes local filesystem backend)
pip install fast-storages

# With a specific backend
pip install fast-storages[azure]
pip install fast-storages[s3]
pip install fast-storages[postgresql]         # psycopg driver (recommended)
pip install fast-storages[postgresql-asyncpg]  # asyncpg driver

# Everything
pip install fast-storages[all]

Requires Python 3.10+

Quick Start

1. Set up the StorageManager in your app's lifespan

from contextlib import asynccontextmanager
from pathlib import Path

from fastapi import FastAPI

from fast_storages import StorageManager

@asynccontextmanager
async def lifespan(app: FastAPI):
    manager = StorageManager()

    # Register a "default" storage backed by the local filesystem
    manager.add(
        "default",
        backend="local",
        config={"media_root": "./uploads", "media_url": "/files"},
    )

    app.state.storage_manager = manager
    yield
    await manager.aclose_all()

app = FastAPI(lifespan=lifespan)

2. Use get_storage() in your route handlers

from fastapi import Depends, UploadFile
from fastapi.responses import StreamingResponse

from fast_storages import Storage, UploadFileReader, get_storage, guess_content_type

@app.post("/upload")
async def upload(file: UploadFile, storage: Storage = Depends(get_storage())):
    reader = UploadFileReader(file)
    result = await storage.save(
        file.filename or "unnamed",
        reader,
        content_type=file.content_type or guess_content_type(file.filename or ""),
    )
    return {
        "name": result.name,
        "key": result.key,
        "size": result.size,
        "content_type": result.content_type,
        "url": await storage.url(result.key),
    }

@app.get("/download/{name:path}")
async def download(name: str, storage: Storage = Depends(get_storage())):
    stream = await storage.open(name)
    return StreamingResponse(stream, media_type="application/octet-stream")

@app.delete("/files/{name:path}")
async def delete_file(name: str, storage: Storage = Depends(get_storage())):
    await storage.delete(name)
    return {"deleted": name}

That's it — your app now handles file uploads, downloads, and deletes through the storage layer.

Multiple Named Storages

Register as many backends as you need, each under a unique name:

manager = StorageManager()

manager.add(
    "default",
    backend="azure",
    config={
        "connection_string": "DefaultEndpointsProtocol=https;...",
        "container": "uploads",
    },
)

manager.add(
    "avatars",
    backend="local",
    config={"media_root": "/data/avatars", "media_url": "/avatars"},
)

manager.add(
    "db",
    backend="postgresql",
    config={
        "dsn": "postgresql://user:pass@localhost:5432/mydb",
        "serve_url": "http://localhost:8000/db/download",
        "create_table": True,
    },
)

Then inject the specific storage by name:

@app.post("/upload/avatar")
async def upload_avatar(
    file: UploadFile,
    storage: Storage = Depends(get_storage("avatars")),
):
    reader = UploadFileReader(file)
    result = await storage.save(file.filename, reader)
    return {"url": await storage.url(result.key)}

Configuration

Every backend accepts config as plain kwargs, a dict, or a pydantic-settings *StorageSettings instance. All three are equivalent:

# 1. Plain kwargs via dict
manager.add("default", backend="local", config={"media_root": "/data", "media_url": "/media"})

# 2. Direct construction
from fast_storages.backends.local import LocalStorage
storage = LocalStorage(base_path="/data", base_url="/media")

# 3. Env-driven settings
from fast_storages.backends.local import LocalStorageSettings
settings = LocalStorageSettings()  # reads FASTAPI_STORAGE_LOCAL_BASE_PATH, etc.
manager.add("default", backend="local", config=settings)

Environment Variable Prefixes

Backend Configuration Reference

manager.add("default", backend="local", config={
    "media_root": "./uploads",
    "media_url": "/files",
})

manager.add("default", backend="azure", config={
    "connection_string": "DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...;",
    "container": "uploads",
    "custom_url": "https://cdn.example.com/uploads",
})

manager.add("db", backend="postgresql", config={
    "dsn": "postgresql://user:pass@localhost:5432/mydb",
    "table_name": "storage_files",
    "serve_url": "http://localhost:8000/db/download",
    "create_table": True,
})

from fastapi import FastAPI, Depends
from fastapi.responses import StreamingResponse
from fast_storages import Storage, get_storage

app = FastAPI()

# Matches the path portion of serve_url = "http://localhost:8000/db/download"
@app.get("/db/download/{name:path}")
async def db_download(name: str, storage: Storage = Depends(get_storage("db"))):
    """Stream a file from PostgreSQL Large Object storage."""
    stream = await storage.open(name)
    return StreamingResponse(stream, media_type="application/octet-stream")

Schema management with Alembic:

from sqlalchemy.orm import DeclarativeBase
from fast_storages.backends.postgresql_schema import StorageFileMixin

class Base(DeclarativeBase):
    pass

class StorageFile(StorageFileMixin, Base):
    pass  # table is now tracked by Alembic

manager.add("default", backend="s3", config={
    "bucket": "my-bucket",
    "region_name": "us-east-1",
})

Storage API

Every backend implements the same Storage interface:

class Storage(ABC):
    async def save(name, content, *, content_type=None, upload_to=None, context=None) -> FileMeta
    async def open(name, *, chunk_size=65536) -> AsyncIterator[bytes]
    async def delete(name) -> None
    async def exists(name) -> bool
    async def size(name) -> int
    async def url(name, *, expires_in=None) -> str
    async def aclose() -> None

FileMeta — the save() return object

save() returns a frozen FileMeta dataclass with everything you need about the stored file:

result = await storage.save("photo.png", data, content_type="image/png", upload_to="avatars")
result.name          # "photo.png"
result.key           # "avatars/photo.png"
result.size          # 102400
result.content_type  # "image/png"
result.backend       # "local"

All backends also support the async context manager protocol:

async with LocalStorage(base_path="./uploads") as storage:
    await storage.save("hello.txt", b"Hello, world!")

Upload Path Control (upload_to)

save() supports an upload_to parameter to control where a file is stored:

# String prefix — prepends a directory
result = await storage.save("photo.png", content, upload_to="avatars")
result.key   # "avatars/photo.png"
result.name  # "photo.png"

# Callable — full custom logic
def user_upload_path(name: str, context: dict | None) -> str:
    user_id = context["user_id"]
    return f"users/{user_id}/{name}"

result = await storage.save(
    "photo.png", content,
    upload_to=user_upload_path,
    context={"user_id": 42},
)
result.key  # "users/42/photo.png"

Exception Handling

All backends translate provider-specific errors into a common exception hierarchy — your application code never needs to catch SDK-specific exceptions:

StorageError (base)
├── StorageFileNotFoundError
├── StoragePermissionError
├── StorageUnsupportedOperationError
├── StorageConfigError
└── StorageConnectionError

Map them to HTTP responses in FastAPI:

from fast_storages import (
    StorageError,
    StorageFileNotFoundError,
    StoragePermissionError,
    StorageUnsupportedOperationError,
)

@app.exception_handler(StorageFileNotFoundError)
async def _not_found(request, exc):
    raise HTTPException(status_code=404, detail=str(exc))

@app.exception_handler(StoragePermissionError)
async def _forbidden(request, exc):
    raise HTTPException(status_code=403, detail=str(exc))

@app.exception_handler(StorageUnsupportedOperationError)
async def _not_implemented(request, exc):
    raise HTTPException(status_code=501, detail=str(exc))

@app.exception_handler(StorageError)
async def _storage_error(request, exc):
    raise HTTPException(status_code=502, detail=str(exc))

Writing a Custom Backend

1. Subclass Storage and implement all abstract methods:

from fast_storages import Storage, SaveContent
from fast_storages.base import UploadTo, resolve_upload_name

class MyStorage(Storage):
    backend_name = "my-backend"

    async def save(self, name, content, *, content_type=None, upload_to=None, context=None):
        resolved = resolve_upload_name(name, upload_to, context)
        # ... write content, track total_size ...
        return FileMeta(
            name=name, key=resolved, size=total_size,
            content_type=content_type, backend=self.backend_name,
        )

    async def open(self, name, *, chunk_size=65536):
        # ... return an AsyncIterator[bytes] ...

    async def delete(self, name):
        # ... idempotent delete ...

    async def exists(self, name):
        # ... return bool ...

    async def size(self, name):
        # ... return int ...

    async def url(self, name, *, expires_in=None):
        # ... return str ...

2. Register it (optional — enables use by short name):

from fast_storages import register_backend

register_backend("my-backend")(MyStorage)

3. Or use the dotted import path directly:

manager.add("default", backend="mypackage.backends.MyStorage", config={...})

Utilities

License

MIT

Links

• PyPI: https://pypi.org/project/fast-storages/
• Repository: https://github.com/LaxmanChoudhary/fast-storages
• Changelog: https://github.com/LaxmanChoudhary/fast-storages/releases