Integração de Banco de Dados — SQLAlchemy Assíncrono + Alembic

Q: Há diferença entre Mapped[str | None] e Mapped[Optional[str]] ?

São funcionalmente equivalentes. str | None é a sintaxe do Python 3.10+, enquanto Optional[str] é a notação compatível do typing . A primeira é recomendada.

Um banco de dados é como um armazém—se houver poucas portas (conexões), as mercadorias (consultas) se acumulam; se o layout (índices) é desorganizado, encontrar um único item (dado) requer vasculhar o armazém inteiro.

1. O Que Você Vai Aprender

Engine Assíncrona do SQLAlchemy 2.0: create_async_engine, AsyncSession
Declaração de Modelo ORM: DeclarativeBase, Mapped, mapped_column (Novo Estilo 2.0)
Migração Assíncrona com Alembic: Configurar env.py para o modo assíncrono run_migrations_online
Gerenciamento de Sessão Assíncrona: Padrões de Injeção de Dependência async with / yield
Cenário da Alice: Design de Modelo Assíncrono para as Três Tabelas products/prices/users do PriceTracker

2. A História Real da Alice

(1) Problema: Banco de Dados Síncrono Desacelera APIs Assíncronas

Alice usa SQLAlchemy síncrono para interagir com o banco de dados, e cada consulta bloqueia o event loop por 50-100 ms. Quando o número de requisições concorrentes ao PriceTracker chegou a 500, os benefícios do FastAPI assíncrono foram completamente compensados pelas operações síncronas de banco de dados, e a latência P99 disparou dos 50 ms esperados para 2.000 ms. Charlie observou que o pool de conexões do banco de dados tinha sido esgotado, e novas requisições estavam enfileiradas aguardando por conexões.

(2) Solução com SQLAlchemy Assíncrono

O SQLAlchemy 2.0 fornece suporte assíncrono nativo: create_async_engine + AsyncSession. Consultas de banco de dados não bloqueiam mais o event loop, permitindo que o FastAPI aproveite totalmente suas vantagens assíncronas.

PYTHON

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async_session = async_sessionmaker(engine, expire_on_commit=False)

async def get_db():
    async with async_session() as session:
        yield session

(3) Resultado

Após migrar para SQLAlchemy assíncrono, a latência P99 do PriceTracker caiu de 2.000 ms para 80 ms, a utilização da conexão do banco de dados aumentou de 30% para 90%, e o QPS de nó único subiu de 500 para mais de 3.000.

3. Engines e Sessões Assíncronas

(1) Diagrama ER da Arquitetura do Banco de Dados

erDiagram
    users ||--o{ products : creates
    products ||--o{ prices : has
    users {
        int id PK
        string email UK
        string hashed_password
        string role
        string subscription
        datetime created_at
    }
    products {
        int id PK
        string name
        string category
        float base_price
        string description
        int user_id FK
        datetime created_at
    }
    prices {
        int id PK
        int product_id FK
        float price
        string currency
        string source
        datetime recorded_at
    }

(1) ▶ Exemplo: Engine Assíncrona e Configuração do Pool de Conexões

PYTHON

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

DATABASE_URL = "postgresql+asyncpg://pricetracker:secret@localhost:5432/pricetracker"

engine = create_async_engine(
    DATABASE_URL,
    echo=False,          # Definir True para log SQL em dev
    pool_size=20,        # Conexões persistentes
    max_overflow=10,     # Conexões extras quando pool esgotado
    pool_timeout=30,     # Tempo de espera por conexão disponível
    pool_recycle=3600,   # Reciclar conexões após 1 hora
)

async_session = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,  # Acessar objetos após commit
)

Saída:

TEXT

# Execução Bem-sucedida

(2) SQLAlchemy Síncrono vs. Assíncrono: Uma Comparação

Dimensão	SQLAlchemy Síncrono	SQLAlchemy 2.0 Assíncrono
Engine	`create_engine`	`create_async_engine`
Sessão	`Session`	`AsyncSession`
Busca	`session.execute(stmt)`	`await session.execute(stmt)`
Commit	`session.commit()`	`await session.commit()`
Driver	psycopg2	asyncpg
Bloqueante	Sim	Não
Pool de Conexões	QueuePool	AsyncAdaptedQueuePool

4. Declaração de Modelo ORM (Novo Estilo 2.0)

(1) DeclarativeBase + Mapped + mapped_column

O SQLAlchemy 2.0 substitui a declaração Column() antiga por Mapped[type] e mapped_column(), unificando type hints com mapeamentos ORM.

(1) ▶ Exemplo: Modelo de Três Tabelas do PriceTracker

PYTHON

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Float, Integer, DateTime, ForeignKey, Index
from sqlalchemy.orm import relationship
from datetime import datetime

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    role: Mapped[str] = mapped_column(String(50), default="user")
    subscription: Mapped[str] = mapped_column(String(50), default="free")
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    products: Mapped[list["Product"]] = relationship(back_populates="owner")

class Product(Base):
    __tablename__ = "products"
    __table_args__ = (
        Index("ix_products_category", "category"),
        Index("ix_products_name", "name"),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(200), nullable=False)
    category: Mapped[str] = mapped_column(String(100), nullable=False)
    base_price: Mapped[float] = mapped_column(Float, nullable=False)
    description: Mapped[str | None] = mapped_column(String(2000), nullable=True)
    user_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"))
    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    owner: Mapped["User"] = relationship(back_populates="products")
    prices: Mapped[list["Price"]] = relationship(back_populates="product")

class Price(Base):
    __tablename__ = "prices"
    __table_args__ = (
        Index("ix_prices_product_id", "product_id"),
        Index("ix_prices_recorded_at", "recorded_at"),
    )

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
    product_id: Mapped[int] = mapped_column(Integer, ForeignKey("products.id"))
    price: Mapped[float] = mapped_column(Float, nullable=False)
    currency: Mapped[str] = mapped_column(String(3), default="USD")
    source: Mapped[str] = mapped_column(String(100))
    recorded_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)

    product: Mapped["Product"] = relationship(back_populates="prices")

Saída:

TEXT

# Execução Bem-sucedida

(2) Estilo Antigo V1 vs. Novo Estilo V2

Dimensão	Estilo Antigo V1	Novo Estilo V2
Base	`declarative_base()`	`class Base(DeclarativeBase)`
Declaração de Campo	`Column(Integer, primary_key=True)`	`Mapped[int] = mapped_column(...)`
Type hint	Nenhum	`Mapped[type]` tipo completo
Campos Opcionais	`Column(String, nullable=True)`	`Mapped[str \| None]`
Relacionamento	`relationship()`	`Mapped[list["X"]] = relationship()`

5. Gerenciamento de Sessão Assíncrona e Injeção de Dependência

(1) O Padrão de Dependência "yield"

(1) ▶ Exemplo: Dependência de Sessão DB

PYTHON

from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

async def get_db() -> AsyncGenerator[AsyncSession, None]:
    async with async_session() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

Saída:

TEXT

# Função definida com sucesso

(2) ▶ Exemplo: Usando Sessões Assíncronas em Endpoints

PYTHON

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

app = FastAPI()

@app.get("/products/{product_id}")
async def get_product(
    product_id: int,
    db: AsyncSession = Depends(get_db),
):
    stmt = select(Product).where(Product.id == product_id)
    result = await db.execute(stmt)
    product = result.scalar_one_or_none()
    if not product:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="Product not found")
    return {
        "id": product.id,
        "name": product.name,
        "base_price": product.base_price,
    }

Saída:

TEXT

# Função definida com sucesso

6. Migrações Assíncronas com Alembic

(1) Fluxo de Trabalho de Migração

flowchart LR
    A[alembic revision --autogenerate -m desc] --> B[Editar Arquivo de Migração]
    B --> C[alembic upgrade head]
    C --> D[Aplicar ao Banco de Dados]
    D --> E{Precisa de Rollback?}
    E -->|Sim| F[alembic downgrade -1]
    E -->|Não| G[Continuar Desenvolvimento]

(1) ▶ Exemplo: Inicializando o Alembic

BASH

# Instalar Alembic
uv add alembic

# Inicializar Alembic com template assíncrono
cd pricetracker
alembic init -t async alembic

Saída:

TEXT

# Comando executado com sucesso

(2) ▶ Exemplo: Configurando modo assíncrono no alembic/env.py

PYTHON

# alembic/env.py (seções principais)
from sqlalchemy.ext.asyncio import create_async_engine
from app.models import Base  # Importe seus modelos
from app.core.config import settings

target_metadata = Base.metadata

def run_migrations_online():
    connectable = create_async_engine(settings.database_url)

    async def do_run_migrations(connection):
        context = MigrationContext.configure(
            connection=connection,
            target_metadata=target_metadata,
        )
        with context.begin_transaction():
            context.run_migrations()

    with connectable.connect() as connection:
        asyncio.run(do_run_migrations(connection))

Saída:

TEXT

# Função definida com sucesso

(3) ▶ Exemplo: Gerando e Aplicando Migração

BASH

# Auto-gerar migração a partir de mudanças no modelo
alembic revision --autogenerate -m "add users products prices tables"

# Aplicar migração
alembic upgrade head

# Rollback de um passo
alembic downgrade -1

# Verificar versão atual
alembic current

Saída:

TEXT

# Comando executado com sucesso

❓ Perguntas Frequentes

P Qual é a diferença entre asyncpg e psycopg?

R asyncpg é um driver PostgreSQL puramente assíncrono que oferece melhor desempenho; psycopg3 suporta operações assíncronas mas é baseado em uma biblioteca C. asyncpg é recomendado para uso em produção (postgresql+asyncpg://).

P O que expire_on_commit=False significa?

R Por padrão, as propriedades do objeto ORM expiram após um commit (e acessá-las novamente aciona uma consulta). Definir isso como False mantém as propriedades acessíveis, prevenindo problemas de lazy-loading.

P O recurso autogenerate do Alembic detecta todas as mudanças?

R Não. Ele detecta adição ou remoção de tabelas e colunas, bem como mudanças em índices. Não detecta renomeação de colunas, mudanças na semântica de restrições e outras mudanças similares; estas requerem edição manual dos arquivos de migração.

P Como deve ser definido o tamanho do pool de conexões?

R Fórmula: pool_size = (núcleos de CPU * 2) + número de discos ativos. O PriceTracker usa pool_size=20 + max_overflow=10 para lidar com milhões de consultas.

P Há diferença entre Mapped[str | None] e Mapped[Optional[str]]?

R São funcionalmente equivalentes. str | None é a sintaxe do Python 3.10+, enquanto Optional[str] é a notação compatível do typing. A primeira é recomendada.

P Código SQLAlchemy síncrono pode ser usado em um contexto assíncrono?

R Você não pode chamar diretamente métodos síncronos do SQLAlchemy dentro de uma função async. Envolva-os usando run_in_executor ou use exclusivamente a API assíncrona.

📖 Resumo

A Engine Assíncrona do SQLAlchemy 2.0 (create_async_engine + AsyncSession) possui um event loop não-bloqueante que aproveita totalmente as vantagens assíncronas do FastAPI
Mapped[type] + mapped_column() é o novo estilo 2.0, que unifica type hints com mapeamentos ORM
O padrão de dependência yield gerencia o ciclo de vida de sessões assíncronas: commit, rollback e close automáticos
Migrações assíncronas do Alembic são inicializadas usando o template -t async, e o template env.py configura a engine assíncrona
O modelo de três tabelas do PriceTracker (users/products/prices) inclui estratégia de indexação e suporta consultas em milhões de registros

📝 Exercícios

Exercício Básico (Dificuldade ⭐): Configure create_async_engine para conectar ao PostgreSQL, crie async_sessionmaker, e escreva a dependência yield get_db. Dica: create_async_engine(DATABASE_URL, pool_size=20)
Exercício Avançado (Dificuldade ⭐⭐): Defina os modelos Product e Price para o PriceTracker (usando o estilo DeclarativeBase + Mapped), incluindo relações de chave estrangeira e índices. Dica: ForeignKey("products.id") + relationship()
Desafio (Dificuldade: ⭐⭐⭐): Inicialize o Alembic em modo assíncrono, configure env.py, use autogenerate para gerar arquivos de migração para três tabelas, e aplique-os ao banco de dados usando upgrade head. Dica: alembic init -t async alembic + Modificar target_metadata do env.py

---|

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

Integração de Banco de Dados — SQLAlchemy Assíncrono + Alembic

1. O Que Você Vai Aprender

2. A História Real da Alice

(1) Problema: Banco de Dados Síncrono Desacelera APIs Assíncronas

(2) Solução com SQLAlchemy Assíncrono

(3) Resultado

3. Engines e Sessões Assíncronas

(1) Diagrama ER da Arquitetura do Banco de Dados

(1) ▶ Exemplo: Engine Assíncrona e Configuração do Pool de Conexões

(2) SQLAlchemy Síncrono vs. Assíncrono: Uma Comparação

4. Declaração de Modelo ORM (Novo Estilo 2.0)

(1) DeclarativeBase + Mapped + mapped_column

(1) ▶ Exemplo: Modelo de Três Tabelas do PriceTracker

(2) Estilo Antigo V1 vs. Novo Estilo V2

5. Gerenciamento de Sessão Assíncrona e Injeção de Dependência

(1) O Padrão de Dependência "yield"

(1) ▶ Exemplo: Dependência de Sessão DB

(2) ▶ Exemplo: Usando Sessões Assíncronas em Endpoints

6. Migrações Assíncronas com Alembic

(1) Fluxo de Trabalho de Migração

(1) ▶ Exemplo: Inicializando o Alembic

(2) ▶ Exemplo: Configurando modo assíncrono no alembic/env.py

(3) ▶ Exemplo: Gerando e Aplicando Migração

❓ Perguntas Frequentes

📖 Resumo

📝 Exercícios