Exercício Abrangente da Fase 2 — Implementação em Loop Fechado da Funcionalidade Central do PriceTracker

Q: Se get_db é declarado múltiplas vezes em uma cadeia DI, haverá múltiplas conexões?

Não. get_db é executado apenas uma vez por requisição, e todas as dependências compartilham a mesma AsyncSession .

Q: Como testo o fluxo de requisição inteiro?

Substitua get_db com TestClient + dependency_overrides e teste passo a passo: Sem token → 401, Token Free → 200 (mas com restrições), Token Pro → permissões completas.

A Fase 2 é como a montagem final de um carro—o motor (banco de dados), volante (roteamento), cintos de segurança (autenticação) e painel (middleware) já foram construídos, e agora estão sendo montados em um carro completo pronto para hitting the road.

1. O Que Você Vai Aprender

Refinar a cadeia DI de três camadas: get_db → get_current_user → require_subscription("pro")
Proteção JWT Ponta a Ponta: Políticas de Grupo de Rotas para Endpoints Públicos vs. Endpoints Autenticados
Configuração de middleware CORS: Permitir que o domínio frontend do Bob acesse recursos entre domínios
Fluxo de trabalho de teste ponta a ponta para operações CRUD completas, paginação e pré-carregamento de dados associados
Verificação do Cenário da Alice: Usuários Pro podem importar milhões de pontos de dados de preço em lote, enquanto usuários Free são limitados a 1.000 entradas.

2. A História Real da Alice

(1) Problema: Funções estão espalhadas e não podem ser integradas para teste

Alice implementou middleware, injeção de dependência, banco de dados, CRUD e autenticação JWT separadamente, mas cada funcionalidade foi testada independentemente. Quando ela tentou integrar todas essas funcionalidades, descobriu que a ordem de execução do middleware CORS e autenticação JWT conflitavam, a sessão DB foi criada duas vezes na cadeia DI, e a lógica de limitação de taxa e verificações de permissão para usuários Pro sobrepunham-se.

(2) Uma Abordagem Sistemática e Integrada

O Exercício Abrangente da Fase 2 integra todos os módulos na hierarquia e ordem corretas: CORS (camada mais externa) → Limitação de Taxa → Autenticação → Lógica de Negócio → Banco de Dados (camada mais interna), com cada camada tendo responsabilidades claramente definidas e operando independentemente das outras.

(3) Resultado

O serviço central do PriceTracker integrado está totalmente operacional: o frontend do Bob acessou com sucesso recursos entre domínios, todos os endpoints protegidos requerem um token JWT, usuários Pro foram importados em lote sem problemas, e usuários Free estão corretamente limitados por taxa.

3. Visão Geral da Arquitetura da Fase 2

(1) Fluxo Completo de Requisição

flowchart TD
    Client[Bob Frontend] --> CORS[Middleware CORS]
    CORS --> RateLimit[Middleware Limitação]
    RateLimit --> Router[FastAPI Router]
    Router --> AuthDI{Auth Necessária?}
    AuthDI -->|Sim| GetDB[Dependência get_db]
    AuthDI -->|Não| Handler[Handler]
    GetDB --> GetUser[get_current_user]
    GetUser --> CheckSub[require_subscription]
    CheckSub --> Handler
    Handler --> Repo[Camada Repository]
    Repo --> DB[(PostgreSQL)]
    
    RateLimit -.->|429| Reject[Rate Limited]
    GetUser -.->|401| Unauth[Não Autorizado]
    CheckSub -.->|403| Forbid[Proibido]

(2) Política de Grupo de Rotas

Grupo de Rotas	Prefixo	Autenticação	Descrição
public	`/`	Nenhuma	health, docs
api_v1	`/api/v1`	JWT necessário	Todos os endpoints de negócio
admin	`/api/v1/admin`	JWT + Admin	Endpoints Administrativos

4. Integrando o Código Completo

(1) Pontos de Entrada da Aplicação e Configuração do Middleware

(1) ▶ Exemplo: Configuração completa para main.py

PYTHON

from fastapi import FastAPI
from starlette.middleware.cors import CORSMiddleware

from app.api.routes import products, prices, auth
from app.core.config import settings

app = FastAPI(
    title=settings.app_name,
    version="1.0.0",
    description="PriceTracker SaaS API - Serviço de rastreamento de preços para e-commerce",
)

# Ordem do middleware: último registrado = mais interno (executa primeiro na requisição)
# CORS deve ser o mais externo (registrado por último)
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Incluir routers - público primeiro, depois protegidos
app.include_router(auth.router, tags=["auth"])
app.include_router(
    products.router,
    prefix="/api/v1",
    dependencies=[Depends(get_current_user)],
    tags=["products"],
)
app.include_router(
    prices.router,
    prefix="/api/v1",
    dependencies=[Depends(get_current_user)],
    tags=["prices"],
)

@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "pricetracker"}

Saída:

TEXT

# Função definida com sucesso

(2) Implementação Completa de uma Cadeia DI de Três Camadas

(2) ▶ Exemplo: app/core/deps.py

PYTHON

from fastapi import Depends, HTTPException, Header
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from jose import jwt, JWTError

from app.db import async_session
from app.core.config import settings
from app.core.security import verify_password
from app.models import User
from sqlalchemy import select

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")

# Nível 1: Sessão do banco de dados
async def get_db():
    async with async_session() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise

# Nível 2: Usuário atual
async def get_current_user(
    token: str = Depends(oauth2_scheme),
    db: AsyncSession = Depends(get_db),
):
    credentials_exception = HTTPException(
        status_code=401,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(
            token, settings.secret_key, algorithms=[settings.algorithm]
        )
        username = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception

    stmt = select(User).where(User.email == username)
    result = await db.execute(stmt)
    user = result.scalar_one_or_none()
    if user is None:
        raise credentials_exception
    return user

# Nível 3: Verificação de assinatura
def require_subscription(min_level: str = "free"):
    LEVELS = {"free": 0, "pro": 1, "enterprise": 2}

    async def check(user: User = Depends(get_current_user)):
        if LEVELS.get(user.subscription, 0) < LEVELS.get(min_level, 0):
            raise HTTPException(
                status_code=403,
                detail=f"Requires {min_level} plan. Current: {user.subscription}",
            )
        return user
    return check

Saída:

TEXT

# Função definida com sucesso

(3) Processo de Verificação de Permissão

flowchart LR
    Request[Requisição com Token] --> ParseToken[Parse JWT Token]
    ParseToken --> ValidToken{Token Válido?}
    ValidToken -->|Não| Return401[Retornar 401]
    ValidToken -->|Sim| QueryUser[Consultar Usuário do DB]
    QueryUser --> UserExists{Usuário Encontrado?}
    UserExists -->|Não| Return401
    UserExists -->|Sim| CheckSub{Nível de Assinatura?}
    CheckSub -->|Free| AllowBasic[Permitir Endpoints Básicos]
    CheckSub -->|Pro| AllowPro[Permitir Endpoints Pro]
    CheckSub -->|Enterprise| AllowAll[Permitir Todos os Endpoints]

(3) ▶ Exemplo: Rota de Produto Protegida

PYTHON

from fastapi import APIRouter, Depends, HTTPException, Path, Query
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Optional

from app.core.deps import get_db, get_current_user, require_subscription
from app.schemas import ProductCreate, ProductResponse, ProductUpdate
from app.services import ProductService

router = APIRouter()

@router.get("/products", response_model=list[ProductResponse])
async def list_products(
    category: Optional[str] = Query(None),
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
    user=Depends(get_current_user),
):
    service = ProductService(db)
    return await service.list_products(category=category, skip=skip, limit=limit)

@router.post("/products", response_model=ProductResponse, status_code=201)
async def create_product(
    product: ProductCreate,
    db: AsyncSession = Depends(get_db),
    user=Depends(get_current_user),
):
    service = ProductService(db)
    return await service.create_product(product, user_id=user.id)

@router.get("/products/{product_id}", response_model=ProductResponse)
async def get_product(
    product_id: int = Path(gt=0),
    db: AsyncSession = Depends(get_db),
    user=Depends(get_current_user),
):
    service = ProductService(db)
    product = await service.get_product(product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    return product

Saída:

TEXT

# Função definida com sucesso

(4) ▶ Exemplo: Importação em Lote de Endpoints (Incluindo Limites de Assinatura)

PYTHON

@router.post("/products/{product_id}/prices/bulk")
async def bulk_import_prices(
    product_id: int = Path(gt=0),
    prices: list[PriceCreate] = ...,
    db: AsyncSession = Depends(get_db),
    user=Depends(require_subscription("free")),  # Mesmo usuários free podem importar
):
    # Plano Free: máximo 1000 preços por importação
    limits = {"free": 1000, "pro": 100000, "enterprise": 1000000}
    max_import = limits.get(user.subscription, 1000)
    
    if len(prices) > max_import:
        raise HTTPException(
            status_code=403,
            detail=f"Import limit: {max_import} for {user.subscription} plan. Upgrade at /pricing",
        )
    
    service = PriceService(db)
    return await service.bulk_import_prices(product_id, prices)

Saída:

TEXT

# Função definida com sucesso

❓ Perguntas Frequentes

P Por que CORS é registrado por último na sequência de middleware?

R CORS precisa adicionar cabeçalhos CORS a todas as respostas (incluindo erros 4xx). Se CORS não for a camada mais externa, a resposta 429 retornada pelo middleware de limitação de taxa pode carecer do cabeçalho CORS, impedindo o frontend de ler a mensagem de erro.

P Dependências de grupo de rotas e dependências de endpoint são executadas múltiplas vezes?

R Não. O FastAPI armazena em cache a mesma dependência em nível de requisição. O Depends(get_current_user) de um grupo de rotas e o Depends(get_current_user) de um endpoint compartilham o mesmo resultado de execução.

P Se get_db é declarado múltiplas vezes em uma cadeia DI, haverá múltiplas conexões?

R Não. get_db é executado apenas uma vez por requisição, e todas as dependências compartilham a mesma AsyncSession.

P Como usuários Free podem fazer upgrade para Pro?

R Integração de pagamento Stripe (além do escopo desta aula). Aqui, usamos o campo User.subscription para simulação; em um ambiente de produção, isso é atualizado via callback de pagamento.

P Como testo o fluxo de requisição inteiro?

R Substitua get_db com TestClient + dependency_overrides e teste passo a passo: Sem token → 401, Token Free → 200 (mas com restrições), Token Pro → permissões completas.

P O que vem depois que a Fase 2 estiver completa?

R A Fase 3 introduz recursos avançados—push em tempo real WebSocket, tarefas assíncronas Celery, cache Redis, upload de arquivos, testes pytest e personalização OpenAPI.

📖 Resumo

Fase 2 integra cinco módulos principais—middleware, DI, banco de dados, CRUD e JWT—em um serviço completo e totalmente funcional
Uma cadeia DI de três camadas (get_db → get_current_user → require_subscription) gerencia centralmente autenticação e autorização
Políticas de grupo de rotas: "public" não requer autenticação; "api_v1" é protegido por JWT; "admin" requer JWT e permissões de Admin
CORS deve ser registrado no nível mais externo para garantir que todas as respostas (incluindo respostas de erro) incluam cabeçalhos CORS
Níveis de assinatura são implementados via cadeia DI: Free é limitado a 1.000 importações, Pro a 100.000, e Enterprise a 1.000.000.

📝 Exercícios

Problema Básico (Dificuldade ⭐): Substitua a implementação CRUD em memória da Fase 1 por uma implementação SQLAlchemy assíncrona, garantindo que todos os endpoints injetem a sessão do banco de dados via Depends(get_db). Dica: Padrão Repository + AsyncSession
Exercício Avançado (Dificuldade ⭐⭐): Implemente grupos de rotas—o grupo "public" (/health, /auth/login) não requer autenticação, enquanto todos os endpoints de negócio sob "api/v1" são protegidos por JWT. Use Swagger UI para testar login e acesso a endpoints protegidos. Dica: APIRouter(dependencies=[Depends(get_current_user)])
Desafio (Dificuldade: ⭐⭐⭐): Implemente uma cadeia DI de três camadas completa + importação em lote de endpoints: Usuários Pro podem importar 100.000 entradas de preço; Usuários Free receberão um erro 403 se excederem 1.000 entradas. Também demonstre completamente o processo registro → login → importação no Swagger UI. Dica: require_subscription("free") + dicionário limits

---|

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.

Sobre Nós

404 Not Found

Exercício Abrangente da Fase 2 — Implementação em Loop Fechado da Funcionalidade Central do PriceTracker

1. O Que Você Vai Aprender

2. A História Real da Alice

(1) Problema: Funções estão espalhadas e não podem ser integradas para teste

(2) Uma Abordagem Sistemática e Integrada

(3) Resultado

3. Visão Geral da Arquitetura da Fase 2

(1) Fluxo Completo de Requisição

(2) Política de Grupo de Rotas

4. Integrando o Código Completo

(1) Pontos de Entrada da Aplicação e Configuração do Middleware

(1) ▶ Exemplo: Configuração completa para main.py

(2) Implementação Completa de uma Cadeia DI de Três Camadas

(2) ▶ Exemplo: app/core/deps.py

(3) Processo de Verificação de Permissão

(3) ▶ Exemplo: Rota de Produto Protegida

(4) ▶ Exemplo: Importação em Lote de Endpoints (Incluindo Limites de Assinatura)

❓ Perguntas Frequentes

📖 Resumo

📝 Exercícios