Prática Integrada com Projetos Interessantes usando SQLAlchemy ORM

O SQLAlchemy é um dos frameworks de Mapeametno Objeto-Relacional (ORM) mais populares em Python, fornecendo uma maneira eficiente e flexível de operar bancos de dados. Este artigo introduz como realizar operações de banco de dados utilizando o ORM do SQLAlchemy.

Instalação

pip install sqlalchemy

Para conectar a bencos de dados específicos, é necessário instalar os drivers correspondentes:

# PostgreSQL
pip install psycopg2-binary

# MySQL
pip install mysql-connector-python

# SQLite (incluído na biblioteca padrão do Python, sem instalação adicional)

Conceitos Principais

Engine: Responsável por gerenciar a comunicação com o banco de dados.
Session: Gerencia todas as operações de persistência de objetos.
Model: Classes que representam tabelas no banco de dados.
Query: Objeto usado para construir e executar consultas.

Estabelecendo Conexão com o Banco de Dados

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# Configuração do motor de conexão (exemplo com PostgreSQL)
db_engine = create_engine('postgresql://user:pass@localhost:5432/mydb', echo=False)

# Fábrica de sessões
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)

# Criação de uma sessão
db_session = SessionFactory()

Definição dos Modelos de Dados

from sqlalchemy import Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import relationship, declarative_base

Base = declarative_base()

# Tabela associativa para relacionamento muitos-para-muitos
post_tag_association = Table('post_tags', Base.metadata,
    Column('post_id', Integer, ForeignKey('posts.id')),
    Column('tag_id', Integer, ForeignKey('tags.id'))
)

class Author(Base):
    __tablename__ = 'authors'
    
    id = Column(Integer, primary_key=True)
    username = Column(String(50), nullable=False, unique=True)
    email = Column(String(100), unique=True)
    
    # Relacionamento um-para-muitos
    articles = relationship("Article", back_populates="creator")

class Article(Base):
    __tablename__ = 'articles'
    
    id = Column(Integer, primary_key=True)
    title = Column(String(150), nullable=False)
    body = Column(String(2000))
    author_id = Column(Integer, ForeignKey('authors.id'))
    
    # Relacionamento muitos-para-um
    creator = relationship("Author", back_populates="articles")
    
    # Relacionamento muitos-para-muitos
    categories = relationship("Category", secondary=post_tag_association, back_populates="articles")

class Category(Base):
    __tablename__ = 'categories'
    
    id = Column(Integer, primary_key=True)
    name = Column(String(40), unique=True, nullable=False)
    
    articles = relationship("Article", secondary=post_tag_association, back_populates="categories")

Geração do Esquema do Banco de Dados

# Cria todas as tabelas definidas nos modelos
Base.metadata.create_all(bind=db_engine)

# Para remover todas as tabelas:
# Base.metadata.drop_all(bind=db_engine)

Operações CRUD Básicas

Criação de Registros

new_author = Author(username="carlos", email="carlos@example.com")
db_session.add(new_author)
db_session.commit()

# Inserção em lote
db_session.add_all([
    Author(username="ana", email="ana@example.com"),
    Author(username="pedro", email="pedro@example.com")
])
db_session.commit()

Leitura de Dados

# Recuperar todos os autores
all_authors = db_session.query(Author).all()

# Obter o primeiro resultado
first_author = db_session.query(Author).first()

# Buscar por ID
specific_author = db_session.query(Author).get(1)

Atualização de Registros

author_to_update = db_session.query(Author).get(1)
author_to_update.username = "carlos_silva"
db_session.commit()

# Atualização em massa
db_session.query(Author).filter(Author.username.like("ca%")).update(
    {"username": "carlos_modificado"}, synchronize_session=False
)
db_session.commit()

Exclusão de Dados

author_to_delete = db_session.query(Author).get(1)
db_session.delete(author_to_delete)
db_session.commit()

# Exclusão em massa
db_session.query(Author).filter(Author.username == "ana").delete(synchronize_session=False)
db_session.commit()

Construção de Consultas

Filtros e Seleção

from sqlalchemy import or_, func

# Consulta com filtro exato
found_author = db_session.query(Author).filter(Author.username == "pedro").first()

# Busca com padrão
authors = db_session.query(Author).filter(Author.username.like("%dr%")).all()

# Condição IN
authors = db_session.query(Author).filter(Author.username.in_(["carlos", "pedro"])).all()

# Filtros múltiplos
authors = db_session.query(Author).filter(
    Author.username.startswith("ca"),
    Author.email.contains("example.com")
).all()

# Condição OR
authors = db_session.query(Author).filter(
    or_(Author.username == "carlos", Author.username == "pedro")
).all()

Junções e Agregações

# Contagem total de autores
total_authors = db_session.query(func.count(Author.id)).scalar()

# Junção interna com filtro
results = db_session.query(Author, Article).join(Article).filter(
    Article.title.contains("Python")
).all()

# Junção externa
results = db_session.query(Author, Article).outerjoin(Article).all()

Manipulação de Relacionamentos

# Criando objetos relacionados
author = Author(username="maria", email="maria@example.com")
article = Article(title="Meu primeiro post", body="Conteúdo do artigo.", creator=author)
db_session.add(article)
db_session.commit()

# Acessando dados relacionados
print(f"O artigo '{article.title}' foi criado por {article.creator.username}")
print(f"Artigos de {author.username}:")
for art in author.articles:
    print(f"  - {art.title}")

# Adicionando categorias (relacionamento N:N)
python_cat = Category(name="Python")
orm_cat = Category(name="SQLAlchemy")

article.categories.append(python_cat)
article.categories.append(orm_cat)
db_session.commit()

Gerenciamento de Transações

# Tratamento de exceções
try:
    temp_user = Author(username="temp_user", email="temp@example.com")
    db_session.add(temp_user)
    db_session.commit()
except Exception as error:
    db_session.rollback()
    print(f"Erro na transação: {error}")

# Uso de ponto de salvamento
savepoint = db_session.begin_nested()
try:
    another_user = Author(username="sp_user", email="sp@example.com")
    db_session.add(another_user)
    savepoint.commit()
except:
    savepoint.rollback()

Recomendações de Uso

Gerenciamento de Sessão: Crie uma nova sessão por requisição/unidade de trabalho e feche-a ao final.
Tratamento de Erros: Sempre implemente tratamento de exceções e realize rollback quando necessário.
Otimização de Carregamento: Use eager loading (como joinedload) para evitar o problema N+1 em consultas com relacionamentos.
Validação de Dados: Implemente validações no nível do modelo ou da aplicação para manter a integridade dos dados.

from contextlib import contextmanager

@contextmanager
def get_session():
    session = SessionFactory()
    try:
        yield session
        session.commit()
    except Exception:
        session.rollback()
        raise
    finally:
        session.close()

# Exemplo de utilização
with get_session() as session:
    new_author = Author(username="context_author", email="ctx@example.com")
    session.add(new_author)

Tags: SQLAlchemy ORM Python postgresql banco de dados

Publicado em 6-9 20:04 por Thomas

Doido Dev