Uploads resumíveis em FastAPI: S3 multipart, checkpoints e revalidação

1) Quando e por que usar

Uploads resumíveis não são só um luxo para UX: são uma necessidade quando arquivos grandes, redes instáveis ou clientes distribuídos aparecem no seu fluxo. Em serviços que recebem vídeos, datasets científicos ou backups, falhas parciais são regra, não exceção. Tentar reiniciar um upload inteiro a cada falha custa banda, CPU e paciência do usuário.

O S3 multipart upload foi projetado exatamente para isso: dividir um objeto em partes que podem ser enviadas paralelamente e finalizadas no servidor. Com isso você ganha capacidade de recomeçar apenas partes faltantes e paralelizar throughput. Porém, apenas usar multipart por si só não resolve todos os problemas práticos: é preciso checkpoints no cliente/servidor, revalidação dos dados e tratamento de aborts.

Na prática, adotar uploads resumíveis resolve três problemas concretos: reduzir reenvio completo de arquivos, permitir retry/parallelismo e melhorar resiliência ao estado do cliente.

O jeito errado (antipadrões que você já viu)

Um antipadrão comum: oferecer um endpoint que recebe o arquivo por streaming e, ao falhar, o cliente reenvia tudo. Ou ainda, partir para chunking no cliente mas manter o estado apenas na memória do processo do servidor — qualquer deploy ou crash perde o progresso.

Problemas evidentes do jeito errado:

• Reenvio completo do arquivo em falhas — desperdício de banda e custos.
• Incapacidade de retomar após reinicialização do cliente/servidor.
• Risco de duplicação de objetos no S3 ou uploads incompletos aumentando custo.

O jeito certo — visão geral

O padrão correto combina: S3 multipart para transferência resiliente, checkpoints (persistidos no cliente e/ou em um store leve: Redis, DB ou disco) para recomeço, e revalidação no final (checksums/ETag) para garantir integridade. Complementar com políticas de expiração e jobs de limpeza evita acumular uploads incompletos no S3.

Se objetivo é confiabilidade em produção, trate o upload como uma transação distribuída: checkpoint + revalidação + timeout/abort automático.

2) Comparação com alternativas

Antes de implementar, compare as alternativas: upload direto com multipart S3, proxied streaming via FastAPI, protocolos especializados (tus.io), ou usar serviços gerenciados (CloudFront/S3 Signed URLs com gerenciamento client-side).

S3 multipart (prós/contras)

Prós:

• Amplo suporte, alta performance e custo previsível.
• Permite parallel uploads e reenvio de partes individuais.
• Integra bem com presigned URLs (reduz carga no app server).

Contras:

• Complexidade: criar/upload/complete/abort + gerenciamento de partes.
• Necessita de lógica extra para checkpoints e validação.

Proxy via FastAPI (prós/contras)

Fazer upload passando pelo servidor (stream proxied) simplifica controle de acesso, permite inspeção e transformação em tempo real. Mas você penaliza banda do servidor e perde paralelismo nativo do S3. Útil para transformações pequenas ou quando não se pode expor S3 diretamente ao cliente.

Protocolos especializados: tus.io

tus provê resumable uploads com padrões de controle e clients prontos. É excelente se você quer padronização, mas implica mais infra (server/tus server) e integração adicional. Em muitos casos, S3 multipart + presigned URLs + checkpoint é mais simples e suficiente.

3) Implementação progressiva

Vamos revisar uma implementação prática em camadas, do mínimo viável até produção sólida. Em cada etapa eu mostro o jeito errado e o certo.

Etapa 0 — Spike rápido (jeito errado)

Você cria um endpoint /upload que aceita MultipartFile e salva no disco. Se der erro no upload, cliente reenvia tudo. Isso funciona para arquivos pequenos, mas explode em arquivos grandes ou clientes móveis.

Etapa 1 — Multipart presigned URLs (mínimo viável correto)

Crie um endpoint que inicia um multipart upload no S3 e retorna presigned URLs para as partes. O cliente usa essas URLs para enviar partes diretamente para S3 e depois chama um endpoint para completar o upload.

Contexto: o código abaixo cria um multipart upload e gera presigned URLs para cada parte. Ele roda no servidor FastAPI e delega transferência para o cliente.

from fastapi import FastAPI, HTTPException
import boto3
import math

app = FastAPI()
s3 = boto3.client('s3')
BUCKET = 'meu-bucket'

@app.post('/start-upload')
def start_upload(key: str, total_size: int, part_size: int = 5 * 1024 * 1024):
    # Cria multipart upload e gera URLs para cada parte
    response = s3.create_multipart_upload(Bucket=BUCKET, Key=key)
    upload_id = response['UploadId']
    parts = []
    n_parts = math.ceil(total_size / part_size)
    for i in range(1, n_parts + 1):
        url = s3.generate_presigned_url(
            'upload_part',
            Params={'Bucket': BUCKET, 'Key': key, 'UploadId': upload_id, 'PartNumber': i},
            ExpiresIn=3600
        )
        parts.append({'part_number': i, 'url': url})
    return {'upload_id': upload_id, 'parts': parts}

O que acontece: o cliente recebe upload_id e uma lista de URLs para cada parte. Ele faz PUT para essas URLs enviando os bytes das partes. O servidor não manipula o conteúdo, somente gerencia o upload lifecycle.

Etapa 2 — Checkpoints no cliente (jeito certo)

Checkpointismo simples: cliente salva um JSON com upload_id, key e lista de partes já enviadas (os ETags retornados). Em caso de falha ele relê o checkpoint e só reenvia partes faltantes.

Exemplo de cliente Python: o script abaixo lê um arquivo, divide em chunks, guarda checkpoint local e reenvia só partes faltantes.

import os
import json
import requests
from hashlib import md5

CHECKPOINT_FILE = 'upload.checkpoint.json'

def upload_file(file_path, upload_meta):
    # upload_meta: {'upload_id': str, 'parts': [{'part_number':1,'url':...}, ...]}
    file_size = os.path.getsize(file_path)
    checkpoint = {}
    if os.path.exists(CHECKPOINT_FILE):
        checkpoint = json.load(open(CHECKPOINT_FILE))
        if checkpoint.get('file_path') != file_path:
            checkpoint = {}
    uploaded = {p['part_number']: p.get('etag') for p in checkpoint.get('uploaded_parts', [])}

    with open(file_path, 'rb') as f:
        for p in upload_meta['parts']:
            num = p['part_number']
            url = p['url']
            if num in uploaded:
                continue
            start = (num - 1) * 5 * 1024 * 1024
            f.seek(start)
            chunk = f.read(5 * 1024 * 1024)
            resp = requests.put(url, data=chunk)
            resp.raise_for_status()
            etag = resp.headers.get('ETag')
            # save checkpoint
            checkpoint = {'file_path': file_path, 'upload_id': upload_meta['upload_id'], 'uploaded_parts': checkpoint.get('uploaded_parts', [])}
            checkpoint['uploaded_parts'].append({'part_number': num, 'etag': etag})
            json.dump(checkpoint, open(CHECKPOINT_FILE, 'w'))

O que acontece: cada PUT retorna um ETag (geralmente MD5 do segmento para uploads simples). O cliente registra esse ETag e usa depois na chamada de complete. Em falhas, o cliente relê o checkpoint e continua.

Etapa 3 — Complete + revalidação (jeito certo com verificação)

Depois de subir as partes, o cliente chama um endpoint do servidor com upload_id e lista de part_number e ETag. O servidor invoca complete_multipart_upload. É recomendado revalidar: checar que número de partes bate com o esperado, confirmar checksums ou comparar tamanho final esperado.

Exemplo de endpoint que completa e revalida. Aqui assumimos que a lista de partes foi enviada pelo cliente.

from fastapi import Body

@app.post('/complete-upload')
def complete_upload(key: str, upload_id: str, parts: list = Body(...)):
    # parts: [{'PartNumber':1,'ETag':'"..."'}, ...]
    # Validacoes basicas
    if not parts:
        raise HTTPException(status_code=400, detail='parts required')
    multipart = {'Parts': [{'ETag': p['ETag'], 'PartNumber': p['PartNumber']} for p in parts]}
    resp = s3.complete_multipart_upload(Bucket=BUCKET, Key=key, UploadId=upload_id, MultipartUpload=multipart)
    # opcional: validar objeto final
    head = s3.head_object(Bucket=BUCKET, Key=key)
    return {'location': resp.get('Location'), 'size': head['ContentLength']}

O que acontece: o S3 monta o objeto final a partir das partes. A chamada head_object permite validar tamanho; se você precisa de checksum global, calcule no cliente e compare (p.ex. SHA256) com metadados, ou execute verificação por job background baixando e verificando.

4) Integracao com ecossistema real

Em produção você precisará integrar checkpoints, filas (Celery/RQ), armazenamento de metadata (Postgres/Redis), monitoramento e políticas de timeout. Abaixo mostro padrões práticos testados em projetos reais.

Persistência de checkpoints

Preferível: checkpoint server-side persistido (Postgres/Redis). Alternativa leve: checkpoint no cliente e apenas upload_id no servidor. Cada opção tem trade-offs:

• Client-only checkpoint: simples, sem escrita extra no servidor, mas se cliente perde o checkpoint (ex: uninstall) você precisa expirar uploads no S3 automaticamente.
• Server-side checkpoint: você pode mostrar progresso em dashboard, reiniciar uploads e executar políticas de retenção centralizadas.

Exemplo mínimo de schema (Postgres): uploads(id, key, upload_id, total_size, part_size, state, created_at, last_activity). Use índices TTL ou job de limpeza para abortar uploads antigos.

Workers e limpeza

Arquitetura recomendada:

1. API FastAPI gera presigned URLs e grava metadata em DB/Redis.
2. Cliente envia partes e confirma ETags no servidor.
3. Server enfileira um job de complete quando todas as partes chegam.
4. Job de validação pode calcular checksums adicionais ou verificar conteúdo redundante.
5. Job de limpeza quotidiana aborta uploads antigos (s3.abort_multipart_upload).

# exemplo de job de limpeza (pseudo)
from datetime import datetime, timedelta

def cleanup_incomplete_uploads(db_session, s3_client, bucket):
    cutoff = datetime.utcnow() - timedelta(days=1)
    uploads = db_session.query(Upload).filter(Upload.state == 'in_progress', Upload.last_activity < cutoff)
    for u in uploads:
        try:
            s3_client.abort_multipart_upload(Bucket=bucket, Key=u.key, UploadId=u.upload_id)
            u.state = 'aborted'
            db_session.commit()
        except Exception:
            db_session.rollback()

O que acontece: upload incompleto é abortado no S3, evitando custos desnecessários e evitando limbo de partes não finalizadas.

Revalidação pós-complete

Se integridade é crítica, você deve revalidar o objeto final. Duas estratégias:

• Cliente envia checksum (SHA256) do arquivo e servidor compara com cálculo em background (download parcial ou pleno).
• Gera-se manifest com checksums das partes; após complete, job reconstrói hash final a partir das partes para validar sem baixar tudo (quando possível).

Observação: S3 ETag não é SHA256 do objeto quando multipart, por isso não confie no ETag para validação global a menos que arquivo tenha sido upload single-part.

5) Checklist de qualidade para produção

Antes de ir para produção, valide cada item abaixo. Não é uma lista teórica — são pontos que quebram sistemas no dia a dia.

• Checkpoints persistidos: cliente ou servidor com TTL e fallback.
• Timeout e cleanup: policy para abortar uploads inativos.
• Revalidação: checksum global ou validação por job.
• Idempotência: endpoints aceitam replays controlados (start-upload pode retornar id existente).
• Métrica e observabilidade: log de parts, tamanho final, latência, erros por parte.
• Segurança: presigned URLs curtas, validação de key/ACL, policy para restringir buckets.
• Rate limiting: proteger endpoint de start-upload para evitar criação massiva de uploads.
• Política de custos: prever gastos com partes armazenadas temporariamente no S3.

Erros comuns e armadilhas

Lista prática de problemas que já vi em code review:

• Contar com ETag como checksum global em multipart uploads — errado.
• Gerar presigned URLs com ExpiresIn muito longo (1 dia) — aumenta risco se exposto.
• Não validar PartNumber consistente — parte fora de ordem pode confundir complete.
• Guardar checkpoint somente em memória — perde tudo em crash/scale.
• Não tratar aborts: múltiplos uploads incompletos consomem espaço e custos.

Dica: mantenha os presigned URLs curtos (ex: 1 hora) e use um refresh flow simples para renovar se o usuário demorar.

Conclusão prática

Implementar uploads resumíveis em FastAPI com S3 multipart é uma engenharia acessível quando você segmenta o problema: delegue transferência ao S3, gerencie lifecycle e checkpoints, e valide a integridade no final. O pior rival do sistema não é a falha pontual, mas a ausência de timeout/limpeza e de validação.

Resumindo os passos práticos que você deve aplicar hoje:

1. Implemente start-upload que cria upload_id + presigned URLs.
2. Faça o cliente persistir checkpoint local com upload_id e ETags.
3. Server-side: aceite complete com lista de ETags e invoque complete_multipart_upload.
4. Agende job de validação e limpeza para abortar uploads antigos.
5. Monitore métricas e aprenda com erros (etags faltando, completes falhando).

Seguindo esse padrão você tem um sistema escalável, tolerante a falhas e com custo previsível. Em code review, priorize checkpoints persistidos, validações e jobs de limpeza — estes três pontos resolvem a maioria das dores reais.

Se precisar, posso gerar um boilerplate completo de FastAPI + Celery + Postgres para esse fluxo, com scripts cliente prontos e testes de integração que simulem falhas e retomadas.