Ir para o conteúdo

Object storage — MinIO / S3

AsyncMinIOClient é uma fachada async sobre o pacote oficial minio. Cobre o que serviço FastAPI típico precisa: bucket (ensure/exists/list/remove), object I/O (put/get/stream/stat/list/remove/copy) e presigned URLs (GET/PUT). Operações avançadas (versioning, lifecycle XML, SSE-KMS, multipart fine-tuning) ficam acessíveis via atributo .client.

Por que esse wrapper existe

minio-py é síncrono. Chamar client.put_object(...) direto dentro de uma rota FastAPI bloqueia o event loop durante o upload inteiro. O wrapper envolve cada chamada em asyncio.to_thread, então o loop continua respondendo enquanto a operação roda no executor.

Instalação

pip install "tempest-fastapi-sdk[minio]"
# ou:
uv add "tempest-fastapi-sdk[minio]"

O pacote minio é lazy-loaded — só carrega quando AsyncMinIOClient é instanciado. Projetos sem storage não precisam do extra.

Configuração via settings mixin

from tempest_fastapi_sdk import (
    BaseAppSettings,
    MinIOSettings,
    ServerSettings,
)


class Settings(
    ServerSettings,
    MinIOSettings,
    BaseAppSettings,
):
    """Service settings — herda MinIO defaults."""

.env:

MINIO_ENDPOINT=minio.internal:9000
MINIO_ACCESS_KEY=...
MINIO_SECRET_KEY=...
MINIO_SECURE=true
MINIO_REGION=us-east-1
MINIO_DEFAULT_BUCKET=uploads

Wiring no create_app()

from contextlib import asynccontextmanager
from collections.abc import AsyncGenerator

from fastapi import FastAPI
from tempest_fastapi_sdk import AsyncMinIOClient

from src.core.settings import settings


# settings.minio_kwargs() mapeia MINIO_* -> endpoint/access_key/secret_key/
# default_bucket/secure/region, então não precisa repetir campo a campo.
storage = AsyncMinIOClient(**settings.minio_kwargs())


@asynccontextmanager
async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
    """Garante que o bucket padrão existe antes de servir tráfego."""
    await storage.ensure_bucket()
    yield


def create_app() -> FastAPI:
    """Build the configured FastAPI instance."""
    return FastAPI(lifespan=lifespan)

Receitas

Upload de UploadFile (FastAPI)

from fastapi import APIRouter, UploadFile

from src.api.app import storage

router = APIRouter()


@router.post("/files")
async def upload_file(file: UploadFile) -> dict[str, str]:
    """Persiste o arquivo recebido no bucket padrão."""
    body = await file.read()
    etag = await storage.put_object(
        file.filename or "unnamed",
        body,
        content_type=file.content_type or "application/octet-stream",
        metadata={"original-name": file.filename or ""},
    )
    return {"key": file.filename or "unnamed", "etag": etag}

Streaming de download

Atalho: download_response (ou DownloadUtils)

O AsyncMinIOClient.download_response(key, ...) já faz stat + stream + Content-Disposition/Type/Length numa chamada só — e o DownloadUtils(minio) embrulha isso. O exemplo manual abaixo é só pra mostrar as peças.

Use para arquivos grandes — chunk-a-chunk evita carregar tudo em memória:

from fastapi import APIRouter
from starlette.responses import Response

from src.api.app import storage

router = APIRouter()


@router.get("/files/{key}")
async def download_file(key: str) -> Response:
    """Stream do objeto no bucket padrão (uma chamada)."""
    return await storage.download_response(key)

Presigned URL — upload direto do browser

Padrão recomendado pra arquivos grandes: o cliente faz PUT direto no MinIO/S3, os bytes não passam pelo FastAPI.

from datetime import timedelta
from uuid import uuid4

from fastapi import APIRouter
from pydantic import BaseModel

from src.api.app import storage

router = APIRouter()


class PresignedUploadResponse(BaseModel):
    key: str
    url: str


@router.post("/uploads/presign")
async def presign_upload() -> PresignedUploadResponse:
    """Devolve URL temporária pro cliente fazer PUT direto."""
    key = f"uploads/{uuid4().hex}"
    url = await storage.presigned_put_url(key, expires=timedelta(minutes=15))
    return PresignedUploadResponse(key=key, url=url)

Cliente JS:

const { key, url } = await fetch("/uploads/presign", { method: "POST" }).then(r => r.json());
await fetch(url, { method: "PUT", body: file });

Presigned URL — download temporário

Para servir arquivos privados sem rotear bytes pela API:

from datetime import timedelta

from fastapi import APIRouter

from src.api.app import storage

router = APIRouter()


@router.get("/files/{key}/url")
async def get_download_url(key: str) -> dict[str, str]:
    """URL de download válida por 1 hora."""
    url = await storage.presigned_get_url(key, expires=timedelta(hours=1))
    return {"url": url}

Endpoint público separado para presigned URLs (v0.88.0+)

Cenário comum em produção: o backend fala com o MinIO por uma rede privada rápida (servus-storage:9000, sem TLS), mas o browser não alcança esse host — precisa de um host público com HTTPS. Se você assinar a presigned URL com o endpoint interno, o link vem com servus-storage:9000 e o navegador não abre.

Solução: MINIO_PUBLIC_ENDPOINT. As presigned URLs (presigned_get_url / presigned_put_url) passam a ser assinadas contra o host público, enquanto todas as operações servidor→MinIO continuam no endpoint interno.

# .env
MINIO_ENDPOINT=servus-storage:9000            # rede interna Docker (ops)
MINIO_SECURE=false
MINIO_PUBLIC_ENDPOINT=https://storage.example.com   # browser (presigned)
# MINIO_PUBLIC_SECURE=true                     # opcional; https:// já implica true

Por que dois clients e não um replace de host

A presigned URL é assinada (SigV4) incluindo o header Host. Trocar o host depois de assinar invalida a assinatura. Por isso o SDK mantém um segundo minio.Minio (mesmas credenciais) só para assinar contra o host público — o AsyncMinIOClient.client interno segue fazendo put/get/stat/ensure_bucket pela rede privada.

Sem MINIO_PUBLIC_ENDPOINT

Comportamento inalterado: presigned URLs são assinadas com MINIO_ENDPOINT (modo endpoint único). O split é 100% opt-in.

O proxy do host público precisa rotear para a API S3 do MinIO (porta 9000) com TLS e repassar o Host correto (a assinatura valida o host).

Operações em lote — presign / upload / download (v0.133.0+)

Endpoints de listagem normalmente precisam resolver uma chave por linha — uma página de perfis, cada um com sua foto. Fazer isso num for com await presigned_get_url(...) serializa os N hops de thread (cada chamada do minio roda em asyncio.to_thread). Os três métodos batch disparam o fan-out de uma vez, com um teto de concorrência:

  • presigned_get_urls(keys) → dict[str, str] (chave → URL)
  • put_objects(items) → dict[str, str] (chave → ETag)
  • get_objects_bytes(keys) → dict[str, bytes] (chave → payload)
from fastapi import APIRouter

from src.api.app import storage

router = APIRouter()


@router.post("/files/urls")
async def sign_many(keys: list[str]) -> dict[str, str]:
    """Assina uma página de chaves de uma vez, em vez de uma por request."""
    return await storage.presigned_get_urls(keys)

Chaves duplicadas são deduplicadas (cada objeto é assinado/baixado uma vez), e o retorno é um dict — faça o lookup por chave com result.get(row.key).

No serviço: file_urls para páginas

Se o serviço usa StoredFileServiceMixin, prefira file_urls([...]) — o par em lote de file_url. Ele descarta chaves None/vazias e devolve dict, ideal pra montar uma página de respostas:

users = [...]  # linhas da página
urls = await user_service.file_urls([u.profile_picture for u in users])
for user in users:
    user.profile_picture_url = urls.get(user.profile_picture)  # None se vazia

Semântica fail-fast

Os três métodos são fail-fast: a primeira falha aborta o lote e propaga (asyncio.gather padrão) — mesmo comportamento de rodar as operações uma a uma. Precisa tolerar falha parcial? Rode os itens individualmente e trate cada exceção.

Teto de concorrência (max_concurrency, padrão 16)

Cada operação vai pra um thread do executor default. Agendar milhares de uma vez satura o pool e estoura memória. Um asyncio.Semaphore limita quantas rodam ao mesmo tempo, preservando a ordem. Ajuste via max_concurrency= (mínimo 1; 0 ou negativo levanta ValueError).

Upload em lote usa PutObjectItem, que espelha os argumentos por-objeto de put_object (content-type, metadata, length para streams):

import asyncio

from tempest_fastapi_sdk import AsyncMinIOClient, PutObjectItem

from src.core.settings import settings

storage = AsyncMinIOClient(**settings.minio_kwargs())

# No seu código estes vêm do disco (`Path(...).read_bytes()`) ou do upload.
thumb_a = b"\xff\xd8\xff\xdb bytes do primeiro JPEG"
thumb_b = b"\xff\xd8\xff\xdb bytes do segundo JPEG"


async def main() -> None:
    """Run this example."""
    etags = await storage.put_objects(
        [
            PutObjectItem(key="thumbs/a.jpg", data=thumb_a, content_type="image/jpeg"),
            PutObjectItem(key="thumbs/b.jpg", data=thumb_b, content_type="image/jpeg"),
        ]
    )


asyncio.run(main())

get_objects_bytes carrega tudo em memória

Como get_object_bytes, o lote é para objetos pequenos — cada payload vira bytes na RAM. Para arquivos grandes, faça streaming individual com stream_object.

Listar objetos por prefixo

from fastapi import APIRouter

from src.api.app import storage

router = APIRouter()


@router.get("/files")
async def list_files(prefix: str = "") -> list[str]:
    """Lista chaves no bucket padrão sob ``prefix``."""
    return await storage.list_objects(prefix)

list_objects devolve [] quando nada bate — em linha com a convenção do SDK ("nenhum match não é erro").

Copiar / mover

import asyncio

from tempest_fastapi_sdk import AsyncMinIOClient

from src.core.settings import settings

storage = AsyncMinIOClient(**settings.minio_kwargs())


async def main() -> None:
    """Run this example."""
    await storage.copy_object("uploads/draft-1", "uploads/final-1")
    await storage.remove_object("uploads/draft-1")


asyncio.run(main())

Quando NÃO usar AsyncMinIOClient

  • Quando você precisa de operações fora das listadas (SSE-KMS, ACLs S3 v2, bucket replication). Use storage.client.<método> direto — minio-py continua acessível.
  • Para uploads gigantes (> 5 GiB) com retomada — minio-py faz multipart automático mas não suporta tus ou resume. Considere tus.io separadamente.

Recap

  • AsyncMinIOClient é fachada async sobre o pacote oficial minio, no extra [minio]: bucket, objeto e URL pré-assinada, que é o que um serviço FastAPI costuma precisar.
  • A configuração vem do settings mixin, e o cliente é montado no lifespan do create_app() — uma instância por processo, não uma por request.
  • URL pré-assinada é a forma de entregar arquivo privado sem passar os bytes pelo seu processo.
  • Operação fora da fachada não é bloqueada: chame storage.client.<método> e use o minio-py direto, em vez de esperar que a fachada cresça.
  • Para alternar disco local e MinIO por configuração, o backend pluggável de upload é o caminho — a fachada é para quem já decidiu usar MinIO.

Próximos passos

  • O backend pluggable de upload MinIOUploadStorage está disponível desde a v0.24.0 — para o pipeline que alterna disco local ↔ MinIO/S3 via flag de settings, veja a receita de uploads.