Python Admin

Boas Práticas de SQLAlchemy ORM em Python: Models, Relacionamentos e Sessions para Times Ágeis Já leu

O que é SQLAlchemy ORM e por que você precisa dela SQLAlchemy é a biblioteca mais madura e poderosa para trabalhar com bancos de dados em Python. A sigla ORM significa Object-Relational Mapping, ou seja, ela mapeia objetos Python diretamente para tabelas do banco de dados. Em vez de escrever SQL puro, você trabalha com classes Python que representam suas entidades, e a biblioteca cuida da tradução para SQL. A grande vantagem é que você ganha abstração e segurança. Não precisa se preocupar com sintaxe SQL específica de cada banco de dados, não está vulnerável a SQL injection, e o código fica mais legível e manutenível. Se você precisar mudar de PostgreSQL para MySQL, basta alterar a string de conexão — seu código Python permanece praticamente igual. Models: Definindo suas Entidades Estrutura Básica de um Model Um Model é uma classe Python que representa uma tabela no banco de dados. Cada atributo da classe é uma coluna, e cada instância

O que é SQLAlchemy ORM e por que você precisa dela

SQLAlchemy é a biblioteca mais madura e poderosa para trabalhar com bancos de dados em Python. A sigla ORM significa Object-Relational Mapping, ou seja, ela mapeia objetos Python diretamente para tabelas do banco de dados. Em vez de escrever SQL puro, você trabalha com classes Python que representam suas entidades, e a biblioteca cuida da tradução para SQL.

A grande vantagem é que você ganha abstração e segurança. Não precisa se preocupar com sintaxe SQL específica de cada banco de dados, não está vulnerável a SQL injection, e o código fica mais legível e manutenível. Se você precisar mudar de PostgreSQL para MySQL, basta alterar a string de conexão — seu código Python permanece praticamente igual.

Models: Definindo suas Entidades

Estrutura Básica de um Model

Um Model é uma classe Python que representa uma tabela no banco de dados. Cada atributo da classe é uma coluna, e cada instância é um registro. Você define um Model herdando de uma classe base que SQLAlchemy fornece.

from sqlalchemy import create_engine, Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime

# Criamos a base que todas as nossas classes vão herdar
Base = declarative_base()

class Usuario(Base):
    __tablename__ = 'usuarios'

    id = Column(Integer, primary_key=True)
    nome = Column(String(100), nullable=False)
    email = Column(String(120), unique=True, nullable=False)
    data_criacao = Column(DateTime, default=datetime.utcnow)

# Criamos o engine (conexão com o banco)
engine = create_engine('sqlite:///banco.db')

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

O __tablename__ define o nome da tabela no banco de dados. O Column especifica tipo, restrições (primary_key, nullable, unique). O engine é responsável por gerenciar a conexão com o banco — neste exemplo, usamos SQLite por simplicidade, mas você pode usar PostgreSQL, MySQL, Oracle, etc.

Tipos de Dados e Validações

SQLAlchemy oferece tipos de dados robustos que mapeiam para tipos nativos do banco. Além disso, você pode adicionar validações simples direto nas colunas.

from sqlalchemy import Boolean, Float, Text, Enum
import enum

class StatusPedido(enum.Enum):
    PENDENTE = "pendente"
    PROCESSANDO = "processando"
    CONCLUIDO = "concluido"

class Pedido(Base):
    __tablename__ = 'pedidos'

    id = Column(Integer, primary_key=True)
    descricao = Column(Text)  # Texto longo
    valor = Column(Float, nullable=False)  # Números decimais
    ativo = Column(Boolean, default=True)  # Booleanos
    status = Column(Enum(StatusPedido), default=StatusPedido.PENDENTE)
    quantidade = Column(Integer, default=1)

Cada tipo de coluna tem seu mapeamento automático para o banco. Float vira DECIMAL, Boolean vira um campo booleano (ou INTEGER em alguns bancos), Enum permite usar enumerações Python de forma segura.

Relacionamentos: Conectando Models

One-to-Many (Um para Muitos)

Este é o relacionamento mais comum. Um usuário pode ter muitos pedidos, mas cada pedido pertence a apenas um usuário. Você define isso com ForeignKey e relationship.

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Usuario(Base):
    __tablename__ = 'usuarios'

    id = Column(Integer, primary_key=True)
    nome = Column(String(100), nullable=False)
    # Aqui vinculamos os pedidos do usuário
    pedidos = relationship('Pedido', back_populates='usuario')

class Pedido(Base):
    __tablename__ = 'pedidos'

    id = Column(Integer, primary_key=True)
    descricao = Column(String(255), nullable=False)
    usuario_id = Column(Integer, ForeignKey('usuarios.id'), nullable=False)
    # Aqui referenciamos o usuário
    usuario = relationship('Usuario', back_populates='pedidos')

O ForeignKey garante integridade referencial no banco de dados. O relationship é um atributo Python que permite acessar os dados relacionados facilmente. O back_populates cria um link bidirecional — você acessa usuario.pedidos ou pedido.usuario automaticamente.

Many-to-Many (Muitos para Muitos)

Quando duas entidades se relacionam de forma bidirecional sem hierarquia, usamos tabelas de associação. Por exemplo, alunos e cursos: um aluno pode estar em vários cursos, e um curso pode ter vários alunos.

from sqlalchemy import Table

# Tabela de associação (sem classe correspondente)
aluno_curso = Table(
    'aluno_curso',
    Base.metadata,
    Column('aluno_id', Integer, ForeignKey('alunos.id'), primary_key=True),
    Column('curso_id', Integer, ForeignKey('cursos.id'), primary_key=True)
)

class Aluno(Base):
    __tablename__ = 'alunos'

    id = Column(Integer, primary_key=True)
    nome = Column(String(100), nullable=False)
    cursos = relationship('Curso', secondary=aluno_curso, back_populates='alunos')

class Curso(Base):
    __tablename__ = 'cursos'

    id = Column(Integer, primary_key=True)
    titulo = Column(String(150), nullable=False)
    alunos = relationship('Aluno', secondary=aluno_curso, back_populates='cursos')

O argumento secondary aponta para a tabela de junção. Isso permite acesso simples: aluno.cursos retorna uma lista de cursos, sem você precisar escrever queries manualmente.

One-to-One (Um para Um)

Menos comum, mas útil quando uma entidade tem exatamente um relacionamento com outra. Usamos uselist=False para isso.

class Pessoa(Base):
    __tablename__ = 'pessoas'

    id = Column(Integer, primary_key=True)
    nome = Column(String(100), nullable=False)
    passaporte = relationship('Passaporte', uselist=False, back_populates='pessoa')

class Passaporte(Base):
    __tablename__ = 'passaportes'

    id = Column(Integer, primary_key=True)
    numero = Column(String(20), unique=True, nullable=False)
    pessoa_id = Column(Integer, ForeignKey('pessoas.id'), unique=True)
    pessoa = relationship('Pessoa', back_populates='passaporte')

O uselist=False faz com que pessoa.passaporte retorne um objeto único, não uma lista.

Sessions: Gerenciando o Ciclo de Vida dos Objetos

O que é uma Session

A Session é fundamental em SQLAlchemy ORM. Ela funciona como um gerenciador de contexto que rastreia mudanças nos objetos e sincroniza com o banco de dados. Você não trabalha diretamente com o banco — tudo passa pela Session.

from sqlalchemy.orm import sessionmaker

# Cria a fábrica de sessions
SessionLocal = sessionmaker(bind=engine)

# Cria uma nova session
session = SessionLocal()

try:
    # Cria um novo usuário
    novo_usuario = Usuario(nome='João Silva', email='joao@example.com')

    # Adiciona à session (mas ainda não está no banco)
    session.add(novo_usuario)

    # Commita as mudanças (agora vai para o banco)
    session.commit()

    print(f"Usuário criado com ID: {novo_usuario.id}")
finally:
    # Sempre fecha a session
    session.close()

Entenda o fluxo: você cria objetos Python, adiciona à session, e faz commit para persisti-los no banco. A session funciona como um buffer — você pode fazer múltiplas operações e confirmar tudo de uma vez.

Operações CRUD

CRUD significa Create, Read, Update, Delete. Vamos cobrir cada uma com exemplos práticos.

from sqlalchemy.orm import sessionmaker

SessionLocal = sessionmaker(bind=engine)

# CREATE
def criar_usuario(nome, email):
    session = SessionLocal()
    try:
        usuario = Usuario(nome=nome, email=email)
        session.add(usuario)
        session.commit()
        return usuario
    except Exception as e:
        session.rollback()
        print(f"Erro ao criar: {e}")
    finally:
        session.close()

# READ - Buscar um por ID
def obter_usuario_por_id(user_id):
    session = SessionLocal()
    try:
        usuario = session.query(Usuario).filter(Usuario.id == user_id).first()
        return usuario
    finally:
        session.close()

# READ - Buscar vários
def listar_usuarios():
    session = SessionLocal()
    try:
        usuarios = session.query(Usuario).all()
        return usuarios
    finally:
        session.close()

# UPDATE
def atualizar_usuario(user_id, novo_nome):
    session = SessionLocal()
    try:
        usuario = session.query(Usuario).filter(Usuario.id == user_id).first()
        if usuario:
            usuario.nome = novo_nome
            session.commit()
        return usuario
    except Exception as e:
        session.rollback()
        print(f"Erro ao atualizar: {e}")
    finally:
        session.close()

# DELETE
def deletar_usuario(user_id):
    session = SessionLocal()
    try:
        usuario = session.query(Usuario).filter(Usuario.id == user_id).first()
        if usuario:
            session.delete(usuario)
            session.commit()
    except Exception as e:
        session.rollback()
        print(f"Erro ao deletar: {e}")
    finally:
        session.close()

Note que sempre usamos try/finally para garantir que a session seja fechada, mesmo com erros. O rollback() desfaz mudanças não commitadas — é essencial para manter consistência.

Queries Avançadas e Filtering

SQLAlchemy permite queries expresivas que são traduzidas para SQL.

from sqlalchemy import and_, or_

session = SessionLocal()

# Filtro simples
usuarios = session.query(Usuario).filter(Usuario.nome == 'João Silva').all()

# Múltiplos filtros (AND implícito)
usuarios = session.query(Usuario).filter(
    Usuario.nome == 'João Silva',
    Usuario.email.like('%@gmail.com')
).all()

# OR
usuarios = session.query(Usuario).filter(
    or_(
        Usuario.nome == 'João',
        Usuario.nome == 'Maria'
    )
).all()

# Ordenação
usuarios = session.query(Usuario).order_by(Usuario.nome).all()

# Limite
primeiros_5 = session.query(Usuario).limit(5).all()

# Contar
total = session.query(Usuario).count()

# Acessando relacionamentos
usuario = session.query(Usuario).filter(Usuario.id == 1).first()
if usuario:
    # Acessa todos os pedidos do usuário
    for pedido in usuario.pedidos:
        print(pedido.descricao)

session.close()

O método filter() aceita condições Python que são automaticamente traduzidas para SQL. Operadores como ==, .like(), .in_() funcionam intuitivamente.

Lazy Loading vs Eager Loading

Por padrão, SQLAlchemy carrega relacionamentos sob demanda (lazy loading). Isso pode causar múltiplas queries. Eager loading carrega tudo de uma vez.

from sqlalchemy.orm import joinedload

session = SessionLocal()

# Lazy loading (padrão) - vai fazer uma query para cada pedido acessado
usuarios = session.query(Usuario).all()
for usuario in usuarios:
    print(usuario.pedidos)  # Cada acesso gera uma query

# Eager loading - carrega tudo em uma única query
usuarios = session.query(Usuario).options(joinedload(Usuario.pedidos)).all()
for usuario in usuarios:
    print(usuario.pedidos)  # Já está em memória, sem query extra

session.close()

Eager loading é mais eficiente quando você sabe que vai precisar dos dados relacionados. Use .options(joinedload(...)) para carregá-los junto.

Casos de Uso Prático e Padrões

Integração com FastAPI

Um padrão comum é usar SQLAlchemy com FastAPI. Aqui está uma aplicação funcional completa.

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from pydantic import BaseModel

# Setup
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# Model
class Produto(Base):
    __tablename__ = "produtos"
    id = Column(Integer, primary_key=True)
    nome = Column(String(100), nullable=False)
    preco = Column(Integer, nullable=False)

Base.metadata.create_all(bind=engine)

# Schemas Pydantic
class ProdutoSchema(BaseModel):
    id: int
    nome: str
    preco: int

    class Config:
        from_attributes = True

class ProdutoCreate(BaseModel):
    nome: str
    preco: int

# Dependency para injetar session
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# API
app = FastAPI()

@app.post("/produtos/", response_model=ProdutoSchema)
def criar_produto(produto: ProdutoCreate, db: Session = Depends(get_db)):
    novo = Produto(nome=produto.nome, preco=produto.preco)
    db.add(novo)
    db.commit()
    db.refresh(novo)
    return novo

@app.get("/produtos/{produto_id}", response_model=ProdutoSchema)
def obter_produto(produto_id: int, db: Session = Depends(get_db)):
    return db.query(Produto).filter(Produto.id == produto_id).first()

@app.get("/produtos/", response_model=list[ProdutoSchema])
def listar_produtos(db: Session = Depends(get_db)):
    return db.query(Produto).all()

O padrão de dependency injection (Depends(get_db)) fornece uma session limpa para cada requisição e a fecha automaticamente.

Tratamento de Erros e Validação

Erros de banco de dados devem ser tratados com cuidado. SQLAlchemy lança exceções que você deve antecipar.

from sqlalchemy.exc import IntegrityError, SQLAlchemyError

session = SessionLocal()

try:
    # Email duplicado vai gerar IntegrityError
    usuario1 = Usuario(nome='Ana', email='ana@test.com')
    usuario2 = Usuario(nome='Bruno', email='ana@test.com')  # Mesmo email

    session.add(usuario1)
    session.add(usuario2)
    session.commit()

except IntegrityError as e:
    session.rollback()
    print("Email já existe no banco!")

except SQLAlchemyError as e:
    session.rollback()
    print(f"Erro genérico de banco: {e}")

finally:
    session.close()

IntegrityError captura violações de constraints (unique, foreign key, etc). SQLAlchemyError é a classe mãe de todos os erros SQLAlchemy.

Conclusão

Os três pilares que você deve levar deste artigo são:

  1. Models são sua ponte com o banco de dados — cada classe herda de Base e mapeia para uma tabela. Defina tipos corretos, restrições claras, e os relacionamentos com precisão. Isso previne bugs e garante integridade dos dados.

  2. Relacionamentos exigem planejamento — One-to-Many, Many-to-Many, One-to-One têm usos distintos. Entenda quando usar cada um, porque isso afeta como você consulta os dados e como o banco os organiza. Use back_populates para manter referências bidirecionais limpas.

  3. Sessions são o coração da ORM — toda interação com o banco passa por ela. Sempre feche sessions em finally, use commit() para persisti-las, e rollback() para desfazer mudanças. Eager loading com joinedload previne queries N+1 desnecessárias.

Referências


Artigos relacionados