FastAPI: do zero ao mercado de trabalho

Antes de escrever código, entenda o que você está aprendendo e onde isso encaixa numa vaga.

FastAPI é um framework web Python para construir APIs. Ele se destacou por três motivos que aparecem literalmente em descrições de vaga:

• Performance: construído sobre ASGI (Starlette + Uvicorn), suporta async/await nativamente e compete com Node.js e Go em throughput de I/O.
• Validação automática: usa Pydantic para validar entrada e saída a partir de type hints. Menos bug, menos código repetido.
• Documentação automática: gera OpenAPI (Swagger UI e ReDoc) sozinho. Em equipe, isso vale ouro: o frontend consome sua API sem você escrever documentação à mão.

FastAPI vs. Django vs. Flask

Como a apostila funciona

• Cada módulo tem teoria curta, código comentado e uma caixa 💼 mercado ligando o tema a entrevistas.
• Os módulos constroem um projeto-fio-condutor: uma API de tarefas (to-do) multi-usuário, que evolui até virar um serviço com autenticação, banco, testes e deploy — um portfólio real.
• Nível dos módulos: Básico Intermediário Avançado Expert

Instalação correta, servidor rodando e a documentação automática aberta no navegador.

Preparando o ambiente

Sempre use ambiente virtual — em entrevistas técnicas, rodar pip install global é um sinal amarelo.

terminal# Crie a pasta e o ambiente virtual
mkdir api-tarefas && cd api-tarefas
python -m venv .venv
source .venv/bin/activate      # Windows: .venv\Scripts\activate

# FastAPI + servidor (o extra [standard] inclui uvicorn e utilidades)
pip install "fastapi[standard]"

O menor app possível

main.pyfrom fastapi import FastAPI

app = FastAPI(title="API de Tarefas", version="0.1.0")

@app.get("/")
def raiz():
    return {"mensagem": "API no ar!"}

terminalfastapi dev main.py     # modo desenvolvimento, com reload automático
# alternativa clássica: uvicorn main:app --reload

Acesse:

• http://127.0.0.1:8000 — sua rota
• http://127.0.0.1:8000/docs — Swagger UI (documentação interativa, gerada sozinha)
• http://127.0.0.1:8000/redoc — ReDoc, visual alternativo

O vocabulário do dia a dia: path params, query params e corpo da requisição.

Métodos HTTP e semântica REST

Path parameters — parte da URL

main.py@app.get("/tarefas/{tarefa_id}")
def obter_tarefa(tarefa_id: int):
    return {"id": tarefa_id}

O type hint int faz três coisas: converte o valor, valida (acessar /tarefas/abc devolve erro 422 automático) e documenta o tipo no Swagger.

Query parameters — depois do ?

main.pyfrom typing import Annotated
from fastapi import Query

@app.get("/tarefas")
def listar_tarefas(
    concluida: bool | None = None,                      # opcional
    limite: Annotated[int, Query(ge=1, le=100)] = 10,  # com validação
    busca: Annotated[str | None, Query(max_length=50)] = None,
):
    return {"concluida": concluida, "limite": limite, "busca": busca}

Chamada: GET /tarefas?concluida=true&limite=20&busca=estudar. A regra é simples: se está na assinatura da função e não está no path, é query param.

Corpo da requisição (request body)

main.pyfrom pydantic import BaseModel

class TarefaEntrada(BaseModel):
    titulo: str
    descricao: str | None = None
    concluida: bool = False

@app.post("/tarefas", status_code=201)
def criar_tarefa(tarefa: TarefaEntrada):
    # tarefa já chega validada e tipada
    return tarefa

Pydantic é metade do valor do FastAPI. Dominar isso separa quem "fez tutorial" de quem trabalha com o framework.

Campos com regras

schemas.pyfrom pydantic import BaseModel, Field, EmailStr
from datetime import datetime
from enum import Enum

class Prioridade(str, Enum):
    baixa = "baixa"
    media = "media"
    alta = "alta"

class TarefaCriar(BaseModel):
    titulo: str = Field(min_length=3, max_length=120)
    descricao: str | None = Field(default=None, max_length=500)
    prioridade: Prioridade = Prioridade.media
    prazo: datetime | None = None
    responsavel_email: EmailStr | None = None   # pip install "pydantic[email]"

Qualquer entrada fora dessas regras devolve 422 Unprocessable Entity com uma lista detalhada de erros — sem você escrever um if sequer.

Validadores customizados

schemas.pyfrom pydantic import field_validator, model_validator

class TarefaCriar(BaseModel):
    titulo: str
    prazo: datetime | None = None
    inicio: datetime | None = None

    @field_validator("titulo")
    @classmethod
    def titulo_sem_espacos_extras(cls, v: str) -> str:
        v = v.strip()
        if not v:
            raise ValueError("título não pode ser vazio")
        return v

    @model_validator(mode="after")
    def prazo_depois_do_inicio(self):
        if self.inicio and self.prazo and self.prazo < self.inicio:
            raise ValueError("prazo deve ser depois do início")
        return self

O padrão profissional: schemas separados por operação

Nunca use um único modelo para tudo. O padrão de mercado é ter variações por caso de uso:

schemas.pyclass TarefaBase(BaseModel):
    titulo: str
    descricao: str | None = None

class TarefaCriar(TarefaBase):          # entrada do POST
    pass

class TarefaAtualizar(BaseModel):       # entrada do PATCH: tudo opcional
    titulo: str | None = None
    descricao: str | None = None
    concluida: bool | None = None

class TarefaSaida(TarefaBase):          # resposta: inclui id, nunca dados sensíveis
    id: int
    concluida: bool
    criada_em: datetime

    model_config = {"from_attributes": True}  # permite criar a partir de objeto do ORM

Uma API profissional se reconhece pela consistência das respostas de erro.

response_model: o contrato de saída

main.py@app.get("/tarefas/{tarefa_id}", response_model=TarefaSaida)
def obter_tarefa(tarefa_id: int):
    tarefa = repositorio.buscar(tarefa_id)   # pode retornar objeto do ORM
    return tarefa                            # FastAPI filtra/valida pela TarefaSaida

response_model filtra a resposta: mesmo que o objeto tenha 30 campos, só saem os declarados no schema. É sua camada de segurança contra vazamento de dados.

Erros com HTTPException

main.pyfrom fastapi import HTTPException, status

@app.get("/tarefas/{tarefa_id}", response_model=TarefaSaida)
def obter_tarefa(tarefa_id: int):
    tarefa = repositorio.buscar(tarefa_id)
    if tarefa is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Tarefa não encontrada",
        )
    return tarefa

Status codes que você precisa saber de cor

Exceções de domínio + handler global (padrão avançado)

Em projetos maiores, a regra de negócio não deve conhecer HTTP. Crie exceções próprias e traduza-as em um único lugar:

exceptions.pyclass ErroDeDominio(Exception):
    def __init__(self, mensagem: str):
        self.mensagem = mensagem

class TarefaNaoEncontrada(ErroDeDominio): ...
class TituloDuplicado(ErroDeDominio): ...

main.pyfrom fastapi import Request
from fastapi.responses import JSONResponse

MAPA_STATUS = {TarefaNaoEncontrada: 404, TituloDuplicado: 409}

@app.exception_handler(ErroDeDominio)
async def handler_dominio(request: Request, exc: ErroDeDominio):
    return JSONResponse(
        status_code=MAPA_STATUS.get(type(exc), 400),
        content={"detail": exc.mensagem},
    )

Sair do main.py gigante é o primeiro passo para parecer (e ser) profissional.

Layout de pastas usado no mercado

estrutura do projetoapi-tarefas/
├── app/
│   ├── main.py            # cria o app, registra routers e handlers
│   ├── core/
│   │   ├── config.py      # settings via variáveis de ambiente
│   │   └── security.py    # hash de senha, JWT
│   ├── api/
│   │   ├── deps.py        # dependências compartilhadas
│   │   └── routers/
│   │       ├── tarefas.py
│   │       └── usuarios.py
│   ├── models/            # tabelas SQLAlchemy
│   ├── schemas/           # modelos Pydantic
│   ├── services/          # regra de negócio
│   └── repositories/      # acesso a dados
├── tests/
├── alembic/               # migrações de banco
├── .env
├── pyproject.toml
└── Dockerfile

APIRouter: dividindo rotas por domínio

app/api/routers/tarefas.pyfrom fastapi import APIRouter

router = APIRouter(prefix="/tarefas", tags=["tarefas"])

@router.get("")
def listar(): ...

@router.post("", status_code=201)
def criar(): ...

app/main.pyfrom fastapi import FastAPI
from app.api.routers import tarefas, usuarios

app = FastAPI(title="API de Tarefas")
app.include_router(tarefas.router)
app.include_router(usuarios.router)

O tags agrupa as rotas no Swagger; o prefix evita repetir o caminho em cada rota. Empresas costumam versionar: prefix="/api/v1/tarefas".

Configuração com pydantic-settings

Credenciais e configurações nunca vão no código. O padrão é ler de variáveis de ambiente com validação:

app/core/config.pyfrom pydantic_settings import BaseSettings   # pip install pydantic-settings
from functools import lru_cache

class Settings(BaseSettings):
    app_name: str = "API de Tarefas"
    database_url: str                    # obrigatório: falha no boot se faltar
    secret_key: str
    access_token_expire_minutes: int = 30
    debug: bool = False

    model_config = {"env_file": ".env"}

@lru_cache
def get_settings() -> Settings:
    return Settings()

.env  (NUNCA comitar — adicione ao .gitignore)DATABASE_URL=postgresql+asyncpg://user:senha@localhost:5432/tarefas
SECRET_KEY=troque-por-um-valor-aleatorio-longo
DEBUG=true

O recurso mais elegante do FastAPI — e a base de tudo o que vem depois: banco, autenticação e testes.

A ideia

Em vez de cada rota criar suas próprias conexões e verificações, ela declara o que precisa e o FastAPI entrega. Isso remove duplicação e permite substituir qualquer peça nos testes.

app/api/deps.pyfrom typing import Annotated
from fastapi import Depends, Query

class Paginacao:
    def __init__(
        self,
        pagina: Annotated[int, Query(ge=1)] = 1,
        tamanho: Annotated[int, Query(ge=1, le=100)] = 20,
    ):
        self.offset = (pagina - 1) * tamanho
        self.limit = tamanho

PaginacaoDep = Annotated[Paginacao, Depends()]

app/api/routers/tarefas.py@router.get("")
def listar(pag: PaginacaoDep):
    return repositorio.listar(offset=pag.offset, limit=pag.limit)

Qualquer rota que precise de paginação reutiliza PaginacaoDep. Um lugar para validar, um lugar para mudar.

Dependências com yield: setup e teardown

O padrão mais importante da sua carreira com FastAPI — é assim que sessões de banco são abertas e fechadas com segurança:

app/api/deps.pyasync def get_db():
    async with SessionLocal() as session:   # abre a sessão
        try:
            yield session                    # entrega para a rota
            await session.commit()           # commit se tudo deu certo
        except Exception:
            await session.rollback()         # desfaz em caso de erro
            raise

DbDep = Annotated[AsyncSession, Depends(get_db)]

Tudo antes do yield roda antes da rota; tudo depois roda ao final da resposta — mesmo se houver exceção. É o equivalente a um context manager por requisição.

Dependências encadeadas

Dependências podem depender de outras. É assim que a autenticação funciona (módulo 9):

conceitoget_db  →  get_usuario_atual  →  get_usuario_admin  →  rota

# a rota declara apenas o último elo:
@router.delete("/usuarios/{id}")
def remover(id: int, admin: Annotated[Usuario, Depends(get_usuario_admin)]):
    ...

Sobrescrevendo dependências (a mágica dos testes)

tests/conftest.pyapp.dependency_overrides[get_db] = get_db_de_teste
# todas as rotas passam a usar o banco de teste, sem mudar 1 linha do app

A stack pedida nas vagas: PostgreSQL, SQLAlchemy assíncrono e migrações versionadas.

terminalpip install sqlalchemy asyncpg alembic

Engine e sessão

app/core/database.pyfrom sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from app.core.config import get_settings

engine = create_async_engine(get_settings().database_url, echo=False)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

Modelos no estilo SQLAlchemy 2.0 (Mapped)

app/models/tarefa.pyfrom datetime import datetime
from sqlalchemy import String, ForeignKey, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.core.database import Base

class Tarefa(Base):
    __tablename__ = "tarefas"

    id: Mapped[int] = mapped_column(primary_key=True)
    titulo: Mapped[str] = mapped_column(String(120), index=True)
    descricao: Mapped[str | None] = mapped_column(String(500))
    concluida: Mapped[bool] = mapped_column(default=False)
    criada_em: Mapped[datetime] = mapped_column(server_default=func.now())

    dono_id: Mapped[int] = mapped_column(ForeignKey("usuarios.id"))
    dono: Mapped["Usuario"] = relationship(back_populates="tarefas")

Repository: consultas isoladas

app/repositories/tarefas.pyfrom sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.tarefa import Tarefa

class TarefaRepository:
    def __init__(self, db: AsyncSession):
        self.db = db

    async def listar(self, dono_id: int, offset: int, limit: int) -> list[Tarefa]:
        stmt = (
            select(Tarefa)
            .where(Tarefa.dono_id == dono_id)
            .order_by(Tarefa.criada_em.desc())
            .offset(offset).limit(limit)
        )
        resultado = await self.db.execute(stmt)
        return list(resultado.scalars())

    async def criar(self, dados: dict, dono_id: int) -> Tarefa:
        tarefa = Tarefa(**dados, dono_id=dono_id)
        self.db.add(tarefa)
        await self.db.flush()      # gera o id sem fechar a transação
        return tarefa

A rota fica limpa: recebe o schema, chama o repository, devolve o response_model. O commit acontece na dependência get_db (módulo 7).

Alembic: migrações versionadas

Alterar tabela na mão não existe em ambiente profissional. O Alembic gera scripts de migração a partir dos seus modelos:

terminalalembic init -t async alembic          # configura para modo async
# edite alembic/env.py: target_metadata = Base.metadata

alembic revision --autogenerate -m "cria tabelas usuario e tarefa"
alembic upgrade head                   # aplica no banco
alembic downgrade -1                   # desfaz a última

O fluxo de login que 90% das APIs do mercado usam — implementado do jeito certo.

terminalpip install pyjwt "passlib[bcrypt]"

1. Hash de senha (nunca guarde senha em texto)

app/core/security.pyfrom passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_senha(senha: str) -> str:
    return pwd_context.hash(senha)

def verificar_senha(senha: str, hash_: str) -> bool:
    return pwd_context.verify(senha, hash_)

2. Criação do token JWT

app/core/security.pyimport jwt
from datetime import datetime, timedelta, timezone
from app.core.config import get_settings

ALGORITMO = "HS256"

def criar_access_token(sub: str) -> str:
    settings = get_settings()
    payload = {
        "sub": sub,                                        # id do usuário
        "exp": datetime.now(timezone.utc)
               + timedelta(minutes=settings.access_token_expire_minutes),
    }
    return jwt.encode(payload, settings.secret_key, algorithm=ALGORITMO)

3. Rota de login (padrão OAuth2 Password Flow)

app/api/routers/auth.pyfrom fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm

router = APIRouter(tags=["auth"])

@router.post("/token")
async def login(
    form: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: DbDep,
):
    usuario = await UsuarioRepository(db).por_email(form.username)
    if not usuario or not verificar_senha(form.password, usuario.senha_hash):
        raise HTTPException(401, detail="Credenciais inválidas",
                            headers={"WWW-Authenticate": "Bearer"})
    return {
        "access_token": criar_access_token(sub=str(usuario.id)),
        "token_type": "bearer",
    }

4. Protegendo rotas: a dependência get_usuario_atual

app/api/deps.pyfrom fastapi.security import OAuth2PasswordBearer
import jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

async def get_usuario_atual(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: DbDep,
) -> Usuario:
    erro = HTTPException(401, detail="Token inválido ou expirado",
                         headers={"WWW-Authenticate": "Bearer"})
    try:
        payload = jwt.decode(token, get_settings().secret_key, algorithms=[ALGORITMO])
        usuario_id = payload.get("sub")
    except jwt.PyJWTError:
        raise erro
    usuario = await UsuarioRepository(db).por_id(int(usuario_id))
    if usuario is None:
        raise erro
    return usuario

UsuarioAtual = Annotated[Usuario, Depends(get_usuario_atual)]

app/api/routers/tarefas.py@router.get("", response_model=list[TarefaSaida])
async def minhas_tarefas(usuario: UsuarioAtual, db: DbDep, pag: PaginacaoDep):
    return await TarefaRepository(db).listar(usuario.id, pag.offset, pag.limit)

No Swagger, aparece o botão Authorize: faça login, cole o token, e todas as rotas protegidas passam a funcionar na interface.

Além do básico: o que diferencia um sênior

• Refresh tokens: access token curto (15–30 min) + refresh token longo, armazenado com possibilidade de revogação, para renovar sem novo login.
• Escopos/roles: claim scopes no token + dependência get_usuario_admin que valida permissão → 403 quando falta.
• Onde guardar o token no frontend: cookie HttpOnly + Secure protege contra XSS; localStorage é mais simples porém exposto. Saber discutir o trade-off é ponto em entrevista.

O que roda ao redor de todas as rotas — e o erro de CORS que todo time frontend vai te cobrar.

CORS: o primeiro bug de integração da sua carreira

Navegadores bloqueiam requisições de um domínio para outro por padrão. Quando o frontend (ex.: localhost:5173) chamar sua API (localhost:8000), vai falhar até você liberar:

app/main.pyfrom fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173", "https://meuapp.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Middleware customizado

Middleware intercepta toda requisição/resposta. Exemplo clássico: medir tempo e propagar um ID de correlação para os logs:

app/main.pyimport time, uuid
from fastapi import Request

@app.middleware("http")
async def request_context(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
    inicio = time.perf_counter()

    response = await call_next(request)

    duracao_ms = (time.perf_counter() - inicio) * 1000
    response.headers["X-Request-ID"] = request_id
    response.headers["X-Process-Time-Ms"] = f"{duracao_ms:.1f}"
    return response

Lifespan: inicialização e encerramento

Conexões com Redis, pools, clientes HTTP: crie uma vez no boot, feche no shutdown.

app/main.pyfrom contextlib import asynccontextmanager
import httpx

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(timeout=10)   # startup
    yield
    await app.state.http.aclose()                    # shutdown

app = FastAPI(lifespan=lifespan)

Background tasks

Para tarefas rápidas que não devem atrasar a resposta (e-mail de boas-vindas, log de auditoria):

app/api/routers/usuarios.pyfrom fastapi import BackgroundTasks

@router.post("", status_code=201)
async def cadastrar(dados: UsuarioCriar, tasks: BackgroundTasks, db: DbDep):
    usuario = await servico.cadastrar(db, dados)
    tasks.add_task(enviar_email_boas_vindas, usuario.email)  # roda após a resposta
    return usuario

Limite: background task roda no mesmo processo. Para trabalho pesado, recorrente ou que precisa de retry, use fila (Celery/ARQ — módulo 16).

O tema que mais derruba candidatos em entrevistas de FastAPI. Depois deste módulo, não derruba mais você.

A regra de ouro

o erro clássico@app.get("/ruim")
async def ruim():
    time.sleep(5)          # 🚨 congela a API inteira por 5s

@app.get("/ok")
async def ok():
    await asyncio.sleep(5)  # ✅ só esta requisição espera

@app.get("/tambem-ok")
def tambem_ok():
    time.sleep(5)           # ✅ roda em thread separada

Se precisar rodar algo bloqueante dentro de uma rota async: await run_in_threadpool(funcao_pesada, args) (de fastapi.concurrency).

Concorrência real: várias chamadas ao mesmo tempo

app/services/dashboard.pyimport asyncio, httpx

async def montar_dashboard(http: httpx.AsyncClient):
    # três chamadas externas em paralelo: tempo total = a mais lenta, não a soma
    clima, cotacao, noticias = await asyncio.gather(
        http.get("https://api.clima.com/hoje"),
        http.get("https://api.cambio.com/usd-brl"),
        http.get("https://api.noticias.com/top"),
    )
    return {"clima": clima.json(), "cotacao": cotacao.json(),
            "noticias": noticias.json()}

Streaming de respostas

app/api/routers/relatorios.pyfrom fastapi.responses import StreamingResponse

async def gerar_csv():
    yield "id,titulo,concluida\n"
    async for tarefa in repositorio.iterar_todas():
        yield f"{tarefa.id},{tarefa.titulo},{tarefa.concluida}\n"

@router.get("/relatorio.csv")
async def relatorio():
    return StreamingResponse(gerar_csv(), media_type="text/csv")

O mesmo padrão serve para Server-Sent Events (respostas de IA token a token, por exemplo) usando media_type="text/event-stream".

WebSockets: comunicação bidirecional

app/api/routers/chat.pyfrom fastapi import WebSocket, WebSocketDisconnect

conexoes: list[WebSocket] = []

@router.websocket("/ws/chat")
async def chat(ws: WebSocket):
    await ws.accept()
    conexoes.append(ws)
    try:
        while True:
            msg = await ws.receive_text()
            for conexao in conexoes:
                await conexao.send_text(msg)
    except WebSocketDisconnect:
        conexoes.remove(ws)

Em produção com múltiplos workers, a lista em memória não funciona — usa-se Redis Pub/Sub para distribuir mensagens entre processos (conceito que vale mencionar em entrevista).

Nenhum tópico pesa mais numa contratação de pleno/sênior do que saber testar. É também o que os testes técnicos avaliam explicitamente.

terminalpip install pytest pytest-asyncio httpx aiosqlite

Teste síncrono simples com TestClient

tests/test_health.pyfrom fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_raiz_responde():
    resposta = client.get("/")
    assert resposta.status_code == 200
    assert resposta.json()["mensagem"] == "API no ar!"

Infra de teste profissional: banco isolado + overrides

tests/conftest.pyimport pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from app.main import app
from app.api.deps import get_db
from app.core.database import Base

engine_teste = create_async_engine("sqlite+aiosqlite:///:memory:")
SessaoTeste = async_sessionmaker(engine_teste, expire_on_commit=False)

@pytest.fixture
async def db():
    async with engine_teste.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    async with SessaoTeste() as sessao:
        yield sessao
    async with engine_teste.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)   # banco limpo por teste

@pytest.fixture
async def cliente(db):
    app.dependency_overrides[get_db] = lambda: db
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

Testando o fluxo real: cadastro → login → rota protegida

tests/test_tarefas.pyimport pytest

@pytest.mark.asyncio
async def test_fluxo_completo(cliente):
    # 1. cadastro
    r = await cliente.post("/usuarios", json={
        "email": "ana@ex.com", "senha": "Segura123!"})
    assert r.status_code == 201

    # 2. login (OAuth2 usa form-data, não JSON)
    r = await cliente.post("/token", data={
        "username": "ana@ex.com", "password": "Segura123!"})
    token = r.json()["access_token"]

    # 3. cria tarefa autenticada
    headers = {"Authorization": f"Bearer {token}"}
    r = await cliente.post("/tarefas", headers=headers,
                           json={"titulo": "Estudar testes"})
    assert r.status_code == 201

    # 4. sem token → 401
    r = await cliente.get("/tarefas")
    assert r.status_code == 401

@pytest.mark.asyncio
async def test_validacao_recusa_titulo_curto(cliente):
    r = await cliente.post("/tarefas", json={"titulo": "ab"})
    assert r.status_code in (401, 422)   # conforme a ordem das checagens

Mock de serviços externos

tests/test_email.pyfrom unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_cadastro_dispara_email(cliente):
    with patch("app.services.email.enviar", new=AsyncMock()) as mock:
        await cliente.post("/usuarios", json={
            "email": "a@b.com", "senha": "Segura123!"})
        mock.assert_awaited_once()

Como sua API se comporta com 10 mil usuários — o assunto favorito de entrevistas de sênior.

Cache com Redis

terminalpip install redis

app/services/cache.pyimport json
from redis.asyncio import Redis

redis = Redis.from_url("redis://localhost:6379", decode_responses=True)

async def obter_estatisticas(usuario_id: int, db) -> dict:
    chave = f"stats:usuario:{usuario_id}"

    if cacheado := await redis.get(chave):        # cache hit
        return json.loads(cacheado)

    stats = await calcular_estatisticas_pesadas(db, usuario_id)  # cache miss
    await redis.set(chave, json.dumps(stats), ex=300)  # TTL de 5 min
    return stats

async def invalidar_estatisticas(usuario_id: int):
    await redis.delete(f"stats:usuario:{usuario_id}")  # chame ao alterar tarefas

Esse é o padrão cache-aside. Saiba discutir os dois problemas clássicos: invalidação (quando limpar?) e stampede (cache expira e mil requisições recalculam juntas — mitigação: lock ou TTL com jitter).

Paginação: offset vs cursor

Feeds e listas infinitas usam cursor; telas administrativas com número de páginas usam offset. Devolva metadados: {"items": [...], "total": 132, "proxima": "..."} .

Rate limiting

app/api/deps.pyasync def limitar_taxa(request: Request):
    """Janela fixa: máx. 60 requisições/minuto por IP."""
    ip = request.client.host
    chave = f"rate:{ip}"
    atual = await redis.incr(chave)
    if atual == 1:
        await redis.expire(chave, 60)
    if atual > 60:
        raise HTTPException(429, detail="Muitas requisições, tente em instantes")

# aplica no router inteiro:
router = APIRouter(dependencies=[Depends(limitar_taxa)])

Em produção também se usa a biblioteca slowapi ou rate limit no gateway/nginx. Cite o algoritmo token bucket como evolução da janela fixa.

Outros ganhos rápidos que caem em entrevista

• Índices no banco: a otimização nº 1. Toda coluna usada em WHERE/ORDER BY frequente merece análise (EXPLAIN ANALYZE).
• Pool de conexões: configure pool_size/max_overflow no engine; abrir conexão por requisição mata performance.
• GZip: app.add_middleware(GZipMiddleware, minimum_size=1000) para respostas grandes.
• ORJSONResponse: serialização JSON mais rápida (default_response_class=ORJSONResponse).
• Medir antes de otimizar: use locust ou k6 para teste de carga; fale de p95/p99, não de média.

"Funciona na minha máquina" não é entregável. Vaga que pede FastAPI quase sempre pede Docker junto.

Dockerfile de produção (multi-stage)

Dockerfile# ---- estágio de build: instala dependências ----
FROM python:3.12-slim AS build
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# ---- estágio final: imagem enxuta ----
FROM python:3.12-slim
WORKDIR /app
COPY --from=build /install /usr/local
COPY app/ app/
COPY alembic/ alembic/
COPY alembic.ini .

RUN useradd --create-home apiuser
USER apiuser                                 # nunca rode como root

EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker-compose para desenvolvimento

docker-compose.ymlservices:
  api:
    build: .
    ports: ["8000:8000"]
    env_file: .env
    depends_on:
      db: {condition: service_healthy}
    command: uvicorn app.main:app --host 0.0.0.0 --reload

  db:
    image: postgres:16
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: senha
      POSTGRES_DB: tarefas
    volumes: [pgdata:/var/lib/postgresql/data]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user"]
      interval: 3s
      retries: 10

  redis:
    image: redis:7

volumes:
  pgdata:

Servindo em produção: workers

Um processo Python usa um núcleo. Em produção, rode múltiplos workers:

terminal# regra de bolso: (2 x núcleos) + 1 para apps com I/O
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4
# alternativa tradicional: gunicorn com worker class do uvicorn
gunicorn app.main:app -k uvicorn.workers.UvicornWorker -w 4

Na frente, um reverse proxy (nginx, Traefik ou o load balancer da nuvem) cuida de TLS, compressão e distribuição. Em Kubernetes, o padrão muda: 1 worker por container e o próprio cluster escala réplicas.

Health checks: obrigatório para orquestração

app/api/routers/health.py@router.get("/health", include_in_schema=False)
async def health():
    return {"status": "ok"}

@router.get("/ready", include_in_schema=False)
async def ready(db: DbDep):
    await db.execute(text("SELECT 1"))   # confirma dependências vivas
    return {"status": "ready"}

CI com GitHub Actions

.github/workflows/ci.ymlname: CI
on: [push, pull_request]
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: {python-version: "3.12"}
      - run: pip install -r requirements.txt -r requirements-dev.txt
      - run: ruff check app tests          # lint
      - run: pytest --cov=app --cov-fail-under=80

Pipeline completo de mercado: lint → testes → build da imagem → push no registry → deploy (com alembic upgrade head como passo de release). Opções acessíveis para portfólio: Render, Railway, Fly.io; em nuvem grande: AWS ECS/App Runner, GCP Cloud Run.

Produção sem observabilidade é dirigir de olhos fechados. É o que separa quem "sobe API" de quem opera API.

Logging estruturado (JSON)

Logs em texto livre são inúteis em escala. O padrão é JSON, um evento por linha, com contexto:

app/core/logging.pyimport structlog   # pip install structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ]
)
logger = structlog.get_logger()

uso no middleware do módulo 10structlog.contextvars.bind_contextvars(request_id=request_id, path=request.url.path)
logger.info("request_concluida", status=response.status_code, duracao_ms=duracao_ms)

# saída:
# {"event":"request_concluida","status":200,"duracao_ms":41.2,
#  "request_id":"8f2c...","path":"/tarefas","timestamp":"2026-07-18T14:02:11Z"}

O request_id propagado permite correlacionar todos os logs de uma mesma requisição — e segui-la entre microserviços.

Métricas com Prometheus

terminal + app/main.pypip install prometheus-fastapi-instrumentator

from prometheus_fastapi_instrumentator import Instrumentator
Instrumentator().instrument(app).expose(app)   # cria a rota /metrics

Isso expõe automaticamente contagem de requisições, latência por rota e status codes — que o Prometheus coleta e o Grafana exibe. Vocabulário para entrevista: os golden signals — latência (p95/p99), tráfego, taxa de erro e saturação.

Rastreamento de erros com Sentry

app/main.pyimport sentry_sdk   # pip install "sentry-sdk[fastapi]"

sentry_sdk.init(dsn=get_settings().sentry_dsn, traces_sample_rate=0.1)
# toda exceção não tratada vira um evento com stack trace, request e usuário

Tracing distribuído (OpenTelemetry)

Em arquiteturas com vários serviços, o OpenTelemetry instrumenta FastAPI, SQLAlchemy e httpx para gerar spans: você vê que a requisição gastou 12ms na API A, 230ms no banco e 90ms na API B. Citar OTel em entrevista de sênior conta ponto — é o padrão da indústria.

Os temas de conversa de sênior: quando dividir serviços, como processar trabalho pesado e como versionar sem quebrar clientes.

Filas de tarefas: quando BackgroundTasks não basta

Processamento pesado (gerar PDF, transcodificar vídeo, importar planilha de 1M de linhas) não pode rodar no processo web. O padrão é fila + worker:

fluxoAPI (rápida)  →  publica job na fila (Redis/RabbitMQ)  →  responde 202 Accepted
                                    │
worker (processo separado)  ←  consome, processa, salva resultado
cliente consulta  GET /jobs/{id}  →  {"status": "processando" | "concluido"}

exemplo com Celery# app/worker.py
from celery import Celery
celery = Celery("tarefas", broker="redis://localhost:6379/0",
                backend="redis://localhost:6379/1")

@celery.task(bind=True, max_retries=3)
def gerar_relatorio(self, usuario_id: int):
    try:
        ...  # trabalho pesado, com retry automático
    except ErroTransitorio as exc:
        raise self.retry(exc=exc, countdown=30)

# na rota:
job = gerar_relatorio.delay(usuario.id)
return {"job_id": job.id}

Alternativas modernas e mais leves para FastAPI: ARQ (async nativo) e Dramatiq. O conceito importa mais que a ferramenta: desacoplar recebimento de processamento, com retry e idempotência.

Monolito modular vs microserviços

Resposta madura em entrevista: "começaria com monolito modular (routers/services por domínio, como esta apostila estrutura) e extrairia serviços quando houver dor real". Distribuir cedo demais é o erro clássico.

Comunicação entre serviços

• REST/HTTP: simples, universal; síncrono (acopla disponibilidade).
• Eventos (RabbitMQ/Kafka): assíncrono, resiliente; exige pensar em entrega duplicada → consumidores idempotentes.
• gRPC: binário e rápido, contratos protobuf; comum em comunicação interna de alta frequência.

Versionamento de API sem quebrar clientes

• Prefixo de rota (/api/v1/...) é o mais comum; header de versão é alternativa.
• Mudanças aditivas (campo novo opcional) não exigem versão nova; mudanças destrutivas (remover/renomear campo, mudar tipo) exigem.
• Deprecie com aviso: header Deprecation, changelog e prazo de desligamento.

Segurança além do JWT (OWASP API Top 10, resumido)

• BOLA (o nº 1): usuário A acessa /tarefas/42 do usuário B. Toda query deve filtrar por dono: WHERE dono_id = :usuario_atual — autorização em nível de objeto, sempre.
• Mass assignment: resolvido pelos schemas de entrada restritos (módulo 4).
• Injeção: ORM com parâmetros já protege de SQL injection; nunca formate SQL com f-string.
• Exposição de dados: response_model estrito; erros 500 sem stack trace.
• Rate limiting em rotas de login (força bruta) — módulo 13.

Transformando tudo o que você estudou em contratação.

O projeto de portfólio que funciona

Recrutadores técnicos abrem seu GitHub por ~3 minutos. Um único projeto completo vale mais que dez tutoriais clonados. Checklist do projeto desta apostila pronto para o portfólio:

• README com: o que é, como rodar em 1 comando (docker compose up), print do Swagger, decisões de arquitetura e um diagrama simples.
• Autenticação JWT completa, autorização por dono (anti-BOLA), schemas separados.
• PostgreSQL + Alembic (nada de SQLite/create_all em produção).
• Pasta tests/ com fluxo completo e casos de erro; badge de CI verde.
• Dockerfile multi-stage + docker-compose; deploy em Render/Railway/Fly com link vivo no README.
• Logs estruturados e rota /health — detalhes que sinalizam senioridade.
• Um diferencial: WebSocket de notificações, relatório em fila, ou integração com IA (FastAPI servindo um modelo).

As 15 perguntas de entrevista mais prováveis

Tente responder em voz alta antes de abrir cada resposta.

Roadmap de estudo sugerido

Últimos conselhos

• Explique enquanto codifica. Testes técnicos ao vivo avaliam comunicação tanto quanto código.
• Leia a documentação oficial (fastapi.tiangolo.com) — é excelente e citar detalhes dela em entrevista demonstra profundidade.
• Contribua ou publique: uma issue resolvida, um artigo explicando N+1 ou Depends no seu LinkedIn vira assunto de entrevista a seu favor.
• Não decore — reconstrua. Apague o projeto e refaça a autenticação de memória; onde travar é o que falta estudar.