aulify · Sebenta

UC · Unidade de Competência · UC02831

Interfaces entre BD e linguagens

Sebenta · Drivers, ORM, migrações

50h · 4.5 pontos crédito Curso: T. Desenv. Software, T. Inform. Gestão ↗ Referencial oficial SNQ

Apresentação

UC02831 (50h · 4,5 pts) ensina como ligar código aplicacional à base de dados. Começa com drivers nativos (SQL "à mão") e progride para ORMs (SQLAlchemy), migrações (Alembic), pools de conexão e patterns avançados.

Conexão directa

Drivers Python

BD	Driver
SQLite	`sqlite3` (built-in)
MySQL/MariaDB	`mysql-connector-python`, `PyMySQL`
PostgreSQL	`psycopg2`, `psycopg` (v3), `asyncpg`
SQL Server	`pyodbc`
Oracle	`cx_Oracle`

Padrão básico

import psycopg2

conn = psycopg2.connect(
    host="localhost",
    database="aulify",
    user="app",
    password=os.environ["DB_PASSWORD"]
)
cur = conn.cursor()

cur.execute("SELECT id, nome FROM Cliente WHERE id = %s", (42,))
row = cur.fetchone()
print(row)  # (42, 'Ana Silva')

cur.close()
conn.close()

with automatiza fecho:

with psycopg2.connect(...) as conn:
    with conn.cursor() as cur:
        cur.execute("SELECT 1")
        print(cur.fetchone())
# conn fecha automaticamente

Prepared statements

Regra crítica: nunca concatenar input em SQL.

# ERRADO — SQL injection!
nome = request.form["nome"]
cur.execute(f"SELECT * FROM users WHERE nome = '{nome}'")

# CORRECTO — placeholder
cur.execute("SELECT * FROM users WHERE nome = %s", (nome,))

O driver substitui %s (ou ? em sqlite3) escapando o valor automaticamente.

Múltiplos parâmetros:

cur.execute(
    "INSERT INTO Cliente (nome, email, idade) VALUES (%s, %s, %s)",
    ("Ana Silva", "ana@ex.pt", 17)
)

Cursor patterns

# Uma linha
cur.execute("SELECT * FROM Cliente WHERE id = %s", (1,))
row = cur.fetchone()    # tupla ou None

# Todas
cur.execute("SELECT * FROM Cliente")
rows = cur.fetchall()   # lista de tuplas

# Iterar (streaming, eficiente para muitos dados)
cur.execute("SELECT * FROM Encomenda")
for row in cur:
    processar(row)

# Como dict
from psycopg2.extras import RealDictCursor
cur = conn.cursor(cursor_factory=RealDictCursor)
cur.execute("SELECT id, nome FROM Cliente WHERE id = %s", (1,))
row = cur.fetchone()
print(row["nome"])

Transações

try:
    cur.execute("UPDATE Conta SET saldo = saldo - 100 WHERE id = 1")
    cur.execute("UPDATE Conta SET saldo = saldo + 100 WHERE id = 2")
    conn.commit()
except Exception:
    conn.rollback()
    raise

with conn: faz commit em sucesso, rollback em excepção:

with conn:
    with conn.cursor() as cur:
        cur.execute("...")
        cur.execute("...")
# commit automático aqui

ORM · SQLAlchemy

Porque ORM

Pros: - Modela BD como classes Python. - Menos boilerplate. - Migrações automáticas. - Portável entre BDs. - Composição segura de queries.

Contras: - Curva de aprendizagem. - Pode gerar queries ineficientes (N+1). - Abstracção pode esconder comportamento.

Para projectos médios+, vale a pena.

Setup

pip install sqlalchemy psycopg2-binary

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, DECIMAL, DateTime
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from datetime import datetime

Base = declarative_base()

engine = create_engine(
    "postgresql://app:pwd@localhost/aulify",
    echo=False  # True para ver SQL gerado
)
Session = sessionmaker(bind=engine)

Definir modelos

class Cliente(Base):
    __tablename__ = "cliente"

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

    encomendas = relationship("Encomenda", back_populates="cliente",
                              cascade="all, delete-orphan")

    def __repr__(self):
        return f"<Cliente(id={self.id}, nome={self.nome!r})>"


class Encomenda(Base):
    __tablename__ = "encomenda"

    id = Column(Integer, primary_key=True)
    cliente_id = Column(Integer, ForeignKey("cliente.id"), nullable=False)
    total = Column(DECIMAL(10, 2))
    data = Column(DateTime, default=datetime.utcnow)

    cliente = relationship("Cliente", back_populates="encomendas")

Criar tabelas

Base.metadata.create_all(engine)

Em produção usa-se Alembic (ver à frente). Para desenvolvimento rápido create_all é OK.

CRUD

session = Session()

# CREATE
ana = Cliente(nome="Ana Silva", email="ana@ex.pt")
session.add(ana)
session.commit()
# session refresh: ana.id agora tem valor

# READ
ana = session.query(Cliente).filter_by(email="ana@ex.pt").first()
todos = session.query(Cliente).all()
n = session.query(Cliente).count()

# UPDATE
ana.nome = "Ana Maria Silva"
session.commit()

# DELETE
session.delete(ana)
session.commit()

session.close()

Queries

Filtros:

# WHERE simples
session.query(Cliente).filter_by(email="ana@ex.pt").first()
session.query(Cliente).filter(Cliente.nome.like("A%")).all()
session.query(Cliente).filter(Cliente.id.in_([1, 2, 3])).all()
session.query(Cliente).filter(Cliente.nome != None).all()

# Múltiplos
session.query(Encomenda).filter(
    Encomenda.total > 100,
    Encomenda.data >= "2026-01-01"
).all()

# OR
from sqlalchemy import or_
session.query(Cliente).filter(
    or_(Cliente.email.like("%@gmail.com"),
        Cliente.email.like("%@yahoo.com"))
).all()

Ordenação:

session.query(Cliente).order_by(Cliente.nome).all()
session.query(Cliente).order_by(Cliente.criado.desc()).limit(10).all()

Agregações:

from sqlalchemy import func

session.query(func.count(Cliente.id)).scalar()  # número total
session.query(func.sum(Encomenda.total)).scalar()
session.query(func.avg(Encomenda.total)).scalar()

# Group by
results = (session.query(Encomenda.cliente_id, func.sum(Encomenda.total))
                  .group_by(Encomenda.cliente_id)
                  .all())

JOINs:

# Implícito (via relationship)
ana = session.query(Cliente).first()
for enc in ana.encomendas:    # lazy load: query extra!
    print(enc.total)

# Explícito
results = (session.query(Cliente, Encomenda)
                  .join(Encomenda)
                  .filter(Encomenda.total > 100)
                  .all())
for cliente, enc in results:
    print(cliente.nome, enc.total)

N+1 problem

# MAU
clientes = session.query(Cliente).all()
for c in clientes:
    print(c.nome, len(c.encomendas))
# Com 100 clientes: 1 + 100 queries

# BOM — eager loading
from sqlalchemy.orm import joinedload

clientes = session.query(Cliente).options(
    joinedload(Cliente.encomendas)
).all()
# 1 query com JOIN

Outras estratégias: subqueryload, selectinload (mais eficiente em colecções grandes).

Migrações com Alembic

Schema da BD muda ao longo do tempo. Migrações controladas evitam dor:

pip install alembic
alembic init migrations

Editar alembic.ini para apontar à BD; editar migrations/env.py para detectar modelos.

# Gerar migração comparando modelos com BD actual
alembic revision --autogenerate -m "Adicionar coluna telefone"

# Aplicar
alembic upgrade head

# Reverter última
alembic downgrade -1

# Estado actual
alembic current

Cada migração é um ficheiro Python:

def upgrade():
    op.add_column("cliente", sa.Column("telefone", sa.String(20)))

def downgrade():
    op.drop_column("cliente", "telefone")

Commit dos ficheiros de migração no Git. Equipa toda tem a mesma BD evolutiva.

Connection pool

Abrir conexões à BD é caro (TCP handshake, autenticação, etc.). Pool mantém conexões abertas e reutiliza:

engine = create_engine(
    "postgresql://app:pwd@host/db",
    pool_size=10,          # conexões mantidas
    max_overflow=20,       # adicionais temporárias
    pool_pre_ping=True,    # verifica antes de usar
    pool_recycle=3600,     # reciclar a cada hora
)

Em aplicações web, cada request pega uma conexão do pool e devolve-a.

Padrões avançados

Repository pattern

Encapsula acesso a dados numa classe:

class ClienteRepository:
    def __init__(self, session):
        self.session = session

    def get(self, id):
        return self.session.query(Cliente).get(id)

    def get_by_email(self, email):
        return self.session.query(Cliente).filter_by(email=email).first()

    def list_active(self):
        return self.session.query(Cliente).filter_by(activo=True).all()

    def save(self, cliente):
        self.session.add(cliente)
        self.session.commit()

Vantagens: lógica de query centralizada, fácil mockar em testes.

Unit of Work

Agrupa várias operações como unidade atómica. SQLAlchemy session já é unit of work — commit aplica todas as alterações pendentes.

NoSQL — visão geral

Bases não-relacionais para casos específicos:

Tipo	Exemplo	Uso ideal
Documento	MongoDB	Conteúdo estruturado mas flexível
Key-value	Redis	Cache, sessões
Grafo	Neo4j	Redes sociais, recomendações
Coluna	Cassandra	Big data, escrita massiva
Pesquisa	Elasticsearch	Pesquisa full-text

Exemplo MongoDB com pymongo:

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017")
db = client.aulify
col = db.alunos

# Insert
col.insert_one({"nome": "Ana", "notas": [15, 16, 14]})

# Find
ana = col.find_one({"nome": "Ana"})

# Update
col.update_one({"nome": "Ana"}, {"$set": {"idade": 18}})

Boas práticas

Sempre prepared statements — nunca concatenar.
Connection pool em apps web.
Migrações versionadas em projectos sérios.
Eager loading para evitar N+1.
Logs de queries lentas em produção.
Backup automatizado e testado.
Pool stats para tunar pool_size correcto.
Read replicas em apps com muita leitura.

Apêndices

A · Cheatsheet SQLAlchemy

session.query(M).filter(M.col == val).first()
session.query(M).filter(M.col.like("%abc%")).all()
session.query(M).filter(M.col.in_([1,2,3])).all()
session.query(M).order_by(M.col.desc()).limit(10).all()
session.query(M).count()
session.query(func.sum(M.col)).scalar()
session.query(M).options(joinedload(M.rel)).all()
session.add(obj); session.commit()
session.delete(obj); session.commit()

B · Glossário

Connection pool. Conjunto reutilizável de conexões. Driver. Biblioteca que liga linguagem a BD específica. Migração. Mudança versionada de schema. N+1. Anti-pattern: 1 query principal + N para relacionados. ORM. Object-Relational Mapping.

C · Recursos

SQLAlchemy docs · sqlalchemy.org
Alembic docs · alembic.sqlalchemy.org
Real Python · ORM · realpython.com/python-sqlalchemy/