Uploads — disco local + S3 / MinIO

UploadUtils escolhe o backend uma vez no construtor: passe uma pasta para gravar em disco local, ou um AsyncMinIOClient para gravar num bucket S3/MinIO. O resto do código de upload é idêntico nos dois casos. Requer o extra [upload] (e [minio] quando usar MinIO).

Mudança em v0.41.0 (breaking)

O backend agora vem no construtor — o antigo save(file, storage=...) por chamada foi removido. save() devolve a key de storage (relativa), e delete() virou async. Veja a migração no fim.

Validação fica no UploadUtils

Tamanho, extensão, MIME, magic bytes e content_validator são validados no UploadUtils antes de qualquer byte ir pro backend — o storage só recebe dados já validados.

Disco local

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]:
    """Valida e grava em disco; devolve a key (relativa ao base dir)."""
    key = await uploads.save(file)
    return {"key": str(key)}

MinIO / S3

Passe o AsyncMinIOClient direto — nada mais muda:

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."""
    # idêntico ao caso local:
    key = await uploads.save(file, filename="logo.png")   # grava no bucket


asyncio.run(main())

Centralize em resources.py

Construa o uploads (e o minio) uma vez em src/api/dependencies/resources.py e injete via Depends(get_uploads), em vez de instanciar por request. O get_uploads é glue do seu projeto (não vem do SDK) — um provider que devolve a instância única:

# 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

Restringir extensões (allowlist)

Passe allowed_extensions no construtor com o conjunto de extensões que você aceita. Tudo fora da lista é rejeitado com HTTP 415 (InvalidFileTypeException) antes de qualquer byte ser lido — então um .zip malicioso nunca chega ao backend nem ocupa memória:

from tempest_fastapi_sdk import UploadUtils

# Só modelos ONNX — qualquer outra extensão é bloqueada.
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]:
    """Aceita só .onnx / .ort; um .zip levanta 415 aqui dentro do save()."""
    key = await uploads.save(file)   # file.zip -> InvalidFileTypeException (415)
    return {"key": str(key)}

Ponto e case são normalizados

{".onnx", ".ort"}, {"onnx", "ort"} e {".ONNX"} são equivalentes — o UploadUtils tira o ponto inicial e baixa pra minúsculo. A extensão vem de Path(file.filename).suffix, então modelo.ONNX passa e pacote.zip não.

Extensão não é o conteúdo

Conferir extensão impede o engano honesto e o .zip óbvio, mas o nome do arquivo é controlado pelo cliente. Pra formatos com assinatura conhecida (imagens, PDF) ligue verify_magic_bytes=True + allowed_mimetypes={...} pra casar os bytes reais contra a allowlist. Formatos binários sem assinatura no sniff_mime (como .onnx / .ort) devem manter verify_magic_bytes=False (o default) — senão o sniff não reconhece a assinatura e rejeita tudo. Pra esses, valide o conteúdo com um content_validator=... no save().

Via settings (.env)

Quando preferir configurar por ambiente, o UploadSettings já expõe UPLOAD_ALLOWED_EXTENSIONS (e 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,
)

Alternar por settings

Escolha o argumento do construtor conforme uma flag do seu Settings — não precisa de backend pluggável manual:

# 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 é um campo do seu Settings; o SDK só carrega UPLOAD_DIR / UPLOAD_MAX_SIZE_BYTES / UPLOAD_ALLOWED_EXTENSIONS / UPLOAD_ALLOWED_MIMETYPES via UploadSettings.)

Operações comuns

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())

Trocar um arquivo (avatar, anexo) — replace

O caso clássico: o usuário manda uma foto de perfil nova e você quer gravar a nova e apagar a antiga. Em vez de fazer save + delete na mão (e arriscar apagar pelo backend errado), 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 é o que está salvo hoje no model (pode ser None no 1º upload)
    new_key = await uploads.replace(
        user.profile_picture, file, filename=f"{user.id}.jpg"
    )
    user.profile_picture = str(new_key)


asyncio.run(main())

A ordem importa — e o replace acerta pra você

O replace grava a nova primeiro e só então apaga a antiga. Se a validação reprovar o arquivo novo (extensão/MIME/tamanho), a antiga fica intacta — você nunca fica sem imagem nenhuma. Passe old_key=None no primeiro upload (não há nada pra apagar) e o método só salva. Tudo passa pelo mesmo backend configurado (local ou MinIO), evitando o erro de salvar num e apagar no outro.

Para baixar o que foi enviado (local ou MinIO), use o DownloadUtils — ele aceita o mesmo backend no construtor.

Streaming direto pro backend — write_stream e UploadResult

save() é a porta de entrada pra UploadFile do FastAPI e devolve a key. Quando os bytes não vêm de um formulário — proxy de outro serviço, resultado de um job, arquivo gerado na hora — fale com o backend direto: write_stream consome um AsyncIterator[bytes] sem buffer do arquivo inteiro na memória, e devolve um 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,
    )

Campo	Tipo	Conteúdo
key	str	Identificador canônico — caminho relativo no local, key S3 no MinIO
size	int	Bytes gravados
path	Path ou None	Caminho em disco, só quando o backend escreve em filesystem
url	str ou None	URL de download (presigned ou estática), quando o backend sabe gerar

max_size_bytes= corta o stream ao cruzar o teto (FileTooLargeException), e validator= inspeciona os primeiros bytes — as duas checagens rodam enquanto grava, sem esperar o arquivo terminar.

Quando usar presigned PUT direto

Pra arquivos > 50 MB, evite buffer em memória — mande o cliente fazer PUT direto no MinIO via URL presigned. Veja Storage MinIO/S3.

Migração de < v0.41.0

• UploadUtils("./dir") continua igual (disco local).
• UploadUtils("./tmp") + save(file, storage=MinIOUploadStorage(client)) → vira UploadUtils(client) + save(file).
• save() agora devolve a key (relativa), não um caminho absoluto — guarde a key e use DownloadUtils.download(key) pra servir.
• utils.delete(path) (sync) → await utils.delete(key) (async).

Recap

• UploadUtils escolhe o backend uma vez, no construtor: uma pasta grava em disco, um AsyncMinIOClient grava no bucket. O resto do código de upload não muda.
• allowed_extensions é allowlist, não denylist — e o UploadSettings deixa configurar por ambiente.
• save() recebe o UploadFile do FastAPI e devolve a key; é a key que você guarda no banco, não o caminho.
• replace cobre o caso do avatar trocado sem deixar órfão, e write_stream evita carregar arquivo grande na memória.
• Acima de ~50 MB, o caminho é presigned PUT: o cliente manda direto para o bucket, e o seu processo não vira gargalo.