Implementação Completa de CRUD — Do Armazenamento de Dados ao Endpoint da API

Q: O Repository obrigatoriamente deve ser uma classe?

Não, não precisa ser; você também pode usar uma função. No entanto, uma classe encapsula o estado da sessão DB, tornando mais fácil compartilhar a sessão entre métodos, então a abordagem de classe é recomendada.

Q: Qual é a diferença entre flush e commit ?

flush envia SQL ao banco de dados mas não faz commit da transação; pode recuperar IDs auto-incrementados e acionar verificações de restrição do banco. commit faz commit da transação, tornando as mudanças persistentes. Em dependências yield , recomendamos usar flush seguido de commit automático.

Q: O problema N+1 só ocorre dentro de loops?

Ocorre principalmente ao iterar sobre atributos associados. No entanto, também pode ser acionado pela serialização Pydantic—se response_model acessa campos associados. Certifique-se de pré-carregar os dados.

Q: Há um limite de tamanho para bulk_create ?

Não há limite imposto pelo framework, mas o PostgreSQL limita o número de parâmetros vinculados por operação a 32.767. Inserções na faixa de milhares não são problema, mas para dezenas de milhares, recomendamos dividir em lotes.

Q: Quão lenta é a paginação offset com grandes conjuntos de dados?

OFFSET 1000000 primeiro precisa escanear o primeiro 1 milhão de linhas antes de pulá-las; o tempo de consulta é proporcional ao offset. Para conjuntos de dados na faixa de milhões, recomendamos usar paginação por cursor.

Q: Como lidar com falhas parciais dentro de uma transação?

O rollback associado ao yield é acionado automaticamente quando uma exceção ocorre. Se você precisa de um rollback parcial, use um savepoint ( await session.begin_nested() ).

1. O Que Você Vai Aprender

Repository Pattern: Encapsulamento ProductRepository/PriceRepository
Paginação e Ordenação: Os modelos de consulta Pydantic para limit/offset e order_by
Pré-carregamento de Dados Relacionados: selectinload vs joinedload—Compromissos de Desempenho
Gerenciamento de Transações: Garantindo Consistência em Operações Cross-Table
Cenário da Alice: API de Importação em Lote de Preços—Processamento Transacional para Inserir 1.000 Registros de Preço em uma Única Requisição

2. A História Real da Alice

(1) Problema: SQL espalhado pelas funções de roteamento

No código do PriceTracker da Alice, consultas SQL são escritas diretamente dentro das funções handler de rota, e há 30 instâncias de lógica duplicada select(Product).where(...) em 20 endpoints. Quando ela precisou adicionar uma condição de "filtro de soft delete" a todas as consultas de produtos, Alice teve que fazer mudanças em 15 lugares; perder apenas duas delas fez com que produtos deletados aparecessem nos resultados.

(2) Soluções com o Modelo de Repositório

Encapsule operações de banco de dados dentro da classe Repository; funções de rota devem apenas chamar métodos do Repository. Para adicionar um filtro global, basta fazer uma única mudança no Repository.

PYTHON

class ProductRepository:
    async def get_all(self, db: AsyncSession, filters: ProductFilter) -> list[Product]:
        stmt = select(Product).where(Product.deleted == False)
        # Adicionar filtros do modelo Pydantic
        if filters.category:
            stmt = stmt.where(Product.category == filters.category)
        result = await db.execute(stmt)
        return result.scalars().all()

(3) Resultado

A lógica SQL está centralizada no Repository, tornando a função de roteamento mais concisa (5 linhas em vez de 15). Condições de filtro globais podem ser modificadas em um único lugar para surtir efeito em todo o sistema, então Bob no front end não verá mais produtos deletados.

3. Arquitetura em Camadas e Modelos de Armazém de Dados

(1) Arquitetura de quatro camadas

flowchart TD
    Router[Camada Router] -->|Chamar| Service[Camada Service]
    Service -->|Chamar| Repository[Camada Repository]
    Repository -->|Consultar| SQLAlchemy[SQLAlchemy ORM]
    SQLAlchemy -->|SQL| Database[(PostgreSQL)]
    
    style Router fill:#e3f2fd
    style Service fill:#fff3e0
    style Repository fill:#e8f5e9
    style SQLAlchemy fill:#fce4ec

Nível	Responsabilidades	Exemplos
Router	Validação de Parâmetros, Formato de Resposta	`@app.get("/products")`
Service	Lógica de Negócios, Orquestração de Transações	`ProductService.create_with_price()`
Repository	Acesso a Dados, Wrappers SQL	`ProductRepository.get_by_id()`
ORM	Mapeamento Objeto-Relacional	`select(Product).where(...)`

(1) ▶ Exemplo: ProductRepository

PYTHON

from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession

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

    async def get_by_id(self, product_id: int) -> Product | None:
        stmt = select(Product).where(Product.id == product_id)
        result = await self.db.execute(stmt)
        return result.scalar_one_or_none()

    async def get_all(
        self,
        category: str | None = None,
        skip: int = 0,
        limit: int = 20,
    ) -> list[Product]:
        stmt = select(Product).offset(skip).limit(limit)
        if category:
            stmt = stmt.where(Product.category == category)
        result = await self.db.execute(stmt)
        return list(result.scalars().all())

    async def create(self, product: ProductCreate) -> Product:
        db_product = Product(**product.model_dump())
        self.db.add(db_product)
        await self.db.flush()
        await self.db.refresh(db_product)
        return db_product

    async def update(self, product_id: int, data: dict) -> Product | None:
        stmt = (
            update(Product)
            .where(Product.id == product_id)
            .values(**data)
            .returning(Product)
        )
        result = await self.db.execute(stmt)
        return result.scalar_one_or_none()

    async def delete(self, product_id: int) -> bool:
        stmt = delete(Product).where(Product.id == product_id)
        result = await self.db.execute(stmt)
        return result.rowcount > 0

TEXT

# Função definida com sucesso

(2) ▶ Exemplo: PriceRepository

PYTHON

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

    async def get_by_product(
        self,
        product_id: int,
        min_price: float | None = None,
        max_price: float | None = None,
        skip: int = 0,
        limit: int = 50,
    ) -> list[Price]:
        stmt = (
            select(Price)
            .where(Price.product_id == product_id)
            .order_by(Price.recorded_at.desc())
            .offset(skip)
            .limit(limit)
        )
        if min_price is not None:
            stmt = stmt.where(Price.price >= min_price)
        if max_price is not None:
            stmt = stmt.where(Price.price <= max_price)
        result = await self.db.execute(stmt)
        return list(result.scalars().all())

    async def bulk_create(self, prices: list[PriceCreate]) -> list[Price]:
        db_prices = [Price(**p.model_dump()) for p in prices]
        self.db.add_all(db_prices)
        await self.db.flush()
        return db_prices

TEXT

# Função definida com sucesso

4. Paginação e Ordenação

(1) Modelo de Consulta Pydantic

Encapsule parâmetros de paginação em um modelo Pydantic para evitar ter que declarar parâmetros de query repetidamente em cada endpoint.

(1) ▶ Exemplo: Modelo de Consulta Paginada

PYTHON

from pydantic import BaseModel, Field
from typing import Optional

class PaginationParams(BaseModel):
    skip: int = Field(0, ge=0, description="Número de registros para pular")
    limit: int = Field(20, ge=1, le=100, description="Máx de registros para retornar")

class ProductFilter(PaginationParams):
    category: Optional[str] = Field(None, max_length=100)
    min_price: Optional[float] = Field(None, ge=0, description="Preço mín em USD")
    max_price: Optional[float] = Field(None, ge=0, description="Preço máx em USD")
    sort_by: str = Field("name", pattern="^(name|base_price|created_at)$")
    sort_order: str = Field("asc", pattern="^(asc|desc)$")

TEXT

# Execução Bem-sucedida

(2) ▶ Exemplo: Endpoints Usando o Modelo de Paginação

PYTHON

from fastapi import FastAPI, Depends, Query
from sqlalchemy.ext.asyncio import AsyncSession

app = FastAPI()

@app.get("/products", response_model=list[ProductResponse])
async def list_products(
    category: Optional[str] = Query(None),
    min_price: Optional[float] = Query(None, ge=0),
    max_price: Optional[float] = Query(None, ge=0),
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
):
    repo = ProductRepository(db)
    return await repo.get_all(
        category=category,
        min_price=min_price,
        max_price=max_price,
        skip=skip,
        limit=limit,
    )

TEXT

# Função definida com sucesso

(2) Comparação de Estratégias de Paginação

Estratégia	Implementação	Vantagens	Desvantagens
Paginação Offset	`OFFSET n LIMIT m`	Simples, suporta salto de página	Desempenho ruim com offsets grandes
Paginação por Cursor	`WHERE id > cursor LIMIT m`	Bom desempenho com grandes conjuntos de dados	Não suporta salto de página
Paginação Keyset	`WHERE created_at < last LIMIT m`	Paginação por campo de ordenação funciona bem	Requer um campo ordenado

5. Pré-carregamento de Dados Relacionados

(1) O Problema N+1

(1) ▶ Exemplo: Demonstração do Problema N+1

PYTHON

# RUIM: consultas N+1 - 1 consulta para produtos + N consultas para preços
stmt = select(Product)
result = await db.execute(stmt)
products = result.scalars().all()
for p in products:
    # Cada acesso aciona uma consulta separada!
    print(len(p.prices))  # N consultas

TEXT

# Execução Bem-sucedida

(2) ▶ Exemplo: Pré-carregamento com selectinload

PYTHON

from sqlalchemy.orm import selectinload, joinedload

# BOM: 2 consultas no total - produtos + preços (usando SELECT IN)
stmt = select(Product).options(selectinload(Product.prices))
result = await db.execute(stmt)
products = result.scalars().all()
for p in products:
    print(len(p.prices))  # Sem consultas extras

TEXT

# Execução Bem-sucedida

(2) selectinload vs joinedload

Dimensão	selectinload	joinedload
Estratégia SQL	SELECT ... WHERE id IN (...)	LEFT OUTER JOIN
Número de consultas	2 consultas	1 consulta
Volume de Dados	Preciso (sem linhas duplicadas)	JOIN pode produzir linhas duplicadas
Casos de Uso	Um-para-Muitos (Conjunto)	Muitos-para-Um/Um-para-Um
Desempenho	Melhor para grandes conjuntos de dados	Melhor para conjuntos com poucas relações
Remover duplicatas	Não necessário	Necessário `unique()`

(3) ▶ Exemplo: Consulta de produto com preços pré-carregados

PYTHON

async def get_product_with_prices(
    self, product_id: int
) -> Product | None:
    stmt = (
        select(Product)
        .options(selectinload(Product.prices))
        .where(Product.id == product_id)
    )
    result = await self.db.execute(stmt)
    return result.scalar_one_or_none()

TEXT

# Função definida com sucesso

6. Gerenciamento de Transações

(1) Transações Cross-Table

(1) ▶ Exemplo: Transação de Importação em Lote de Preços

PYTHON

from sqlalchemy.ext.asyncio import AsyncSession

class PriceService:
    def __init__(self, db: AsyncSession):
        self.db = db
        self.price_repo = PriceRepository(db)
        self.product_repo = ProductRepository(db)

    async def bulk_import_prices(
        self,
        product_id: int,
        prices: list[PriceCreate],
    ) -> list[Price]:
        # Verificar se o produto existe
        product = await self.product_repo.get_by_id(product_id)
        if not product:
            raise HTTPException(status_code=404, detail="Product not found")

        # Validar que todos os preços referenciam o mesmo produto
        for p in prices:
            p.product_id = product_id

        # Inserção em lote dentro da transação (sessão db gerencia commit/rollback)
        db_prices = await self.price_repo.bulk_create(prices)

        # Atualizar preço base do produto para o mais recente
        latest_price = db_prices[-1]
        await self.product_repo.update(
            product_id,
            {"base_price": latest_price.price},
        )

        return db_prices

TEXT

# Função definida com sucesso

(2) ▶ Exemplo: Endpoint de Importação em Lote

PYTHON

@app.post("/products/{product_id}/prices/bulk", response_model=list[PriceResponse])
async def bulk_import(
    product_id: int = Path(gt=0),
    prices: list[PriceCreate] = ...,
    db: AsyncSession = Depends(get_db),
    user: dict = Depends(get_current_user),
):
    # Verificação de assinatura: Usuários Free limitados a 1000 preços
    if user["subscription"] == "free" and len(prices) > 1000:
        raise HTTPException(
            status_code=403,
            detail="Free plan limited to 1000 prices per import. Upgrade to Pro.",
        )
    
    service = PriceService(db)
    return await service.bulk_import_prices(product_id, prices)

TEXT

# Função definida com sucesso

❓ Perguntas Frequentes

P O Repository obrigatoriamente deve ser uma classe?

R Não, não precisa ser; você também pode usar uma função. No entanto, uma classe encapsula o estado da sessão DB, tornando mais fácil compartilhar a sessão entre métodos, então a abordagem de classe é recomendada.

P Qual é a diferença entre flush e commit?

R flush envia SQL ao banco de dados mas não faz commit da transação; pode recuperar IDs auto-incrementados e acionar verificações de restrição do banco. commit faz commit da transação, tornando as mudanças persistentes. Em dependências yield, recomendamos usar flush seguido de commit automático.

P O problema N+1 só ocorre dentro de loops?

R Ocorre principalmente ao iterar sobre atributos associados. No entanto, também pode ser acionado pela serialização Pydantic—se response_model acessa campos associados. Certifique-se de pré-carregar os dados.

P Há um limite de tamanho para bulk_create?

R Não há limite imposto pelo framework, mas o PostgreSQL limita o número de parâmetros vinculados por operação a 32.767. Inserções na faixa de milhares não são problema, mas para dezenas de milhares, recomendamos dividir em lotes.

P Quão lenta é a paginação offset com grandes conjuntos de dados?

R OFFSET 1000000 primeiro precisa escanear o primeiro 1 milhão de linhas antes de pulá-las; o tempo de consulta é proporcional ao offset. Para conjuntos de dados na faixa de milhões, recomendamos usar paginação por cursor.

P Como lidar com falhas parciais dentro de uma transação?

R O rollback associado ao yield é acionado automaticamente quando uma exceção ocorre. Se você precisa de um rollback parcial, use um savepoint (await session.begin_nested()).

📖 Resumo

O padrão repository encapsula operações SQL dentro da classe Repository, para que funções de roteamento se concentrem apenas na lógica HTTP
Parâmetros de paginação são encapsulados como um modelo Pydantic para evitar ter que declarar parâmetros de query repetidamente para cada endpoint
selectinload é adequado para pré-carregamento um-para-muitos (sem linhas duplicadas); joinedload é adequado para muitos-para-um (consulta única)
Operações em lote são executadas dentro de uma transação; yield depende do gerenciamento automático de commit/rollback
A importação em lote de preços do PriceTracker suporta milhares de registros; Usuários Free são limitados a 1.000 registros

📝 Exercícios

Problema Básico (Dificuldade ⭐): Implemente os métodos get_by_id e get_all no ProductRepository, injete a Sessão DB nos endpoints, e chame-os. Dica: select(Product).where(Product.id == product_id)
Exercício Avançado (Dificuldade: ⭐⭐): Adicione suporte para consultas paginadas (usando os parâmetros skip e limit), implemente pré-carregamento selectinload(Product.prices), e retorne detalhes do produto que incluem preços. Dica: options(selectinload(...))
Desafio (Dificuldade ⭐⭐⭐): Implemente PriceService.bulk_import_prices, verifique existência do produto, insira preços em lote, e atualize o base_price do produto—tudo em uma única transação. Usuários Free são limitados a 1.000 registros. Dica: bulk_create + update + Depends(get_current_user)

---|

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

Implementação Completa de CRUD — Do Armazenamento de Dados ao Endpoint da API

1. O Que Você Vai Aprender

2. A História Real da Alice

(1) Problema: SQL espalhado pelas funções de roteamento

(2) Soluções com o Modelo de Repositório

(3) Resultado

3. Arquitetura em Camadas e Modelos de Armazém de Dados

(1) Arquitetura de quatro camadas

(1) ▶ Exemplo: ProductRepository

(2) ▶ Exemplo: PriceRepository

4. Paginação e Ordenação

(1) Modelo de Consulta Pydantic

(1) ▶ Exemplo: Modelo de Consulta Paginada

(2) ▶ Exemplo: Endpoints Usando o Modelo de Paginação

(2) Comparação de Estratégias de Paginação

5. Pré-carregamento de Dados Relacionados

(1) O Problema N+1

(1) ▶ Exemplo: Demonstração do Problema N+1

(2) ▶ Exemplo: Pré-carregamento com selectinload

(2) selectinload vs joinedload

(3) ▶ Exemplo: Consulta de produto com preços pré-carregados

6. Gerenciamento de Transações

(1) Transações Cross-Table

(1) ▶ Exemplo: Transação de Importação em Lote de Preços

(2) ▶ Exemplo: Endpoint de Importação em Lote

❓ Perguntas Frequentes

📖 Resumo

📝 Exercícios