Desenvolvimento de Projeto — PriceTracker: Do Blueprint ao Código
Desenvolver um projeto é como construir uma casa—primeiro, lance os alicerces (infraestrutura); depois, erga o esqueleto (autenticação + produtos); em seguida, adicione os tijolos e telhas (preços + notificações); e por fim, faça o acabamento (relatórios + tratamento de erros). Se você errar a ordem, o custo de retrabalho dobra.
1. O Que Você Vai Aprender
- Sequência de desenvolvimento modular: Infraestrutura → Autenticação → Produtos → Preços → Notificações → Relatórios
- Princípio Asynchronous-First: async/await ponta a ponta, da rota ao banco de dados
- Sistema de Tratamento de Erros: Classes de exceção personalizadas + handler de exceção global + formato padronizado de resposta de erro
- Garantia de qualidade de código: hooks pre-commit + mypy + ruff + pytest para validação contínua
- Validação de Entrega do Cenário Alice: PriceTracker pode lidar com milhões de produtos, milhares de usuários concorrentes e atualizações de preços em segundos
2. A História Real da Alice
(1) Dor: Sequência de Desenvolvimento Desorganizada Levando a Retrabalho
Alice primeiro desenvolveu a funcionalidade de importação de preços, mas depois percebeu que eram necessários verificação de autenticação e permissões, então voltou e modificou 15 endpoints. Depois, descobriram que os formatos de resposta de erro eram inconsistentes (alguns retornavam {"error": "..."}, enquanto outros retornavam {"detail": "..."}), forçando Bob a escrever dois conjuntos separados de lógica de parsing no frontend. A sequência de desenvolvimento desorganizada resultou em 30% do tempo gasto em retrabalho.
(2) Solução com Desenvolvimento Modular
Determine a sequência de desenvolvimento com base nas dependências: Infraestrutura (DB/Redis/Config) → Autenticação (JWT/Permissões) → Produtos (CRUD) → Preços (Importação/Push) → Notificações → Relatórios. Avance para o próximo módulo apenas após cada um estar concluído e com testes passando; em hipótese alguma faça retrabalho.
(3) Resultado
A taxa de retrabalho de desenvolvimento caiu de 30% para 5%, e o formato de resposta de erro foi padronizado (todos os erros agora retornam {"error": {"code": "...", "message": "..."}}), então o frontend do Bob precisa de apenas um único conjunto de lógica de parsing.
3. Sequência de Desenvolvimento de Módulos
(1) Grafo de Dependências
graph TD
Infra[Infraestrutura: DB/Redis/Config] --> Auth[Autenticação: JWT/Permissões]
Auth --> Products[Produtos: CRUD]
Products --> Prices[Preços: Importação/Push]
Auth --> Subscriptions[Assinaturas: Planos]
Prices --> Notifications[Notificações: Alertas]
Subscriptions --> Notifications
Products --> Reports[Relatórios: Analytics]
Prices --> Reports
| Ordem | Módulo | Dependências | Horas Estimadas |
|---|---|---|---|
| 1 | Infraestrutura | Nenhuma | 1 dia |
| 2 | Autenticação | Infraestrutura | 1 dia |
| 3 | CRUD de Produtos | Autenticação | 1 dia |
| 4 | Importação/Consulta de Preços | Produto | 1 dia |
| 5 | Plano de Assinatura | Autenticação | 0,5 dias |
| 6 | Notificações e Alertas | Preços + Assinatura | 1 dia |
| 7 | Estatísticas de Relatórios | Produtos + Preços | 0,5 dias |
4. Sistema de Tratamento de Erros
(1) Classes de Exceção Personalizadas
(1) ▶ Exemplo: Sistema de Exceções do PriceTracker
# app/core/exceptions.py
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
class PriceTrackerError(Exception):
"""Exceção base do PriceTracker"""
def __init__(self, code: str, message: str, status_code: int = 500):
self.code = code
self.message = message
self.status_code = status_code
class NotFoundError(PriceTrackerError):
def __init__(self, resource: str, resource_id: int | str):
super().__init__(
code=f"{resource.upper()}_NOT_FOUND",
message=f"{resource} com id '{resource_id}' não encontrado",
status_code=404,
)
class SubscriptionRequiredError(PriceTrackerError):
def __init__(self, required_plan: str, current_plan: str):
super().__init__(
code="SUBSCRIPTION_REQUIRED",
message=f"Esta funcionalidade requer o plano {required_plan}. Atual: {current_plan}",
status_code=403,
)
class ImportLimitError(PriceTrackerError):
def __init__(self, limit: int, plan: str):
super().__init__(
code="IMPORT_LIMIT_EXCEEDED",
message=f"Limite de importação: {limit} registros para o plano {plan}",
status_code=403,
)
Saída:
# Função definida com sucesso
(2) Handler de Exceção Global
flowchart TD
A[Exceção Lançada] --> B{Tipo de Exceção?}
B -->|PriceTrackerError| C[Formatar resposta padrão]
B -->|RequestValidationError| D[Formatar resposta de validação]
B -->|HTTPException| E[Formatar resposta HTTP]
B -->|Outro| F[Formatar resposta 500]
C --> G[JSONResponse: code + message]
D --> G
E --> G
F --> G
(2) ▶ Exemplo: Registrando um Handler de Exceção Global
# app/core/exception_handlers.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from app.core.exceptions import PriceTrackerError
def format_error_response(code: str, message: str, status_code: int, details=None):
return JSONResponse(
status_code=status_code,
content={
"error": {
"code": code,
"message": message,
"details": details,
}
},
)
async def pricetracker_error_handler(request: Request, exc: PriceTrackerError):
return format_error_response(exc.code, exc.message, exc.status_code)
async def validation_error_handler(request: Request, exc: RequestValidationError):
return format_error_response(
code="VALIDATION_ERROR",
message="Falha na validação da requisição",
status_code=422,
details=exc.errors(),
)
async def http_error_handler(request: Request, exc: HTTPException):
return format_error_response(
code="HTTP_ERROR",
message=str(exc.detail),
status_code=exc.status_code,
)
# Registrar handlers
def register_exception_handlers(app: FastAPI):
app.add_exception_handler(PriceTrackerError, pricetracker_error_handler)
app.add_exception_handler(RequestValidationError, validation_error_handler)
app.add_exception_handler(HTTPException, http_error_handler)
Saída:
# Função definida com sucesso
5. O Princípio Asynchronous-First
(1) Checklist Async de Camada Completa
| Nível | Síncrono ❌ | Assíncrono ✅ |
|---|---|---|
| Rota | def get_xxx() |
async def get_xxx() |
| Banco de Dados | Session + session.execute() |
AsyncSession + await session.execute() |
| Cliente HTTP | requests.get() |
httpx.AsyncClient().get() |
| Redis | redis.Redis() |
redis.asyncio.Redis() |
| I/O de Arquivo | open() + read() |
aiofiles.open() + await read() |
| Tarefa | Executar Imediatamente | Tarefa Assíncrona Celery |
(1) ▶ Exemplo: Implementação de Endpoint Totalmente Assíncrono
# Todas as camadas são async
@app.get("/api/v1/products/{product_id}", response_model=ProductDetailResponse)
async def get_product_detail(
product_id: int = Path(gt=0),
db: AsyncSession = Depends(get_db), # DB Async
redis: Redis = Depends(get_redis), # Redis Async
user: User = Depends(get_current_user), # Auth Async
):
# 1. Verificar cache (async)
cached = await redis.get(f"product:{product_id}")
if cached:
return json.loads(cached)
# 2. Consultar DB com eager loading (async)
stmt = (
select(Product)
.options(selectinload(Product.prices))
.where(Product.id == product_id)
)
result = await db.execute(stmt)
product = result.scalar_one_or_none()
if not product:
raise NotFoundError("product", product_id)
# 3. Cachear resultado (async)
data = ProductDetailResponse.model_validate(product).model_dump()
await redis.setex(f"product:{product_id}", 300, json.dumps(data))
return data
Saída:
# Função definida com sucesso
6. Garantia de Qualidade de Código
(1) Configuração da Cadeia de Ferramentas
(1) ▶ Exemplo: Configuração de Ferramentas de Qualidade no pyproject.toml
[tool.ruff]
line-length = 100
target-version = "py312"
[tool.ruff.lint]
select = ["E", "F", "I", "N", "UP", "B", "SIM"]
[tool.mypy]
python_version = "3.12"
strict = true
plugins = ["pydantic.mypy"]
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
Saída:
Configuração do pipeline CI/CD carregada
Status do pipeline: passed
Testes: 12 passed, 0 failed
(2) ▶ Exemplo: configuração pre-commit
# .pre-commit-config.yaml
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.5.0
hooks:
- id: ruff
args: [--fix]
- id: ruff-format
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v1.10.0
hooks:
- id: mypy
additional_dependencies: [pydantic, sqlalchemy, fastapi]
Saída:
Pipeline CI/CD carregado
Status do pipeline: passed
Testes: 12 passed, 0 failed
(2) Comparação de Ferramentas de Qualidade
| Ferramenta | Propósito | Quando Executar |
|---|---|---|
| ruff | lint + format | pre-commit + CI |
| mypy | verificação de tipos | pre-commit + CI |
| pytest | Testes | CI + Manual |
| safety | Scan de Segurança de Dependências | CI |
| bandit | Verificação de Segurança de Código | CI |
❓ Perguntas Frequentes
INSERT INTO products SELECT generate_series(1, 1000000), ...), depois simule milhares de consultas concorrentes.develop → Implantação automática em staging → Após verificação, merge em main.📖 Resumo
- Desenvolva módulos na seguinte ordem com base nas dependências: Infraestrutura → Autenticação → Produtos → Preços → Notificações → Relatórios, para evitar retrabalho
- Sistema de exceções personalizadas + handler global para implementar formato de resposta de erro unificado:
{"error": {"code": "...", "message": "..."}} - Princípio Asynchronous-First: Use
async/awaitem toda a cadeia; envolva código síncrono emrun_in_executor - Garantia de Qualidade de Código: ruff (lint + format) + mypy (verificação de tipos) + pytest (testes) + pre-commit (validação automatizada)
- Validação de Entrega do PriceTracker: Milhões de produtos, milhares de requisições concorrentes, notificações push em segundos e formato de erro padronizado
📝 Exercícios
- Exercício Básico (Dificuldade ⭐): Implemente uma hierarquia de exceções personalizada para o PriceTracker (PriceTrackerError → NotFoundError → SubscriptionRequiredError), e substitua
raise HTTPException(404)porraise NotFoundError("product", 42)no endpoint. Dica: Herdar de Exception - Problema Avançado (Dificuldade ⭐⭐): Implemente um handler de exceção global e registre três handlers—PriceTrackerError, RequestValidationError e HTTPException—para garantir que todas as respostas de erro sigam um formato consistente. Dica:
app.add_exception_handler()+format_error_response() - Desafio (Dificuldade: ⭐⭐⭐): Configure uma cadeia de ferramentas de qualidade de código completa—pyproject.toml (com configurações ruff, mypy e pytest), .pre-commit-config.yaml—e execute
ruff check+mypy app/+pytestpara que todos os testes passem. Dica:uv add --dev ruff mypy pytest+pre-commit install
---|



