Aula 19 - ORM SQLAlchemy 2.0 Assíncrono e Migrações Alembic 🗃️

📑 1. Fundamentos Teóricos & Análise Técnica

O SQLAlchemy 2.0 consolidou uma reformulação completa da biblioteca mais prestigiada de bancos de dados em Python. Ele aposentou o padrão legado de queries baseadas em encadeamento imperativo (session.query(Model).filter(...)) em favor de uma sintaxe puramente declarativa e alinhada ao SQL padrão baseada na função select().

O suporte a I/O Assíncrono (AsyncEngine e AsyncSession) permite integrar bancos relacionais (PostgreSQL com driver asyncpg ou MySQL com aiomysql) de forma não-bloqueante ao loop de eventos do FastAPI.

O gerenciamento de esquema é orquestrado pelo Alembic: 1. O Alembic compara as classes declarativas do SQLAlchemy (Mapped[...]) com o estado real das tabelas na base de dados. 2. O comando alembic revision --autogenerate -m "mensagem" inspeciona as diferenças e produz um arquivo de migração Python contendo as funções upgrade() e downgrade(). 3. As migrações são executadas sequencialmente no deploy, garantindo total reprodutibilidade entre ambientes de desenvolvimento, homologação e produção.

📐 Arquitetura Conceitual & Diagrama de Fluxo

flowchart TD
    Models["Modelos Python (SQLAlchemy DeclarativeBase)"] --> Alembic["Alembic Autogenerate"]
    DB["Banco de Dados PostgreSQL"] --> Alembic
    Alembic --> MigrationFile["Arquivo de Migração (versions/xxx_create_tables.py)"]
    MigrationFile --> Apply["alembic upgrade head"]
    Apply --> UpdatedDB["Banco de Dados Sincronizado e Versionado"]
    style Models fill:#e1f5fe,stroke:#01579b
    style Alembic fill:#fff3e0,stroke:#e65100
    style UpdatedDB fill:#e8f5e9,stroke:#2e7d32

🔍 Pilares e Diretrizes Técnicas

Nesta unidade, aprofundamos os seguintes conceitos fundamentais: - Sintaxe Moderna com Mapped e mapped_column: Tipagem estrita totalmente compatível com mypy e analisadores estáticos. - Driver Assíncrono asyncpg: Comunicação em nível de protocolo binário com o PostgreSQL com taxa de transferência estelar. - Carregamento Explícito de Relações: Prevenção ativa contra o problema das N+1 consultas com selectinload ou joinedload. - Migrações Reversíveis: Garantia de que toda alteração possua um caminho seguro de desfazimento (downgrade).

🛠️ 2. Implementação Prática em Python, SQLAlchemy 2.0 e Alembic

Abaixo está a implementação técnica de referência, estruturada com padrões de engenharia de software e foco em robustez:

// models_and_db.py (SQLAlchemy 2.0 Async e Modelos Tipados)
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import String, ForeignKey, select
from typing import List

# 1. Configuração do Engine Assíncrono
DATABASE_URL = "postgresql+asyncpg://postgres:secret@localhost:5432/core_db"
engine = create_async_engine(DATABASE_URL, echo=False)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

# 2. Base Declarativa
class Base(DeclarativeBase):
    pass

# 3. Modelos com Tipagem Mapped
class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(150), unique=True, index=True)
    orders: Mapped[List["Order"]] = relationship(back_populates="user")

class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    amount: Mapped[float]
    user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
    user: Mapped["User"] = relationship(back_populates="orders")

# 4. Exemplo de Consulta Assíncrona Moderna
async def get_user_by_email(session: AsyncSession, email: str) -> User | None:
    stmt = select(User).where(User.email == email)
    result = await session.execute(stmt)
    return result.scalar_one_or_none()

💡 Análise Passo a Passo do Código

1. Mapped Tipado: Mapped[str] fornece inferência perfeita de tipos sem depender de construtores de colunas legados.
2. Sintaxe select() do 2.0: select(User).where(...) substitui a API legada session.query(), alinhando a sintaxe ao SQL padrão.
3. Sessão sem Expiração: expire_on_commit=False impede acessos preguiçosos (lazy loading) inesperados fora do contexto assíncrono.

🎯 3. Próximos Passos & Sequência Didática