Introdução ao SQLAlchemy

SQLAlchemy é uma biblioteca ORM (Object Relational Mapper) em Python, que atua como uma ponte entre objetos Python e tabelas de banco de dados relacionais. Ele abstrai as operações de banco de dados, permitindo que desenvolvedores interajam com o banco de dados usando objetos Python em vez de escrever SQL puro. Embora o Flask, por si só, não inclua um ORM, o SQLAlchemy é uma escolha popular para gerenciamento de dados em aplicações Flask e FastAPI.

Para começar a usar o SQLAlchemy, você precisa instalá-lo:

pip install sqlalchemy

É importante notar que o SQLAlchemy não interage diretamente com o banco de dados. Ele requer um driver de banco de dados específico, como pymysql para MySQL, psycopg2 para PostgreSQL, ou cx_Oracle para Oracle. A string de conexão para o motor do banco de dados varia conforme o driver e o tipo de banco de dados:

# Exemplo para MySQL com PyMySQL
# mysql+pymysql://<usuário>:<senha>@<host>/<nome_do_banco>[?<opções>]

# Exemplo para Oracle com cx_Oracle
# oracle+cx_oracle://<usuário>:<senha>@<host>:<porta>/<nome_do_banco>[?<chave>=<valor>...]

Executando Comandos SQL Nativos

Mesmo sendo um ORM, o SQLAlchemy oferece maneiras de executar comandos SQL diretamente no banco de dados. Existem duas abordagens principais:

Abordagem com Engine e Cursor Direto

Esta abordagem é mais próxima da interação tradicional com drivers de banco de dados, utilizando o objeto Engine e um cursor para executar comandos SQL e buscar resultados.

from sqlalchemy import create_engine
import pymysql # Importe o driver de banco de dados necessário

# Configura o motor de conexão com o banco de dados
db_engine = create_engine(
    "mysql+pymysql://root:password@127.0.0.1:3306/my_app_db",
    max_overflow=0,      # Número máximo de conexões extras além do pool_size
    pool_size=5,         # Tamanho do pool de conexões
    pool_timeout=30,     # Tempo de espera por uma conexão no pool
    pool_recycle=-1      # Tempo para reciclagem de conexões (em segundos, -1 para nunca)
)

# Obtém uma conexão "bruta" do motor
db_connection = db_engine.raw_connection()
try:
    # Cria um objeto cursor para executar comandos
    db_cursor = db_connection.cursor()
    # Executa um comando SQL
    db_cursor.execute('SELECT * FROM users')
    # Recupera os resultados
    query_results = db_cursor.fetchall()
    print("Resultados da consulta SQL nativa (método 1):", query_results)
finally:
    # Garante que a conexão seja fechada
    db_connection.close()

Abordagem com Session e Texto SQL

Esta abordagem integra a execução de SQL nativo com o sistema de sessão do SQLAlchemy, que é mais comum em operações ORM. Para versões mais recentes do SQLAlchemy (a partir de 2.0), é recomendado usar a função text() para encapsular strings SQL.

from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker, scoped_session

# Assumindo que User e Book são modelos ORM já definidos
# from my_models import User, Book 

db_engine = create_engine("mysql+pymysql://root:password@127.0.0.1:3306/my_app_db")
SessionMaker = sessionmaker(bind=db_engine)
# Usando scoped_session para segurança de threads em ambientes web
current_session = scoped_session(SessionMaker)

try:
    # Exemplo de consulta SELECT
    select_cursor = current_session.execute(text('SELECT id, name FROM books WHERE id > :book_id'), params={"book_id": 1})
    selected_books = select_cursor.fetchall()
    print("Livros selecionados (método 2):", selected_books)

    # Exemplo de INSERT
    insert_cursor = current_session.execute(text('INSERT INTO books (name) VALUES (:book_name)'), params={"book_name": 'O Pequeno Príncipe'})
    current_session.commit() # Confirma a transação
    print("ID do novo livro:", insert_cursor.lastrowid)
except Exception as e:
    current_session.rollback() # Em caso de erro, desfaz a transação
    print(f"Erro ao executar SQL nativo: {e}")
finally:
    current_session.close() # Libera a sessão

Definindo Modelos e Criando Tabelas

A definição de modelos no SQLAlchemy utiliza o padrão Declarative, onde você define classes Python que representam as tabelas do seu banco de dados.

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

# Base para todos os modelos declarativos
Base = declarative_base()

class MyUser(Base):
    __tablename__ = 'users_table' # Nome da tabela no banco de dados

    id = Column(Integer, primary_key=True)
    username = Column(String(50), nullable=False, unique=True, index=True) # Nome de usuário, único e indexado
    email = Column(String(100), nullable=True)
    created_at = Column(DateTime, default=datetime.now) # Data de criação automática

    # Definições de tabela adicionais, como índices compostos ou restrições únicas
    __table_args__ = (
        UniqueConstraint('username', 'email', name='uq_username_email'), # Restrição única combinada
        Index('ix_username', 'username'), # Índice adicional para username
    )

    def __repr__(self):
        return f"<MyUser(id={self.id}, username='{self.username}')>"

class MyBook(Base):
    __tablename__ = 'books_table'

    id = Column(Integer, primary_key=True)
    title = Column(String(100), nullable=False)
    author_id = Column(Integer, ForeignKey('users_table.id')) # Chave estrangeira para MyUser

    def __repr__(self):
        return f"<MyBook(id={self.id}, title='{self.title}')>"

# Cria o motor de banco de dados
db_engine = create_engine("mysql+pymysql://root:password@127.0.0.1:3306/my_app_db")

# Cria todas as tabelas definidas nos modelos que herdam de Base
# O banco de dados ('my_app_db') precisa existir previamente
Base.metadata.create_all(db_engine)

# Para apagar todas as tabelas (cuidado!)
# Base.metadata.drop_all(db_engine)

Operações ORM Básicas (CRUD)

O SQLAlchemy ORM permite manipular dados como se estivesse interagindo com objetos Python. Veja como realizar operações de Criação, Leitura, Atualização e Exclusão (CRUD).

Configuração da Sessão

Para interagir com o banco de dados via ORM, você precisa de um objeto de sessão. A sessão é o ponto de entrada para todas as operações de banco de dados.

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from my_models import MyUser, MyBook # Assumindo que você definiu os modelos acima

db_engine = create_engine("mysql+pymysql://root:password@127.0.0.1:3306/my_app_db")
SessionLocal = sessionmaker(bind=db_engine, autocommit=False, autoflush=False)
db_session = SessionLocal() # Use esta sessão em um contexto simples
# Para aplicações web, use scoped_session (detalhes mais abaixo)
# db_session = scoped_session(SessionLocal) 

Criação de Registros (Adicionar)

# Adicionar um único objeto
new_user = MyUser(username='alice', email='alice@example.com')
db_session.add(new_user)
db_session.commit() # Confirma a transação
print(f"Usuário criado: {new_user.id}")

# Adicionar múltiplos objetos
another_user = MyUser(username='bob', email='bob@example.com')
a_book = MyBook(title='A Arte da Guerra', author_id=new_user.id)
db_session.add_all([another_user, a_book])
db_session.commit()
print(f"Mais usuários e livros adicionados.")

Leitura de Registros (Consultar)

O método query() é o principal para construir consultas. Você pode usar filter() para condições arbitrárias e filter_by() para condições de igualdade simples.

# Consultar todos os usuários
all_users = db_session.query(MyUser).all()
print("Todos os usuários:", all_users)

# Consultar um usuário por ID
user_by_id = db_session.query(MyUser).filter(MyUser.id == 1).first()
print(f"Usuário com ID 1: {user_by_id}")

# Consultar usuários por username usando filter_by
user_by_username = db_session.query(MyUser).filter_by(username='bob').first()
print(f"Usuário 'bob': {user_by_username}")

# Consultar usuários com email específico e ID maior que 0
specific_users = db_session.query(MyUser).filter(MyUser.email == 'alice@example.com', MyUser.id > 0).all()
print("Usuários específicos:", specific_users)

Atualização de Registros

Você pode atualizar registros buscando o objeto e modificando seus atributos, ou usando o método update() para atualizações em massa.

# Atualizar um objeto existente
user_to_update = db_session.query(MyUser).filter(MyUser.username == 'alice').first()
if user_to_update:
    user_to_update.email = 'alice.new@example.com'
    db_session.add(user_to_update) # add() também funciona para atualizar se a PK existe
    db_session.commit()
    print(f"Email de '{user_to_update.username}' atualizado.")

# Atualizar múltiplos registros com update()
# Retorna o número de linhas afetadas
rows_updated = db_session.query(MyUser).filter(MyUser.id == 2).update({"email": "bob.updated@example.com"})
db_session.commit()
print(f"{rows_updated} registro(s) de usuário atualizado(s).")

Exclusão de Registros

# Excluir um registro
user_to_delete = db_session.query(MyUser).filter(MyUser.username == 'alice').first()
if user_to_delete:
    db_session.delete(user_to_delete)
    db_session.commit()
    print(f"Usuário '{user_to_delete.username}' excluído.")

# Excluir registros por condição
# delete() também retorna o número de linhas afetadas
rows_deleted = db_session.query(MyBook).filter(MyBook.id > 1).delete()
db_session.commit()
print(f"{rows_deleted} registro(s) de livro excluído(s).")

Consultas ORM Avançadas

O SQLAlchemy oferece uma gama poderosa de funcionalidades para consultas complexas, incluindo seleção de colunas específicas, uso de expressões SQL, paginação, ordenação e agrupamento.

Seleção de Colunas e Aliases

# Selecionar apenas algumas colunas
partial_users = db_session.query(MyUser.username, MyUser.email).all()
print("Usuários (username, email):", partial_users)

# Usar rótulos (aliases) para as colunas
from sqlalchemy.sql import label
aliased_data = db_session.query(label('user_name', MyUser.username), MyUser.created_at).all()
for item in aliased_data:
    print(f"Nome: {item.user_name}, Criado em: {item.created_at}")

Cláusulas WHERE Flexíveis

# Usando text() para condições SQL nativas (menos recomendado para portabilidade)
users_by_raw_sql = db_session.query(MyUser).filter(text("id < :val OR username = :uname")).params(val=3, uname='bob').all()
print("Usuários por SQL nativo:", users_by_raw_sql)

# Consulta com range (BETWEEN)
users_in_range = db_session.query(MyUser).filter(MyUser.id.between(1, 2)).all()
print("Usuários entre ID 1 e 2:", users_in_range)

# Consulta com IN
users_in_list = db_session.query(MyUser).filter(MyUser.username.in_(['bob', 'charlie'])).all()
print("Usuários na lista:", users_in_list)

# Consulta com NOT (~)
users_not_in_list = db_session.query(MyUser).filter(~MyUser.username.in_(['bob'])).all()
print("Usuários não 'bob':", users_not_in_list)

# Condições AND e OR
from sqlalchemy import and_, or_
complex_query = db_session.query(MyUser).filter(
    or_(
        MyUser.id < 2,
        and_(MyUser.username == 'bob', MyUser.email.like('%@example.com'))
    )
).all()
print("Consulta complexa OR/AND:", complex_query)

# Operadores LIKE e NOT LIKE
users_with_email = db_session.query(MyUser).filter(MyUser.email.like('%@example.com')).all()
print("Usuários com email @example.com:", users_with_email)
users_without_e = db_session.query(MyUser).filter(~MyUser.username.like('a%')).all()
print("Usuários que não começam com 'a':", users_without_e)

Paginação

Para paginar resultados, você pode usar a notação de fatiamento (slice) diretamente no objeto de consulta.

# Exemplo: 2 registros por página, pegar a 2ª página (índice 1)
page_size = 2
page_number = 1 # Segunda página
paginated_users = db_session.query(MyUser).offset(page_number * page_size).limit(page_size).all()
print(f"Página {page_number + 1} de usuários:", paginated_users)

Ordenação

Use order_by() para especificar a ordem dos resultados. Os métodos asc() e desc() são usados para ordem ascendente ou descendente, respectivamente.

# Ordenar por username em ordem descendente
ordered_users_desc = db_session.query(MyUser).order_by(MyUser.username.desc()).all()
print("Usuários ordenados por username (desc):", ordered_users_desc)

# Ordenar por email (asc) e depois por id (desc)
multi_ordered_users = db_session.query(MyUser).order_by(MyUser.email.asc(), MyUser.id.desc()).all()
print("Usuários ordenados por email (asc) e id (desc):", multi_ordered_users)

Agrupamento e Funções Agregadas

Funções agregadas como count(), sum(), min(), max() podem ser usadas com group_by() para sumarizar dados. A cláusula having() é usada para filtrar resultados agrupados.

from sqlalchemy.sql import func

# Contar usuários por parte do email (domínio, por exemplo)
email_domain_counts = db_session.query(
    func.substring_index(MyUser.email, '@', -1).label('domain'),
    func.count(MyUser.id).label('total_users')
).group_by('domain').all()
print("Contagem de usuários por domínio de email:", email_domain_counts)

# Agregação com HAVING
# Supondo que você tem um campo 'status' nos usuários
# db_session.add(MyUser(username='charlie', email='charlie@example.com', status='active'))
# db_session.add(MyUser(username='david', email='david@test.com', status='inactive'))
# db_session.commit()

# Exemplo conceitual (assumindo campo 'status' em MyUser para demonstração)
# if hasattr(MyUser, 'status'):
#     active_users_count = db_session.query(
#         MyUser.status,
#         func.count(MyUser.id)
#     ).group_by(MyUser.status).having(func.count(MyUser.id) > 1).all()
#     print("Status com mais de um usuário:", active_users_count)

Relacionamentos de Chave Estrangeira

Relacionamentos entre tabelas são a espinha dorsal de bancos de dados relacionais e são bem suportados pelo SQLAlchemy. Essencialmente, todos os relacionamentos podem ser vistos como variações do "um-para-muitos".

Relacionamento Um-para-Muitos

Neste relacionamento, o lado "muitos" detém a chave estrangeira para o lado "um". No SQLAlchemy, você define ForeignKey no modelo da tabela "muitos" e usa relationship() para habilitar consultas orientadas a objetos.

Definição do Modelo Um-para-Muitos

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

Base = declarative_base()

class Department(Base):
    __tablename__ = 'departments'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True, nullable=False)

    # Define o relacionamento de volta para Employee (o lado 'um')
    employees = relationship('Employee', back_populates='department')

    def __repr__(self):
        return f"<Department(id={self.id}, name='{self.name}')>"

class Employee(Base):
    __tablename__ = 'employees'
    id = Column(Integer, primary_key=True)
    full_name = Column(String(100), nullable=False)
    department_id = Column(Integer, ForeignKey('departments.id'), nullable=False)

    # Define o relacionamento para Department (o lado 'muitos')
    department = relationship('Department', back_populates='employees')

    def __repr__(self):
        return f"<Employee(id={self.id}, name='{self.full_name}')>"

# db_engine = create_engine("mysql+pymysql://root:password@127.0.0.1:3306/my_app_db")
# Base.metadata.create_all(db_engine) # Crie as tabelas

Adição e Consulta Baseada em Objetos (Um-para-Muitos)

# from my_models import Department, Employee # Assumindo os modelos acima
# db_session = scoped_session(sessionmaker(bind=db_engine)) # Crie uma sessão

# Adicionando objetos e vinculando-os
dep_hr = Department(name='Recursos Humanos')
emp_john = Employee(full_name='John Doe', department=dep_hr) # Vinculando por objeto
emp_jane = Employee(full_name='Jane Smith', department_id=dep_hr.id) # Vinculando por ID (após commit)
db_session.add_all([dep_hr, emp_john, emp_jane])
db_session.commit()

# Consulta direta (lado 'muitos' para 'um')
employee = db_session.query(Employee).filter_by(full_name='John Doe').first()
if employee:
    print(f"Departamento de {employee.full_name}: {employee.department.name}")

# Consulta inversa (lado 'um' para 'muitos')
hr_department = db_session.query(Department).filter_by(name='Recursos Humanos').first()
if hr_department:
    print(f"Funcionários do departamento de {hr_department.name}:")
    for emp in hr_department.employees:
        print(f"- {emp.full_name}")

Relacionamento Muitos-para-Muitos

Relacionamentos muitos-para-muitos são implementados usando uma tabela de associação (ou "tabela intermediária"). Essa tabela contém chaves estrangeiras para as duas tabelas principais.

Definição do Modelo Muitos-para-Muitos

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

Base = declarative_base()

# Tabela de associação (não é um modelo declarativo completo)
# Pode ser definida como uma classe Base para mais flexibilidade
student_course_association = Table(
    'student_course', Base.metadata,
    Column('student_id', Integer, ForeignKey('students.id'), primary_key=True),
    Column('course_id', Integer, ForeignKey('courses.id'), primary_key=True)
)

class Student(Base):
    __tablename__ = 'students'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)

    # Define o relacionamento muitos-para-muitos com Course
    courses = relationship(
        'Course',
        secondary=student_course_association,
        back_populates='students'
    )

    def __repr__(self):
        return f"<Student(id={self.id}, name='{self.name}')>"

class Course(Base):
    __tablename__ = 'courses'
    id = Column(Integer, primary_key=True)
    title = Column(String(100), nullable=False)

    # Define o relacionamento muitos-para-muitos com Student
    students = relationship(
        'Student',
        secondary=student_course_association,
        back_populates='courses'
    )

    def __repr__(self):
        return f"<Course(id={self.id}, title='{self.title}')>"

# db_engine = create_engine("mysql+pymysql://root:password@127.0.0.1:3306/my_app_db")
# Base.metadata.create_all(db_engine) # Crie as tabelas

Adição e Consulta Baseada em Objetos (Muitos-para-Muitos)

# from my_models import Student, Course # Assumindo os modelos acima
# db_session = scoped_session(sessionmaker(bind=db_engine)) # Crie uma sessão

# Criando alunos e cursos
student_anna = Student(name='Anna')
course_math = Course(title='Matemática Avançada')
course_physics = Course(title='Física Moderna')

# Associando diretamente via objetos
student_anna.courses.append(course_math)
student_anna.courses.append(course_physics)

db_session.add_all([student_anna, course_math, course_physics])
db_session.commit()

# Um novo aluno e associando a cursos existentes
student_ben = Student(name='Ben')
existing_math = db_session.query(Course).filter_by(title='Matemática Avançada').first()
student_ben.courses.append(existing_math) # Adiciona um curso existente
db_session.add(student_ben)
db_session.commit()

# Consulta de alunos e seus cursos
anna = db_session.query(Student).filter_by(name='Anna').first()
if anna:
    print(f"Cursos de {anna.name}:")
    for course in anna.courses:
        print(f"- {course.title}")

# Consulta de cursos e seus alunos
math_course = db_session.query(Course).filter_by(title='Matemática Avançada').first()
if math_course:
    print(f"Alunos em '{math_course.title}':")
    for student in math_course.students:
        print(f"- {student.name}")

Consultas com JOIN

O SQLAlchemy ORM facilita a realização de joins para combinar dados de várias tabelas. Você pode usar join() para relacionamentos explícitos.

# from my_models import MyUser, MyBook, Employee, Department, Student, Course # Assumindo os modelos
# db_session = scoped_session(sessionmaker(bind=db_engine)) # Crie uma sessão

# JOIN para Um-para-Muitos (Employee e Department)
# Consulta funcionários e seus departamentos (INNER JOIN por padrão)
employees_with_dept = db_session.query(Employee, Department).join(Department).all()
for emp, dept in employees_with_dept:
    print(f"Funcionário: {emp.full_name}, Departamento: {dept.name}")

# LEFT OUTER JOIN para Um-para-Muitos (inclui departamentos sem funcionários)
departments_with_employees_outer = db_session.query(Department, Employee).join(Employee, isouter=True).all()
# Note que isouter=True no join(Employee) significa que departamentos são o lado esquerdo

# JOIN para Muitos-para-Muitos (Student, student_course_association, Course)
# Consulta alunos e seus cursos
students_and_courses = db_session.query(Student, Course).join(student_course_association).join(Course).all()
for student, course in students_and_courses:
    print(f"Aluno: {student.name}, Curso: {course.title}")

# Ou, se quiser filtrar:
students_in_math = db_session.query(Student).join(student_course_association).join(Course).filter(Course.title == 'Matemática Avançada').all()
for student in students_in_math:
    print(f"Aluno em Matemática Avançada: {student.name}")

scoped_session para Segurança de Threads

Em ambientes multi-thread, como aplicações web (Flask, FastAPI), a gestão de sessões de banco de dados pode ser um desafio. Compartilhar uma única instância de Session globalmente pode levar a problemas de concorrência e dados inconsistentes. O scoped_session do SQLAlchemy resolve isso fornecendo uma sessão por thread ou contexto.

O Problema da Sessão Global

Se você inicializar uma Session diretamente (session = Session()) e a usar como uma variável global ou compartilhada em uma aplicação web, diferentes requisições (que geralmente são processadas em threads separadas) tentarão usar a mesma sessão. Isso pode causar conflitos, dados corrompidos ou erros inesperados, pois as sessões não são inerentemente thread-safe para operações simultâneas.

A Solução: scoped_session

scoped_session é uma fábrica de sessões que gerencia uma instância de Session para o escopo atual (por padrão, o thread atual). Ele usa um objeto threading.local() internamente para garantir que cada thread obtenha sua própria instância de sessão isolada. Quando um thread acessa scoped_session, ele verifica se já existe uma sessão para aquele thread; se não, ele cria uma e a armazena para uso futuro naquele mesmo thread.

Uso de scoped_session

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session

# 1. Configurar o motor de conexão
db_engine = create_engine(
    "mysql+pymysql://root:password@127.0.0.1:3306/my_app_db",
    max_overflow=0,      # Número máximo de conexões extras
    pool_size=10,        # Tamanho do pool de conexões
    pool_timeout=30,     # Tempo de espera por uma conexão
    pool_recycle=3600    # Recicla conexões a cada hora
)

# 2. Criar uma classe SessionMaker
# Esta é a "receita" para criar novas sessões
SessionFactory = sessionmaker(autocommit=False, autoflush=False, bind=db_engine)

# 3. Criar a instância de scoped_session
# Este é o objeto que você usará globalmente em sua aplicação
db_scoped_session = scoped_session(SessionFactory)

# Agora, em qualquer parte do seu código (por exemplo, dentro de uma view do Flask):
# current_user = db_scoped_session.query(MyUser).filter_by(id=1).first()
# ...
# db_scoped_session.commit()
# db_scoped_session.remove() # Importante: liberar a sessão ao final da requisição/thread

# O método `remove()` (ou `close()`) é crucial para liberar a sessão e as conexões de volta ao pool
# Em Flask, isso geralmente é feito em um 'teardown_appcontext' ou 'after_request'

Ao usar db_scoped_session, cada thread de requisição em sua aplicação obterá sua própria sessão isolada, garantindo segurança de thread e prevenindo inconsistências de dados. Lembre-se sempre de chamar db_scoped_session.remove() ou db_scoped_session.close() ao final de cada requisição ou unidade de trabalho para liberar a sessão e suas conexões de volta ao pool.