Guia completo sobre SQLAlchemy em Python: ORM, modelos, consultas, relacionamentos, migrations e boas praticas para trabalhar com bancos de dados.

SQLAlchemy: O que É e Como Funciona | Python Brasil | Python Brasil

O que e SQLAlchemy?

SQLAlchemy e a biblioteca de banco de dados mais popular e poderosa do ecossistema Python. Ela oferece duas interfaces: o Core, que fornece uma camada de abstracao SQL expressiva, e o ORM (Object-Relational Mapping), que permite mapear tabelas do banco de dados para classes Python. Com SQLAlchemy, voce trabalha com objetos Python em vez de escrever SQL bruto, mantendo ao mesmo tempo controle total sobre as consultas geradas.

SQLAlchemy suporta diversos bancos de dados, incluindo PostgreSQL, MySQL, SQLite, Oracle e Microsoft SQL Server, permitindo trocar de banco com minimas alteracoes no codigo.

Instalacao e Configuracao

# Instalacao
# pip install sqlalchemy

# Para PostgreSQL
# pip install sqlalchemy psycopg2-binary

# Para MySQL
# pip install sqlalchemy pymysql

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, Session

# Criando a engine (conexao com o banco)
engine = create_engine('sqlite:///meubanco.db', echo=True)

# Para PostgreSQL
# engine = create_engine('postgresql://usuario:senha@localhost:5432/meubanco')

# Classe base para modelos
class Base(DeclarativeBase):
    pass

Definindo Modelos

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

class Usuario(Base):
    __tablename__ = 'usuarios'

    id: Mapped[int] = mapped_column(primary_key=True)
    nome: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(200), unique=True)
    ativo: Mapped[bool] = mapped_column(default=True)
    criado_em: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow
    )

    # Relacionamento um-para-muitos
    pedidos: Mapped[list['Pedido']] = relationship(back_populates='usuario')

    def __repr__(self):
        return f'<Usuario(id={self.id}, nome={self.nome})>'


class Pedido(Base):
    __tablename__ = 'pedidos'

    id: Mapped[int] = mapped_column(primary_key=True)
    usuario_id: Mapped[int] = mapped_column(ForeignKey('usuarios.id'))
    produto: Mapped[str] = mapped_column(String(200))
    valor: Mapped[float] = mapped_column(Float)
    criado_em: Mapped[datetime] = mapped_column(
        DateTime, default=datetime.utcnow
    )

    # Relacionamento reverso
    usuario: Mapped['Usuario'] = relationship(back_populates='pedidos')

    def __repr__(self):
        return f'<Pedido(id={self.id}, produto={self.produto})>'


# Criando as tabelas no banco
Base.metadata.create_all(engine)

Operacoes CRUD

from sqlalchemy.orm import Session

# CREATE — inserindo dados
with Session(engine) as session:
    novo_usuario = Usuario(nome='Ana Silva', email='ana@email.com')
    session.add(novo_usuario)

    # Inserindo varios de uma vez
    session.add_all([
        Usuario(nome='Bruno Costa', email='bruno@email.com'),
        Usuario(nome='Carla Lima', email='carla@email.com'),
    ])
    session.commit()

# READ — consultando dados
with Session(engine) as session:
    # Busca por chave primaria
    usuario = session.get(Usuario, 1)
    print(usuario.nome)

    # Busca com filtro
    from sqlalchemy import select

    stmt = select(Usuario).where(Usuario.nome == 'Ana Silva')
    resultado = session.execute(stmt).scalar_one_or_none()

    # Todos os usuarios ativos
    stmt = select(Usuario).where(Usuario.ativo == True).order_by(Usuario.nome)
    usuarios = session.execute(stmt).scalars().all()

# UPDATE — atualizando dados
with Session(engine) as session:
    usuario = session.get(Usuario, 1)
    usuario.nome = 'Ana Maria Silva'
    session.commit()

# DELETE — removendo dados
with Session(engine) as session:
    usuario = session.get(Usuario, 3)
    session.delete(usuario)
    session.commit()

Consultas Avancadas

from sqlalchemy import select, func, and_, or_, desc

with Session(engine) as session:
    # Filtros compostos
    stmt = select(Usuario).where(
        and_(
            Usuario.ativo == True,
            Usuario.nome.like('A%')
        )
    )

    # OR
    stmt = select(Usuario).where(
        or_(
            Usuario.email.contains('gmail'),
            Usuario.email.contains('hotmail')
        )
    )

    # Agregacoes
    stmt = select(func.count(Usuario.id)).where(Usuario.ativo == True)
    total = session.execute(stmt).scalar()

    # JOIN com relacionamentos
    stmt = (
        select(Usuario, Pedido)
        .join(Pedido)
        .where(Pedido.valor > 100)
        .order_by(desc(Pedido.criado_em))
    )

    # Agrupamento
    stmt = (
        select(Usuario.nome, func.sum(Pedido.valor).label('total'))
        .join(Pedido)
        .group_by(Usuario.nome)
        .having(func.sum(Pedido.valor) > 500)
    )

    # Paginacao
    stmt = select(Usuario).order_by(Usuario.id).offset(0).limit(10)
    pagina = session.execute(stmt).scalars().all()

Gerenciamento de Sessao

from sqlalchemy.orm import sessionmaker

# Fabrica de sessoes
SessionLocal = sessionmaker(bind=engine)

# Usando como context manager
def criar_usuario(nome: str, email: str) -> Usuario:
    with SessionLocal() as session:
        with session.begin():  # commit automatico ou rollback em caso de erro
            usuario = Usuario(nome=nome, email=email)
            session.add(usuario)
            return usuario

# Pattern para FastAPI / Flask
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Migrations com Alembic

# Instalacao: pip install alembic
# Inicializar: alembic init alembic
# Gerar migration: alembic revision --autogenerate -m "criar tabela usuarios"
# Aplicar: alembic upgrade head
# Reverter: alembic downgrade -1

# Exemplo de migration gerada (alembic/versions/xxx_criar_tabela.py)
"""
def upgrade():
    op.create_table(
        'usuarios',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('nome', sa.String(100), nullable=False),
        sa.Column('email', sa.String(200), unique=True),
    )

def downgrade():
    op.drop_table('usuarios')
"""

Erros Comuns

O erro mais frequente e nao fechar sessoes corretamente, causando vazamento de conexoes. Sempre use context managers (with). Outro erro e acessar atributos de relacionamento fora da sessao, o que causa DetachedInstanceError. Tambem e comum esquecer de chamar session.commit() apos modificacoes, ou nao tratar excecoes que exigem session.rollback(). O uso de echo=True em producao e outro erro que polui logs com SQL desnecessario.

Boas Praticas

Use o estilo moderno com Mapped e mapped_column em vez da API classica Column. Sempre use context managers para sessoes. Configure Alembic para migrations desde o inicio do projeto. Use sessionmaker para criar uma fabrica de sessoes. Separe a definicao de modelos da logica de negocio. Em aplicacoes web, use uma sessao por requisicao. Evite carregar relacionamentos desnecessarios — configure lazy='select' ou use selectinload para otimizar consultas.

Quando Usar

SQLAlchemy e ideal para qualquer projeto Python que interaja com bancos de dados relacionais. O ORM e excelente para aplicacoes com logica de dominio complexa, enquanto o Core e preferivel para scripts de migracao de dados ou consultas que exigem controle fino sobre o SQL gerado.

SQLAlchemy: O que É e Como Funciona | Python Brasil