Cheatsheet FastAPI
Framework Python moderno e rápido para APIs com type hints
FastAPI
Instalação e Setup
Instalar FastAPI
pip install fastapi uvicorn[standard] # criar main.py uvicorn main:app --reload
fastapi é o framework. uvicorn é o servidor ASGI. --reload recarrega o código automaticamente em desenvolvimento. O standard inclui websockets e httptools.
Configuração com Settings
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
secret_key: str
debug: bool = False
class Config:
env_file = ".env"
settings = Settings()BaseSettings lê variáveis de ambiente automaticamente. env_file carrega de um ficheiro .env. Valores com default são opcionais. Ideal para separar config do código.
Primeira aplicação
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def raiz():
return {"msg": "Olá FastAPI!"}FastAPI() cria a instância da aplicação. @app.get() regista um endpoint GET. O dicionário retornado é serializado para JSON automaticamente. Type hints geram validação automática.
Metadata da API
app = FastAPI(
title="Minha API",
description="API de gestão de itens",
version="1.0.0",
contact={"name": "Equipa Dev"},
)title e description aparecem na documentação. version controla a versão da API. contact adiciona info de contacto. Estes metadados enriquecem o Swagger UI.
Documentação automática
# Swagger UI: # http://localhost:8000/docs # ReDoc: # http://localhost:8000/redoc # OpenAPI JSON: # http://localhost:8000/openapi.json
O FastAPI gera documentação interativa automaticamente. /docs mostra Swagger UI com testes. /redoc mostra ReDoc formatado. /openapi.json é o schema OpenAPI completo.
Tags para organização
tags_metadata = [
{"name": "users", "description": "Operações de utilizadores"},
{"name": "items", "description": "Gestão de itens"},
]
app = FastAPI(openapi_tags=tags_metadata)
@app.get("/users", tags=["users"])
def listar_users(): ...tags agrupam endpoints na documentação. openapi_tags adiciona descrições às tags. Cada endpoint pode ter múltiplas tags. Facilita navegação no Swagger UI.
Estrutura do projecto
app/
main.py
models.py
schemas.py
database.py
dependencies.py
routers/
users.py
items.pymain.py é o ponto de entrada. schemas.py define modelos Pydantic. models.py define tabelas SQLAlchemy. routers/ organiza endpoints por módulo. dependencies.py centraliza dependências.
Ambiente virtual
python -m venv .venv source .venv/bin/activate # Linux/Mac .venv\Scripts\activate # Windows pip install fastapi uvicorn pip freeze > requirements.txt
venv isola dependências do projecto. activate activa o ambiente virtual. pip freeze gera o ficheiro de dependências. Sempre use ambientes virtuais por projecto.
Rotas e Endpoints
Métodos HTTP
@app.get("/itens")
def listar(): ...
@app.post("/itens")
def criar(): ...
@app.put("/itens/{id}")
def atualizar(id: int): ...
@app.delete("/itens/{id}")
def apagar(id: int): ...Cada decorador mapeia um verbo HTTP. GET para leitura, POST para criação. PUT actualiza totalmente, PATCH parcialmente. DELETE remove recursos.
Códigos de status
from fastapi import status
@app.post("/itens", status_code=status.HTTP_201_CREATED)
def criar(): ...
@app.delete("/itens/{id}", status_code=204)
def apagar(id: int): ...
# status.HTTP_404_NOT_FOUND
# status.HTTP_403_FORBIDDENstatus_code define o código HTTP da resposta. 201 para recursos criados. 204 para deleção sem corpo. O módulo status tem constantes nomeadas para todos os códigos.
Redirect e status code
from fastapi.responses import RedirectResponse
@app.get("/antigo")
def antigo():
return RedirectResponse(url="/novo", status_code=301)
@app.get("/itens/{id}", status_code=200)
def get_item(id: int, response: Response):
response.headers["X-Custom"] = "valor"
return {"id": id}RedirectResponse com 301 é redirect permanente. O parâmetro Response permite adicionar headers custom. Headers são úteis para rate-limiting e cache-control.
Path parameters
@app.get("/users/{user_id}")
def get_user(user_id: int):
return {"id": user_id}
# /users/42 → user_id = 42
# /users/abc → erro 422Parâmetros entre {} na rota são capturados. O type-hint int valida e converte automaticamente. Valores inválidos retornam erro 422 com detalhes. Não precisa de parsing manual.
Respostas personalizadas
from fastapi.responses import JSONResponse, RedirectResponse, HTMLResponse
return JSONResponse(content={"erro": "Não encontrado"}, status_code=404)
return RedirectResponse(url="/login")
return HTMLResponse(content="<h1>Olá</h1>")JSONResponse permite controlar status e headers. RedirectResponse faz redirect. HTMLResponse retorna HTML directo. Por padrão, dicionários viram JSON automaticamente.
Query parameters
@app.get("/itens")
def listar(skip: int = 0, limit: int = 10):
return itens[skip : skip + limit]
# /itens?skip=0&limit=5Parâmetros com valor default tornam-se query params. skip e limit são opcionais na URL. Sem default, seriam obrigatórios. O FastAPI valida tipos automaticamente.
Response model
class UserOut(BaseModel):
id: int
nome: str
email: str
@app.get("/users/{id}", response_model=UserOut)
def get_user(id: int):
return user_com_password # password é filtradoresponse_model controla os campos retornados. Campos fora do modelo são removidos automaticamente. Protege dados sensíveis como passwords. Gera documentação precisa da resposta.
APIRouter
from fastapi import APIRouter
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/")
def listar(): ...
@router.get("/{user_id}")
def get(user_id: int): ...
# main.py:
app.include_router(router)APIRouter organiza rotas em módulos separados. prefix adiciona caminho base a todas as rotas. include_router() regista no app principal. Essencial para projectos grandes.
Route naming e docs
@app.get(
"/itens",
summary="Listar itens",
description="Retorna todos os itens activos com paginação.",
response_description="Lista de itens",
deprecated=True,
)
def listar(): ...summary é o título curto no Swagger. description adiciona detalhes longos. response_description documenta a resposta. deprecated=True marca como obsoleto na docs.
Parâmetros e Validação
Body com Pydantic
from pydantic import BaseModel
class Item(BaseModel):
nome: str
preco: float
ativo: bool = True
@app.post("/itens")
def criar(item: Item):
return itemParâmetros com tipo BaseModel são lidos do body JSON. O FastAPI valida automaticamente os campos. Erros retornam 422 com detalhes. Não precisa de parsing manual do request.
Form data
from fastapi import Form
@app.post("/login")
def login(
username: str = Form(),
password: str = Form(),
):
return {"user": username}Form() lê dados de formulários HTML (application/x-www-form-urlencoded). Requer python-multipart instalado. Não pode ser combinado com body JSON no mesmo endpoint.
Enum parameters
from enum import Enum
class Orden(str, Enum):
nome = "nome"
preco = "preco"
data = "data"
@app.get("/itens")
def listar(ordem: Orden = Orden.nome):
return {"ordenado_por": ordem.value}Enum restringe valores aceites. Herdar de str permite comparação directa. Valores inválidos retornam 422. A documentação mostra os valores disponíveis como dropdown.
Validação de query
from fastapi import Query
@app.get("/itens")
def listar(
q: str | None = Query(None, min_length=3, max_length=50),
page: int = Query(1, ge=1),
):
...Query() adiciona validações a parâmetros de query. min_length e max_length limitam strings. ge (greater or equal) valida números. None como default torna opcional.
Upload de ficheiros
from fastapi import UploadFile, File
@app.post("/upload")
async def upload(ficheiro: UploadFile = File(...)):
conteudo = await ficheiro.read()
return {
"nome": ficheiro.filename,
"tipo": ficheiro.content_type,
"tamanho": len(conteudo),
}UploadFile gere uploads com streaming. File(...) torna obrigatório. filename e content_type dão metadata. read() é async para ficheiros grandes.
Validação de path
from fastapi import Path
@app.get("/itens/{item_id}")
def get_item(
item_id: int = Path(ge=1, title="ID do Item"),
):
...Path() adiciona constraints a path parameters. ge=1 garante valor positivo. title aparece na documentação. Útil para IDs que devem ser sempre positivos.
Múltiplos ficheiros
from fastapi import UploadFile, File
@app.post("/uploads")
async def uploads(ficheiros: list[UploadFile] = File(...)):
return [
{"nome": f.filename, "tamanho": f.size}
for f in ficheiros
]list[UploadFile] aceita múltiplos ficheiros. O cliente envia com o mesmo nome de campo. size dá o tamanho sem ler tudo. Ideal para galerias e imports em massa.
Headers e Cookies
from fastapi import Header, Cookie
@app.get("/info")
def info(
user_agent: str = Header(),
x_token: str | None = Header(None),
sessao: str | None = Cookie(None),
):
return {"agent": user_agent}Header() lê cabeçalhos HTTP. Hífens viram underscores automaticamente (x-token → x_token). Cookie() lê cookies do request. Ambos suportam validação e defaults.
Parâmetros opcionais
@app.get("/itens")
def listar(
q: str | None = None,
categoria: str | None = None,
preco_max: float | None = None,
):
resultados = itens
if q:
resultados = [i for i in resultados if q in i["nome"]]
return resultadosstr | None = None torna o parâmetro opcional. Sintaxe union type do Python 3.10+. Se omitido na URL, recebe None. Permite filtros dinâmicos sem múltiplos endpoints.
Pydantic e Models
Model básico
from pydantic import BaseModel
class User(BaseModel):
nome: str
email: str
idade: int | None = None
ativo: bool = TrueBaseModel define schemas com validação automática. Type hints definem o tipo esperado. Campos com default são opcionais. int | None aceita inteiro ou null.
Model validators
from pydantic import BaseModel, model_validator
class Registo(BaseModel):
password: str
confirmar: str
@model_validator(mode="after")
def verificar_igualdade(self):
if self.password != self.confirmar:
raise ValueError("Passwords não coincidem")
return self@model_validator valida múltiplos campos em conjunto. mode="after" executa após validação individual. Acede a self com todos os campos. Ideal para confirmações e dependências entre campos.
Field e constraints
from pydantic import BaseModel, Field
class Produto(BaseModel):
nome: str = Field(min_length=1, max_length=200)
preco: float = Field(gt=0, description="Preço em euros")
tags: list[str] = []
quantidade: int = Field(default=0, ge=0)Field() adiciona validações e metadata. gt=0 garante maior que zero. min_length valida tamanho de strings. description aparece na documentação OpenAPI.
Serialização
user = User(nome="Ana", email="ana@site.pt") # para dicionário: dados = user.model_dump() # para JSON: json_str = user.model_dump_json() # de JSON: user2 = User.model_validate_json(json_str)
model_dump() converte para dicionário Python. model_dump_json() serializa para string JSON. model_validate_json() faz parse de JSON para modelo. Substituem os antigos .dict() e .json().
Validadores custom
from pydantic import BaseModel, field_validator
class User(BaseModel):
email: str
@field_validator("email")
@classmethod
def validar_email(cls, v: str) -> str:
if "@" not in v:
raise ValueError("email inválido")
return v.lower()@field_validator cria validação personalizada. @classmethod é obrigatório. ValueError gera erro 422. O valor retornado substitui o original (normalização).
Model config
from pydantic import BaseModel, ConfigDict
class User(BaseModel):
model_config = ConfigDict(
str_strip_whitespace=True,
str_to_lower=True,
frozen=True,
)
nome: str
email: strConfigDict configura comportamento do modelo. str_strip_whitespace remove espaços. frozen=True torna imutável. str_to_lower normaliza para minúsculas.
Nested models
class Endereco(BaseModel):
rua: str
cidade: str
codigo_postal: str
class User(BaseModel):
nome: str
enderecos: list[Endereco] = []
principal: Endereco | None = NoneModelos podem conter outros modelos. list[Endereco] valida arrays de objectos. Endereco | None aceita objecto ou null. A validação é recursiva e automática.
Computed fields
from pydantic import BaseModel, computed_field
class Rect(BaseModel):
largura: float
altura: float
@computed_field
@property
def area(self) -> float:
return self.largura * self.altura@computed_field adiciona campos calculados à serialização. Aparece no JSON de resposta automaticamente. @property torna-o apenas de leitura. Útil para totais e derivados.
Base de Dados
Setup SQLAlchemy
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, DeclarativeBase
SQLALCHEMY_URL = "sqlite:///./app.db"
engine = create_engine(SQLALCHEMY_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine)
class Base(DeclarativeBase):
passcreate_engine() cria a ligação à BD. sessionmaker fábrica de sessões. DeclarativeBase é a base para modelos. check_same_thread é necessário para SQLite.
Relações
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
user: Mapped["User"] = relationship(back_populates="posts")
class User(Base):
posts: Mapped[list["Post"]] = relationship(back_populates="user")ForeignKey cria a ligação na BD. relationship() define acesso em Python. back_populates liga ambos os lados. Mapped[list] para relações one-to-many.
Pydantic + SQLAlchemy
from pydantic import BaseModel, ConfigDict
class UserOut(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
nome: str
email: str
@app.get("/users/{id}", response_model=UserOut)
def get_user(id: int, db: Session = Depends(get_db)):
return db.query(User).filter_by(id=id).first()from_attributes=True permite criar Pydantic a partir de objectos ORM. Substitui o antigo orm_mode. O response_model filtra campos automaticamente. Ponte entre BD e API.
Model SQLAlchemy
from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.orm import Mapped, mapped_column
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
nome: Mapped[str] = mapped_column(String(100))
email: Mapped[str] = mapped_column(unique=True)
ativo: Mapped[bool] = mapped_column(default=True)mapped_column() é a API moderna do SQLAlchemy 2.0. Mapped[type] define o tipo Python. primary_key marca a chave primária. unique impede duplicados.
Migrações (Alembic)
pip install alembic alembic init migrations # gerar migração automática: alembic revision --autogenerate -m "criar users" # aplicar: alembic upgrade head # reverter: alembic downgrade -1
Alembic gere migrações de esquema. --autogenerate compara modelos com a BD. upgrade head aplica todas as pendentes. downgrade -1 reverte a última. Essencial para evolução do esquema.
Dependência de sessão
from sqlalchemy.orm import Session
from fastapi import Depends
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.get("/users")
def listar(db: Session = Depends(get_db)):
return db.query(User).all()Depends(get_db) injecta a sessão em cada request. yield mantém a sessão aberta durante o endpoint. finally garante o fecho. Padrão de dependency injection do FastAPI.
Queries avançadas
from sqlalchemy import select, func # com select (SQLAlchemy 2.0): stmt = select(User).where(User.ativo == True).order_by(User.nome) users = db.scalars(stmt).all() # agregações: total = db.scalar(select(func.count(User.id))) media = db.scalar(select(func.avg(User.idade)))
select() é a API moderna de queries. scalars() retorna objectos directamente. func.count() e func.avg() fazem agregações. where() encadeia condições.
CRUD operações
# Criar user = User(nome="Ana", email="ana@site.pt") db.add(user) db.commit() db.refresh(user) # Ler users = db.query(User).filter(User.ativo == True).all() user = db.query(User).filter_by(id=1).first() # Apagar db.delete(user) db.commit()
add() + commit() insere na BD. refresh() recarrega com dados gerados (id). filter() usa expressões, filter_by() usa kwargs. first() retorna None se não existir.
Async database
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
AsyncSession = async_sessionmaker(engine)
async def get_db():
async with AsyncSession() as session:
yield session
@app.get("/users")
async def listar(db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User))
return result.scalars().all()create_async_engine cria engine assíncrona. async_sessionmaker fábrica de sessões async. await db.execute() executa queries sem bloquear. Requer driver async como asyncpg.
Autenticação e Segurança
OAuth2 + JWT setup
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@app.get("/perfil")
def perfil(token: str = Depends(oauth2_scheme)):
return {"token": token}OAuth2PasswordBearer extrai o token do header Authorization. tokenUrl indica o endpoint de login. O Swagger UI ganha botão "Authorize" automaticamente. Depends injecta o token.
Endpoint de login
from fastapi.security import OAuth2PasswordRequestForm
@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
user = autenticar(form.username, form.password)
if not user:
raise HTTPException(401, "Credenciais inválidas")
token = criar_token({"sub": str(user.id)})
return {"access_token": token, "token_type": "bearer"}OAuth2PasswordRequestForm espera username e password como form data. O response deve ter access_token e token_type. Compatível com o fluxo OAuth2 do Swagger.
Gerar token JWT
from jose import jwt
from datetime import datetime, timedelta
SECRET_KEY = "super-secreto"
ALGORITHM = "HS256"
def criar_token(data: dict, expira_min: int = 30):
payload = data.copy()
payload["exp"] = datetime.utcnow() + timedelta(minutes=expira_min)
return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)jwt.encode() cria o token assinado. exp define expiração (obrigatório). HS256 é o algoritmo de assinatura. SECRET_KEY deve estar em variável de ambiente.
Dependências de auth
from fastapi import Depends, HTTPException
def admin_required(user=Depends(get_user_atual)):
if user.role != "admin":
raise HTTPException(403, "Acesso negado")
return user
@app.delete("/users/{id}", dependencies=[Depends(admin_required)])
def apagar_user(id: int): ...Dependências encadeiam: admin_required depende de get_user_atual. dependencies=[] no decorador protege sem injectar. Retorne 403 para permissões insuficientes. Padrão reutilizável.
Validar token
from jose import JWTError, jwt
from fastapi import HTTPException
def get_user_atual(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: str = payload.get("sub")
if user_id is None:
raise HTTPException(status_code=401)
except JWTError:
raise HTTPException(status_code=401, detail="Token inválido")
return user_idjwt.decode() valida assinatura e expiração. sub é o subject (ID do user). JWTError captura tokens inválidos ou expirados. Retorne 401 para credenciais inválidas.
CORS
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)CORSMiddleware permite requests de outros domínios. allow_origins lista domínios permitidos. allow_credentials permite cookies. Em produção, nunca use ["*"] com credentials.
Hash de password
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verificar(password: str, hashed: str) -> bool:
return pwd_context.verify(password, hashed)CryptContext gere hashing com salt automático. bcrypt é o esquema recomendado. hash() gera hash irreversível. verify() compara input com hash guardado. Nunca guarde passwords em texto simples.
Rate limiting
from slowapi import Limiter
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
@app.get("/api")
@limiter.limit("10/minute")
def api(request: Request):
return {"dados": "..."}slowapi adiciona rate limiting. key_func identifica o cliente (por IP). limit() define máximo por período. Retorna 429 quando excede. Essencial para APIs públicas.
Async e Background
Endpoint async
@app.get("/dados")
async def get_dados():
resultado = await fetch_externo()
return resultadoasync def define um endpoint assíncrono. await espera operações I/O sem bloquear. O servidor atende outros requests enquanto espera. Use para BD, HTTP calls e ficheiros.
Lifespan events
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
# startup: abrir conexões, carregar cache
print("A iniciar...")
yield
# shutdown: fechar conexões, limpar recursos
print("A encerrar...")
app = FastAPI(lifespan=lifespan)lifespan substitui os antigos eventos on_startup/on_shutdown. Código antes do yield corre no arranque. Código depois corre no encerramento. Ideal para inicializar recursos.
async vs sync
# async: para I/O (BD, HTTP, ficheiros)
@app.get("/a")
async def endpoint_a():
dados = await api_externa()
return dados
# sync: para CPU (cálculos, processamento)
@app.get("/b")
def endpoint_b():
resultado = calculo_pesado()
return resultadoUse async para operações I/O (rede, BD). Use def normal para CPU-bound. Endpoints sync correm em threadpool automaticamente. Misturar os dois é perfeitamente válido.
Concurrency com asyncio
import asyncio
@app.get("/paralelo")
async def paralelo():
resultados = await asyncio.gather(
fetch_api_1(),
fetch_api_2(),
fetch_api_3(),
)
return {"resultados": resultados}asyncio.gather() executa múltiplas tasks em paralelo. Todas correm concorrentemente no event loop. Mais rápido que chamadas sequenciais. Ideal para agregar dados de múltiplas APIs.
Background tasks
from fastapi import BackgroundTasks
def enviar_email(to: str, msg: str):
# tarefa demorada (SMTP)
...
@app.post("/registo")
def registo(bg: BackgroundTasks):
bg.add_task(enviar_email, "ana@site.pt", "Bem-vinda!")
return {"msg": "Registado"}BackgroundTasks executa após enviar a resposta. add_task() agenda a função com argumentos. O cliente não espera a tarefa terminar. Ideal para emails, notificações e logs.
Streaming response
from fastapi.responses import StreamingResponse
async def gerar_dados():
for i in range(1000):
yield f"linha {i}\n"
@app.get("/stream")
def stream():
return StreamingResponse(
gerar_dados(),
media_type="text/plain",
)StreamingResponse envia dados progressivamente. O gerador yield produz chunks. media_type define o Content-Type. Ideal para ficheiros grandes, SSE e downloads.
httpx (async HTTP)
import httpx
@app.get("/externo")
async def externo():
async with httpx.AsyncClient() as client:
response = await client.get("https://api.exemplo.pt/dados")
return response.json()httpx.AsyncClient faz pedidos HTTP assíncronos. async with gere a conexão automaticamente. await client.get() não bloqueia o event loop. Substituto moderno do requests para async.
WebSockets
from fastapi import WebSocket
@app.websocket("/ws")
async def websocket(ws: WebSocket):
await ws.accept()
while True:
dados = await ws.receive_text()
await ws.send_text(f"Echo: {dados}")@app.websocket cria endpoint WebSocket. accept() estabelece a conexão. receive_text() e send_text() trocam mensagens. Conexão persistente para tempo real (chat, notificações).
Avançado e Deploy
Middleware
import time
@app.middleware("http")
async def tempo_request(request, call_next):
inicio = time.time()
response = await call_next(request)
duracao = time.time() - inicio
response.headers["X-Process-Time"] = str(duracao)
return response@app.middleware("http") intercepta todos os requests. call_next() passa ao próximo handler. Permite adicionar headers, logging ou métricas. Executa antes e depois do endpoint.
Testes com override
from fastapi.testclient import TestClient
def override_get_db():
db = TestingSessionLocal()
try:
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)dependency_overrides substitui dependências em testes. Troque a BD real por uma de teste. O dicionário mapeia original → substituto. Limpe com .clear() após os testes.
Boas práticas
# 1. Type hints em todo o lado # 2. Pydantic para validação (não manual) # 3. Dependencies para lógica partilhada # 4. APIRouter para organizar # 5. response_model para controlar output
Use type hints sempre — geram validação grátis. Prefira Pydantic a validação manual. Depends() elimina repetição. APIRouter escala o projecto. response_model protege dados sensíveis.
Tratamento de erros
from fastapi import HTTPException
@app.get("/itens/{id}")
def get_item(id: int):
item = procurar(id)
if not item:
raise HTTPException(status_code=404, detail="Item não encontrado")
return item
# handler custom:
@app.exception_handler(ValueError)
async def value_error_handler(request, exc):
return JSONResponse(status_code=400, content={"erro": str(exc)})HTTPException lança erros HTTP com detalhe. exception_handler captura excepções custom globalmente. detail aparece na resposta JSON. Sempre use status codes apropriados.
Deploy com Uvicorn
# Produção: uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 # com Gunicorn: gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker # variáveis: # WEB_CONCURRENCY=4 # PORT=8000
--workers 4 cria múltiplos processos. Gunicorn como process manager em produção. UvicornWorker combina Gunicorn com ASGI. Regra: workers = 2 × CPU cores + 1.
Dependências reutilizáveis
from fastapi import Depends
class Pagination:
def __init__(self, page: int = 1, size: int = 10):
self.offset = (page - 1) * size
self.limit = size
@app.get("/itens")
def listar(pg: Pagination = Depends()):
return db.query(Item).offset(pg.offset).limit(pg.limit).all()Classes com __init__ funcionam como dependências. Depends() sem argumentos usa os defaults. Parâmetros viram query params automaticamente. Reutilize em múltiplos endpoints.
Docker
FROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . EXPOSE 8000 CMD ["uvicorn", "main:app", "--host", "0.0.0.0"]
python:3.12-slim é a imagem base leve. --no-cache-dir reduz tamanho. EXPOSE documenta a porta. --host 0.0.0.0 aceita conexões externas. Multi-stage build para optimizar.
Testes com TestClient
from fastapi.testclient import TestClient
client = TestClient(app)
def test_raiz():
r = client.get("/")
assert r.status_code == 200
assert r.json() == {"msg": "Olá!"}
def test_criar_item():
r = client.post("/itens", json={"nome": "Teste", "preco": 9.99})
assert r.status_code == 201TestClient simula requests HTTP sem servidor. json= envia body JSON. assert valida status e resposta. Compatível com pytest. Execute com pytest tests/.
Logging estruturado
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@app.get("/itens")
def listar():
logger.info("Listagem de itens", extra={"user": "ana"})
return itenslogging da stdlib é a forma padrão. extra adiciona contexto estruturado. level=INFO filtra por severidade. Em produção use JSON logging para agregação (ELK, Grafana).