EnglishEnglish العربيةالعربية PortuguêsPortuguês Japanese日本語
404 Not Found

404 Not Found

nginx

Upload e Download de Arquivos — Arquivos Grandes e Importações em Lote

Fazer upload de arquivos é como receber um pacote—pacotes pequenos são assinados imediatamente (memória), enquanto grandes carregamentos precisam ser descarregados em lotes (escrita via streaming); fazer download é como enviar um pacote—pacotes únicos (FileResponse), e carregamentos em massa via esteira (StreamingResponse).

1. O Que Você Vai Aprender

2. A História Real da Alice

(1) Dor: Erro de falta de memória ao importar um CSV com um milhão de linhas

Bob faz upload de um arquivo CSV contendo um milhão de linhas de dados de preços no PriceTracker todos os meses. A implementação anterior lia o arquivo inteiro na memória antes de fazer o parsing, e um arquivo CSV de 1GB causava o travamento do processo Python por erro de OOM (Out of Memory). Para piorar, o CSV continha dados inválidos (preços negativos, moedas inválidas), e após a importação falhar pela metade, o banco de dados ficava em um estado inconsistente.

(2) Solução Usando Upload via Streaming e Processamento Assíncrono

O UploadFile do FastAPI salva arquivos como arquivos temporários por padrão (que não consomem memória). Combinado com Celery para processamento assíncrono de milhões de linhas de dados, Pydantic para validação linha a linha para pular dados inválidos, e inserções em lote transacionais para garantir consistência.

(3) Resultado

A importação de CSV de 1GB passou de travar por erro de OOM para rodar de forma estável, com o uso de memória caindo de 2GB para 50MB. Um Celery Worker processa um milhão de linhas em cerca de 10 minutos, e Bob pode acompanhar o progresso em tempo real via o endpoint de status da tarefa.

3. Básico sobre UploadFile

(1) UploadFile vs. bytes

(1) ▶ Exemplo: Upload de Arquivo Simples

PYTHON 
from fastapi import FastAPI, UploadFile, File, HTTPException

app = FastAPI()

@app.post("/upload/single")
async def upload_single(file: UploadFile = File(...)):
    # UploadFile: arquivo armazenado como arquivo temporário, NÃO na memória
    content = await file.read()
    return {
        "filename": file.filename,
        "size": len(content),
        "content_type": file.content_type,
    }

Saída:

TEXT 
# Função definida com sucesso
Método Uso de Memória Tamanho de Arquivo Adequado API
bytes Carregar tudo na memória < 2MB file: bytes = File()
UploadFile Arquivos temporários (streaming) Ilimitado file: UploadFile = File()

(2) ▶ Exemplo: Upload de Múltiplos Arquivos

PYTHON 
@app.post("/upload/multiple")
async def upload_multiple(files: list[UploadFile] = File(...)):
    results = []
    for file in files:
        content = await file.read()
        results.append({
            "filename": file.filename,
            "size": len(content),
        })
    return {"uploaded": len(results), "files": results}

Saída:

TEXT 
# Função definida com sucesso

4. Processamento via Streaming de Arquivos Grandes

(1) Leitura Bloco a Bloco e Escrita via Streaming

(1) ▶ Exemplo: Streaming de Arquivos Grandes

PYTHON 
import shutil
from pathlib import Path
from fastapi import FastAPI, UploadFile, File

app = FastAPI()
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)

@app.post("/upload/large")
async def upload_large_file(file: UploadFile = File(...)):
    # Fazer stream do arquivo para o disco - nunca carrega o arquivo inteiro na memória
    dest = UPLOAD_DIR / file.filename
    with open(dest, "wb") as buffer:
        # Copiar em blocos (buffer padrão de 64KB)
        shutil.copyfileobj(file.file, buffer)
    
    file_size = dest.stat().st_size
    return {
        "filename": file.filename,
        "size_bytes": file_size,
        "saved_to": str(dest),
    }

Saída:

TEXT 
# Função definida com sucesso

(2) ▶ Exemplo: Personalizando o tamanho do bloco

PYTHON 
CHUNK_SIZE = 1024 * 1024  # Blocos de 1MB

@app.post("/upload/chunked")
async def upload_chunked(file: UploadFile = File(...)):
    dest = UPLOAD_DIR / file.filename
    bytes_written = 0
    with open(dest, "wb") as buffer:
        while chunk := await file.read(CHUNK_SIZE):
            buffer.write(chunk)
            bytes_written += len(chunk)
    return {"filename": file.filename, "bytes_written": bytes_written}

Saída:

TEXT 
# Função definida com sucesso

5. Importação em Lote CSV/Excel

(1) Fluxo de Upload de Arquivo Grande + Processamento Assíncrono

100% 
sequenceDiagram
    participant Bob as Bob Frontend
    participant API as FastAPI
    participant Temp as Arquivo Temporário
    participant Celery as Celery Worker
    participant DB as PostgreSQL

    Bob->>API: POST /import/csv (UploadFile)
    API->>Temp: Salvar em arquivo temporário
    API-->>Bob: 202 Accepted + task_id
    API->>Celery: Disparar tarefa de parsing
    Celery->>Temp: Ler CSV em blocos
    Celery->>Celery: Validar cada linha (Pydantic)
    Celery->>DB: Inserir em lote as linhas válidas
    Celery-->>Celery: Relatar progresso
    Bob->>API: GET /tasks/{task_id}
    API-->>Bob: Progresso: 75%
    Celery-->>Celery: Tarefa concluída
    Bob->>API: GET /tasks/{task_id}
    API-->>Bob: Status: SUCCESS, importados: 950000

(1) ▶ Exemplo: Endpoint de Upload CSV + Gatilho de Tarefa Celery

PYTHON 
import csv
import io
from fastapi import FastAPI, UploadFile, File, Depends
from app.tasks import import_prices_from_csv

app = FastAPI()
UPLOAD_DIR = Path("uploads")

@app.post("/api/v1/import/csv")
async def import_csv(
    file: UploadFile = File(..., description="Arquivo CSV com dados de preços"),
    user=Depends(require_subscription("pro")),
):
    if not file.filename.endswith(".csv"):
        raise HTTPException(status_code=400, detail="Apenas arquivos CSV são aceitos")
    
    # Salvar arquivo enviado
    dest = UPLOAD_DIR / f"{uuid4()}.csv"
    with open(dest, "wb") as buffer:
        shutil.copyfileobj(file.file, buffer)
    
    # Disparar tarefa Celery para processamento assíncrono
    task = import_prices_from_csv.delay(str(dest), user_id=user.id)
    
    return {
        "task_id": task.id,
        "filename": file.filename,
        "status": "processing",
        "message": "Arquivo enviado. Verifique o status da tarefa para o progresso.",
    }

Saída:

TEXT 
# Função definida com sucesso

(2) ▶ Exemplo: Tarefa Celery que faz parsing de um arquivo CSV linha a linha

PYTHON 
# app/tasks/import_tasks.py
from app.core.celery_app import celery_app
from pydantic import BaseModel, Field, field_validator
import csv

class PriceRow(BaseModel):
    product_id: int = Field(gt=0)
    price: float = Field(gt=0)
    currency: str = Field(default="USD", pattern=r"^[A-Z]{3}$")
    source: str = Field(max_length=100, default="csv_import")

    @field_validator("price")
    @classmethod
    def round_price(cls, v: float) -> float:
        return round(v, 2)

@celery_app.task(bind=True)
def import_prices_from_csv(self, file_path: str, user_id: int):
    valid_rows = []
    invalid_rows = []
    total_rows = 0
    
    with open(file_path, "r") as f:
        reader = csv.DictReader(f)
        for row in reader:
            total_rows += 1
            try:
                validated = PriceRow(**row)
                valid_rows.append(validated.model_dump())
            except Exception as e:
                invalid_rows.append({"row": total_rows, "error": str(e)})
            
            # Relatar progresso a cada 10000 linhas
            if total_rows % 10000 == 0:
                self.update_state(
                    state="PROGRESS",
                    meta={"current": total_rows, "valid": len(valid_rows), "invalid": len(invalid_rows)},
                )
    
    # Inserir em lote as linhas válidas
    batch_insert_prices(valid_rows, batch_size=5000)
    
    # Limpar arquivo temporário
    Path(file_path).unlink(missing_ok=True)
    
    return {
        "total": total_rows,
        "imported": len(valid_rows),
        "skipped": len(invalid_rows),
    }

Saída:

TEXT 
# Função definida com sucesso

(2) Matriz de Processamento de Formato de Arquivo

Formato Biblioteca de Parsing Vantagens Desvantagens
CSV CSV / pandas Leve, via streaming Sem tipo ou problemas de codificação
XLSX openpyxl / pandas Tipado, múltiplas abas Alto uso de memória
JSON json / orjson Estruturado, compatível com Pydantic Tamanho de arquivo grande
Parquet pyarrow Armazenamento colunar, alta taxa de compressão Requer bibliotecas adicionais

6. Download de Arquivos

(1) FileResponse e StreamingResponse

(1) ▶ Exemplo: FileResponse—Download de um Arquivo

PYTHON 
from fastapi import FastAPI
from fastapi.responses import FileResponse
from pathlib import Path

app = FastAPI()

@app.get("/download/prices/csv")
async def download_prices_csv(
    user=Depends(require_subscription("pro")),
):
    # Gerar arquivo CSV (ou usar pré-gerado)
    file_path = Path("exports/prices.csv")
    return FileResponse(
        path=file_path,
        filename="price_data.csv",
        media_type="text/csv",
    )

Saída:

TEXT 
# Função definida com sucesso

(2) ▶ Exemplo: StreamingResponse—Geração de CSV via Streaming

PYTHON 
from fastapi.responses import StreamingResponse
import csv
import io
from app.core.deps import get_db

@app.get("/api/v1/export/prices")
async def export_prices(
    category: str | None = None,
    db: AsyncSession = Depends(get_db),
    user=Depends(require_subscription("pro")),
):
    async def generate_csv():
        output = io.StringIO()
        writer = csv.writer(output)
        writer.writerow(["product_id", "name", "price", "currency", "recorded_at"])
        yield output.getvalue()
        output.seek(0)
        output.truncate(0)
        
        # Fazer stream das linhas em lotes
        offset = 0
        batch_size = 5000
        while True:
            rows = await fetch_price_batch(db, category, offset, batch_size)
            if not rows:
                break
            for row in rows:
                writer.writerow([
                    row.product_id, row.name,
                    row.price, row.currency, row.recorded_at,
                ])
                yield output.getvalue()
                output.seek(0)
                output.truncate(0)
            offset += batch_size
    
    return StreamingResponse(
        generate_csv(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=prices.csv"},
    )

Saída:

TEXT 
# Função definida com sucesso
Tipo de Resposta Caso de Uso Uso de Memória
FileResponse Arquivos existentes Baixo (streaming em nível de SO)
StreamingResponse Geração Dinâmica Muito Baixo (Gerado Linha a Linha)

❓ Perguntas Frequentes

P Quando os arquivos temporários do UploadFile são excluídos?
R Eles são excluídos automaticamente após o término da requisição. Se você precisar retê-los, copie-os para um diretório persistente durante o processo de upload.
P Existe um limite de tamanho de arquivo para uploads?
R O FastAPI não tem limites impostos pelo framework, mas o Uvicorn impõe um limite padrão no tamanho do corpo da requisição. Em ambiente de produção, isso é controlado pelo client_max_body_size do Nginx.
P O que devo fazer se algumas linhas forem inválidas durante uma importação CSV?
R A abordagem depende das suas necessidades de negócio—modo estrito (reverter todas as linhas inválidas) ou modo tolerante (pular linhas inválidas e importar as válidas). O PriceTracker usa o modo tolerante e retorna o número de linhas importadas e o número de linhas ignoradas.
P Como lidar com arquivos Excel?
R Use read_excel() do pandas para fazer o parsing, mas arquivos XLSX não podem ser lidos via streaming (devem ser carregados integralmente). Para arquivos com milhões de linhas, é recomendado convertê-los para o formato CSV antes do upload.
P O gerador para StreamingResponse precisa ser async?
R Não necessariamente; um gerador síncrono também é aceitável. No entanto, um gerador async não bloqueia o event loop, por isso é recomendado.
P Como posso restringir os tipos de arquivo que podem ser enviados?
R Verifique file.content_type (por exemplo, text/csv) e a extensão do arquivo. Note que o content_type pode ser falsificado; use isso apenas para dicas no frontend—parsing e validação no lado do servidor ainda são necessários.

📖 Resumo

📝 Exercícios

  1. Problema Básico (Dificuldade ⭐): Implemente um endpoint de upload CSV que aceita UploadFile, lê o conteúdo e retorna o número de linhas e nomes das colunas. Dica: file: UploadFile = File(...) + csv.DictReader(io.StringIO(content))
  2. Exercício Avançado (Dificuldade ⭐⭐): Implemente um fluxo de upload CSV → parsing assíncrono Celery: Salve o arquivo enviado em um diretório temporário, dispare uma tarefa Celery, retorne um task_id, e use outro endpoint para consultar o status da tarefa e resultados da importação. Dica: shutil.copyfileobj + import_prices.delay(file_path)
  3. Desafio (Dificuldade: ⭐⭐⭐): Implemente um endpoint de exportação com StreamingResponse: Faça stream dos dados de preços do banco de dados, gere e retorne dados CSV linha a linha, mantenha o uso de memória abaixo de 1 MB, e suporte filtragem por categoria de produto. Dica: async def generate() + yield + StreamingResponse(generate(), media_type="text/csv")

---|

Web-Tutorial.com

Equipe Técnica Web-Tutorial

Uma plataforma de tutoriais mantida por diversos desenvolvedores. Cada tutorial é escrito e revisado por profissionais da área correspondente. Trabalhamos para manter nosso conteúdo preciso e confiável — se encontrar algum problema, avise-nos.

100%