Uploads — local disk + S3 / MinIO

UploadUtils picks the backend once at construction: pass a folder to store on local disk, or an AsyncMinIOClient to store in an S3/MinIO bucket. The rest of the upload code is identical either way. Requires the [upload] extra (and [minio] when using MinIO).

Change in v0.41.0 (breaking)

The backend now comes from the constructor — the old per-call save(file, storage=...) was removed. save() returns the storage key (relative), and delete() is now async. See the migration at the end.

Validation stays in UploadUtils

Size, extension, MIME, magic bytes, and content_validator are checked in UploadUtils before any byte reaches the backend — the storage only ever receives validated data.

Local disk

from fastapi import APIRouter, UploadFile

from tempest_fastapi_sdk import UploadUtils

router = APIRouter()
uploads = UploadUtils("var/uploads", max_size_bytes=10 * 1024 * 1024)


@router.post("/files")
async def upload(file: UploadFile) -> dict[str, str]:
    """Validate and write to disk; return the key (relative to base dir)."""
    key = await uploads.save(file)
    return {"key": str(key)}

MinIO / S3

Pass the AsyncMinIOClient directly — nothing else changes:

import asyncio

from fastapi import UploadFile

from tempest_fastapi_sdk import AsyncMinIOClient, UploadUtils

from src.core.settings import settings

file: UploadFile = ...  # comes from the endpoint signature


minio = AsyncMinIOClient(**settings.minio_kwargs())
uploads = UploadUtils(minio, max_size_bytes=10 * 1024 * 1024)


async def main() -> None:
    """Run this example."""
    # identical to the local case:
    key = await uploads.save(file, filename="logo.png")   # writes to the bucket


asyncio.run(main())

Centralize in resources.py

Build uploads (and minio) once in src/api/dependencies/resources.py and inject via Depends(get_uploads), instead of instantiating per request. get_uploads is your project's glue (not from the SDK) — a provider that returns the single instance:

# src/api/dependencies/resources.py
from tempest_fastapi_sdk import UploadUtils

uploads = UploadUtils("var/uploads", max_size_bytes=10 * 1024 * 1024)


def get_uploads() -> UploadUtils:
    """Return the shared UploadUtils instance."""
    return uploads

Restrict extensions (allowlist)

Pass allowed_extensions to the constructor with the set of extensions you accept. Anything outside the list is rejected with HTTP 415 (InvalidFileTypeException) before a single byte is read — so a malicious .zip never reaches the backend nor uses memory:

from tempest_fastapi_sdk import UploadUtils

# ONNX models only — any other extension is blocked.
uploads = UploadUtils(
    "var/models",
    allowed_extensions={".onnx", ".ort"},
    max_size_bytes=200 * 1024 * 1024,
)

from fastapi import APIRouter, UploadFile

from tempest_fastapi_sdk import UploadUtils

uploads = UploadUtils(source="./uploads")
router = APIRouter()


@router.post("/models")
async def upload_model(file: UploadFile) -> dict[str, str]:
    """Accepts only .onnx / .ort; a .zip raises 415 here inside save()."""
    key = await uploads.save(file)   # file.zip -> InvalidFileTypeException (415)
    return {"key": str(key)}

Dot and case are normalized

{".onnx", ".ort"}, {"onnx", "ort"} and {".ONNX"} are equivalent — UploadUtils strips the leading dot and lowercases. The extension comes from Path(file.filename).suffix, so model.ONNX passes and package.zip does not.

Extension is not the content

Checking the extension stops the honest mistake and the obvious .zip, but the filename is client-controlled. For formats with a known signature (images, PDF) turn on verify_magic_bytes=True + allowed_mimetypes={...} to match the real bytes against the allowlist. Binary formats with no signature in sniff_mime (like .onnx / .ort) must keep verify_magic_bytes=False (the default) — otherwise the sniff fails to recognize the signature and rejects everything. For those, validate the content with a content_validator=... on save().

Via settings (.env)

To configure per environment, UploadSettings already exposes UPLOAD_ALLOWED_EXTENSIONS (and UPLOAD_ALLOWED_MIMETYPES):

# .env
UPLOAD_ALLOWED_EXTENSIONS=[".onnx", ".ort"]
UPLOAD_MAX_SIZE_BYTES=209715200

from tempest_fastapi_sdk import UploadUtils

from src.core.settings import settings


uploads = UploadUtils(
    settings.UPLOAD_DIR,
    allowed_extensions=settings.UPLOAD_ALLOWED_EXTENSIONS,
    max_size_bytes=settings.UPLOAD_MAX_SIZE_BYTES,
)

Switching via settings

Choose the constructor argument from a flag on your Settings — no manual pluggable backend needed:

# src/api/dependencies/resources.py
from tempest_fastapi_sdk import AsyncMinIOClient, UploadUtils

from src.core.settings import settings

if settings.UPLOAD_BACKEND == "minio":
    uploads = UploadUtils(AsyncMinIOClient(**settings.minio_kwargs()))
else:
    uploads = UploadUtils(settings.UPLOAD_DIR)

(UPLOAD_BACKEND is a field on your Settings; the SDK only loads UPLOAD_DIR / UPLOAD_MAX_SIZE_BYTES / UPLOAD_ALLOWED_EXTENSIONS / UPLOAD_ALLOWED_MIMETYPES via UploadSettings.)

Common operations

import asyncio

from fastapi import UploadFile

from tempest_fastapi_sdk import UploadUtils

file: UploadFile = ...  # comes from the endpoint signature
uploads = UploadUtils(source="./uploads")


async def main() -> None:
    """Run this example."""
    key = await uploads.save(file, filename="logo.png")  # -> Path("logo.png")
    removed = await uploads.delete(key)                  # async; True/False


asyncio.run(main())

Swap a file (avatar, attachment) — replace

The classic case: the user uploads a new profile picture and you want to save the new one and delete the old one. Instead of doing save + delete by hand (and risking deleting through the wrong backend), use replace:

import asyncio

from fastapi import UploadFile

from tempest_fastapi_sdk import UploadUtils

from src.db.models import UserModel

file: UploadFile = ...  # comes from the endpoint signature
uploads = UploadUtils(source="./uploads")
user = UserModel(name="Ana", email="ana@example.com")


async def main() -> None:
    """Run this example."""
    # old_key is whatever is stored on the model today (may be None on 1st upload)
    new_key = await uploads.replace(
        user.profile_picture, file, filename=f"{user.id}.jpg"
    )
    user.profile_picture = str(new_key)


asyncio.run(main())

Order matters — and replace gets it right for you

replace saves the new file first, then deletes the old one. If validation rejects the new file (extension/MIME/size), the old one is left intact — you never end up with no image at all. Pass old_key=None on the first upload (nothing to delete) and it just saves. Everything goes through the same configured backend (local or MinIO), avoiding the save-here-delete-there mistake.

To download what was uploaded (local or MinIO), use DownloadUtils — it takes the same backend in its constructor.

Streaming straight to the backend — write_stream and UploadResult

save() is the entry point for FastAPI's UploadFile and returns the key. When the bytes do not come from a form — proxying another service, a job's output, a file generated on the fly — talk to the backend directly: write_stream consumes an AsyncIterator[bytes] without buffering the whole file in memory, and returns an UploadResult:

from collections.abc import AsyncIterator

from tempest_fastapi_sdk import LocalUploadStorage, UploadResult

storage: LocalUploadStorage = LocalUploadStorage("./var/uploads")


async def store_report(chunks: AsyncIterator[bytes]) -> UploadResult:
    """Persist a generated report without buffering it in memory."""
    return await storage.write_stream(
        "reports/2026-07.csv",
        chunks,
        content_type="text/csv",
        max_size_bytes=50 * 1024 * 1024,
    )

Field	Type	Content
key	str	Canonical identifier — relative path locally, S3 key on MinIO
size	int	Bytes written
path	Path or None	On-disk path, only when the backend writes to a filesystem
url	str or None	Download URL (presigned or static), when the backend can mint one

max_size_bytes= aborts the stream once the cap is crossed (FileTooLargeException), and validator= inspects the leading bytes — both checks run while writing, without waiting for the upload to finish.

When to use a direct presigned PUT

For files > 50 MB, skip the in-memory buffer — have the client PUT straight to MinIO via a presigned URL. See Storage MinIO/S3.

Migrating from < v0.41.0

• UploadUtils("./dir") is unchanged (local disk).
• UploadUtils("./tmp") + save(file, storage=MinIOUploadStorage(client)) → becomes UploadUtils(client) + save(file).
• save() now returns the key (relative), not an absolute path — store the key and use DownloadUtils.download(key) to serve it.
• utils.delete(path) (sync) → await utils.delete(key) (async).

Recap

• UploadUtils picks the backend once, in the constructor: a folder writes to disk, an AsyncMinIOClient writes to the bucket. The rest of your upload code does not change.
• allowed_extensions is an allowlist, not a denylist — and UploadSettings lets you configure it per environment.
• save() takes FastAPI's UploadFile and returns the key; the key is what you store in the database, not the path.
• replace covers the swapped avatar without leaving an orphan, and write_stream keeps a large file out of memory.
• Above roughly 50 MB the road is a presigned PUT: the client uploads straight to the bucket, and your process stops being the bottleneck.