Banco de Dados com SQLite e SQLAlchemy

Arquivos JSON e CSV funcionam bem para dados simples, mas quando o volume cresce, as consultas ficam complexas ou múltiplos processos precisam acessar os mesmos dados simultaneamente, um banco de dados relacional é a solução adequada. Python oferece suporte nativo ao SQLite e integração elegante com bancos maiores através do SQLAlchemy — o ORM mais usado no ecossistema Python.

SQLite: Banco de Dados Embutido

SQLite é um banco de dados relacional que armazena tudo em um único arquivo. Não requer instalação de servidor — ideal para desenvolvimento, testes e aplicações de pequeno porte.

import sqlite3

# Conectando — cria o arquivo se não existir
conn = sqlite3.connect("escola.db")

# Em memória — útil para testes
conn_mem = sqlite3.connect(":memory:")

# Cursor — executa comandos SQL
cursor = conn.cursor()

# Criando tabela
cursor.execute("""
    CREATE TABLE IF NOT EXISTS alunos (
        id       INTEGER PRIMARY KEY AUTOINCREMENT,
        nome     TEXT    NOT NULL,
        email    TEXT    UNIQUE NOT NULL,
        nota     REAL    DEFAULT 0.0,
        ativo    INTEGER DEFAULT 1,
        criado_em TEXT   DEFAULT (datetime('now'))
    )
""")

conn.commit()
conn.close()

CRUD com sqlite3

import sqlite3
from contextlib import contextmanager

@contextmanager
def get_conn(db_path="escola.db"):
    """Gerenciador de contexto para conexões SQLite."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row   # resultados como dicionários
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


# CREATE — inserindo dados
def inserir_aluno(nome, email, nota):
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO alunos (nome, email, nota) VALUES (?, ?, ?)",
            (nome, email, nota)
        )


# READ — consultando dados
def buscar_alunos(nota_minima=0.0):
    with get_conn() as conn:
        cursor = conn.execute(
            "SELECT * FROM alunos WHERE nota >= ? ORDER BY nota DESC",
            (nota_minima,)
        )
        return [dict(row) for row in cursor.fetchall()]


def buscar_por_id(aluno_id):
    with get_conn() as conn:
        cursor = conn.execute(
            "SELECT * FROM alunos WHERE id = ?",
            (aluno_id,)
        )
        row = cursor.fetchone()
        return dict(row) if row else None


# UPDATE — atualizando dados
def atualizar_nota(aluno_id, nova_nota):
    with get_conn() as conn:
        conn.execute(
            "UPDATE alunos SET nota = ? WHERE id = ?",
            (nova_nota, aluno_id)
        )


# DELETE lógico — desativa em vez de apagar
def desativar_aluno(aluno_id):
    with get_conn() as conn:
        conn.execute(
            "UPDATE alunos SET ativo = 0 WHERE id = ?",
            (aluno_id,)
        )


# Populando o banco
inserir_aluno("Ana Silva",    "ana@email.com",    9.5)
inserir_aluno("Bruno Costa",  "bruno@email.com",  7.0)
inserir_aluno("Carla Souza",  "carla@email.com",  8.5)
inserir_aluno("Diego Lima",   "diego@email.com",  5.5)

print("Alunos com nota >= 7.0:")
for aluno in buscar_alunos(7.0):
    print(f"  {aluno['nome']:15} — {aluno['nota']}")

Sempre use parâmetros (?) em vez de f-strings para montar SQL — isso previne SQL Injection.

Duas observações sobre o código acima. A primeira: o get_conn existe porque o with da própria conexão do sqlite3 não fecha a conexão — ele só faz commit na saída normal e rollback na exceção, e a conexão continua aberta depois do bloco. Quem troca o gerenciador por with sqlite3.connect(...) as conn acha que ganhou o fechamento e não ganhou. A segunda: execute o script duas vezes e a segunda para em sqlite3.IntegrityError: UNIQUE constraint failed: alunos.email. É a restrição UNIQUE fazendo o trabalho dela; para uma carga que pode rodar de novo, o SQLite aceita INSERT ... ON CONFLICT(email) DO NOTHING, que ignora a linha repetida.

Consultas Avançadas

def estatisticas_turma():
    with get_conn() as conn:
        cursor = conn.execute("""
            SELECT
                COUNT(*)           AS total,
                AVG(nota)          AS media,
                MAX(nota)          AS maior,
                MIN(nota)          AS menor,
                SUM(CASE WHEN nota >= 6 THEN 1 ELSE 0 END) AS aprovados
            FROM alunos
            WHERE ativo = 1
        """)
        return dict(cursor.fetchone())


def alunos_por_faixa():
    with get_conn() as conn:
        cursor = conn.execute("""
            SELECT
                CASE
                    WHEN nota >= 9 THEN 'Excelente'
                    WHEN nota >= 7 THEN 'Bom'
                    WHEN nota >= 6 THEN 'Regular'
                    ELSE 'Insuficiente'
                END AS conceito,
                COUNT(*) AS quantidade
            FROM alunos
            WHERE ativo = 1
            GROUP BY conceito
            ORDER BY MIN(nota) DESC
        """)
        return [dict(row) for row in cursor.fetchall()]


stats = estatisticas_turma()
print(f"\nEstatísticas:")
print(f"  Total:     {stats['total']}")
print(f"  Média:     {stats['media']:.2f}")
print(f"  Aprovados: {stats['aprovados']}")

print("\nDistribuição por conceito:")
for faixa in alunos_por_faixa():
    print(f"  {faixa['conceito']:15}: {faixa['quantidade']}")

SQLAlchemy: ORM Moderno

SQLAlchemy permite trabalhar com bancos de dados usando classes Python em vez de SQL bruto. A versão 2.0 trouxe uma API mais limpa e moderna:

pip install sqlalchemy

from sqlalchemy import (
    create_engine, Column, Integer, String,
    Float, Boolean, DateTime, ForeignKey, text
)
from sqlalchemy.orm import DeclarativeBase, relationship, Session
from datetime import datetime, UTC


# Engine — conexão com o banco
engine = create_engine(
    "sqlite:///escola_orm.db",
    echo=False   # echo=True mostra SQL gerado — útil para debug
)


# Base para os modelos
class Base(DeclarativeBase):
    pass


# Modelos
class Turma(Base):
    __tablename__ = "turmas"

    id   = Column(Integer, primary_key=True, autoincrement=True)
    nome = Column(String(50), nullable=False, unique=True)
    ano  = Column(Integer, nullable=False)

    alunos = relationship("Aluno", back_populates="turma")

    def __repr__(self):
        return f"Turma(id={self.id}, nome='{self.nome}')"


class Aluno(Base):
    __tablename__ = "alunos"

    id        = Column(Integer, primary_key=True, autoincrement=True)
    nome      = Column(String(100), nullable=False)
    email     = Column(String(100), unique=True, nullable=False)
    nota      = Column(Float, default=0.0)
    ativo     = Column(Boolean, default=True)
    criado_em = Column(DateTime, default=lambda: datetime.now(UTC))
    turma_id  = Column(Integer, ForeignKey("turmas.id"))

    turma = relationship("Turma", back_populates="alunos")

    @property
    def aprovado(self):
        return self.nota >= 6.0

    def __repr__(self):
        return f"Aluno(id={self.id}, nome='{self.nome}', nota={self.nota})"


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

O default de criado_em recebe uma função, e não um valor, para ser chamada a cada inserção — com default=datetime.now(UTC), todas as linhas teriam a hora em que o módulo foi importado. O datetime.utcnow que se vê em muito material está obsoleto desde o Python 3.12 e emite um DeprecationWarning a cada linha inserida.

Dois avisos sobre o SQLite por trás do ORM. Ele não verifica chave estrangeira por padrão: o ForeignKey("turmas.id") cria a restrição no esquema, mas um aluno com turma_id=999 é gravado sem erro até que cada conexão execute PRAGMA foreign_keys=ON — com SQLAlchemy, num event.listens_for(engine, "connect"). E os modelos acima usam Column, que continua funcionando na versão 2.0, mas não é o estilo que a documentação atual ensina: nela, cada coluna é nome: Mapped[str] = mapped_column(String(100)), e a anotação de tipo passa a informar o nullable e a alimentar o verificador de tipos.

CRUD com SQLAlchemy 2.0

from sqlalchemy.orm import Session
from sqlalchemy import select, update, delete
from sqlalchemy.orm import selectinload

# CREATE
def criar_dados_iniciais():
    with Session(engine) as session:
        turma_a = Turma(nome="Turma A", ano=2024)
        turma_b = Turma(nome="Turma B", ano=2024)
        session.add_all([turma_a, turma_b])
        session.flush()   # obtém IDs sem commitar

        alunos = [
            Aluno(nome="Ana Silva",   email="ana@email.com",   nota=9.5, turma=turma_a),
            Aluno(nome="Bruno Costa", email="bruno@email.com", nota=7.0, turma=turma_a),
            Aluno(nome="Carla Souza", email="carla@email.com", nota=8.5, turma=turma_b),
            Aluno(nome="Diego Lima",  email="diego@email.com", nota=5.5, turma=turma_b),
        ]
        session.add_all(alunos)
        session.commit()
        print("Dados criados com sucesso.")


# READ
def listar_alunos_aprovados():
    with Session(engine) as session:
        stmt = (
            select(Aluno)
            .options(selectinload(Aluno.turma))   # carrega a turma antes de fechar a sessão
            .where(Aluno.nota >= 6.0)
            .where(Aluno.ativo == True)
            .order_by(Aluno.nota.desc())
        )
        return session.scalars(stmt).all()


def buscar_com_turma(turma_nome: str):
    with Session(engine) as session:
        stmt = (
            select(Aluno)
            .join(Turma)
            .where(Turma.nome == turma_nome)
            .order_by(Aluno.nome)
        )
        return session.scalars(stmt).all()


# UPDATE
def atualizar_nota(email: str, nova_nota: float):
    with Session(engine) as session:
        stmt = (
            update(Aluno)
            .where(Aluno.email == email)
            .values(nota=nova_nota)
        )
        session.execute(stmt)
        session.commit()


# DELETE (lógico)
def desativar_aluno(email: str):
    with Session(engine) as session:
        stmt = (
            update(Aluno)
            .where(Aluno.email == email)
            .values(ativo=False)
        )
        session.execute(stmt)
        session.commit()


criar_dados_iniciais()

print("\nAlunos aprovados:")
for aluno in listar_alunos_aprovados():
    status = "✓" if aluno.aprovado else "✗"
    turma  = aluno.turma.nome if aluno.turma else "—"
    print(f"  {status} {aluno.nome:15} — {aluno.nota} ({turma})")

print("\nAlunos da Turma A:")
for aluno in buscar_com_turma("Turma A"):
    print(f"  {aluno.nome}")

O selectinload em listar_alunos_aprovados não é enfeite. Sem ele, o laço final levanta DetachedInstanceError: Parent instance <Aluno> is not bound to a Session; lazy load operation of attribute 'turma' cannot proceed. Um relationship é carregado sob demanda por padrão: a turma só é buscada no banco quando alguém lê aluno.turma, e a função devolve os alunos depois que o with Session já fechou a sessão. O selectinload busca as turmas numa segunda consulta, ainda dentro da sessão. É também por isso que o salvar do repositório adiante chama session.refresh: o commit expira os atributos do objeto, e sem recarregá-los ali o aluno devolvido não pode mais ser lido.

Migrações com Alembic

Para projetos reais, use Alembic para gerenciar mudanças no esquema do banco:

pip install alembic
alembic init migrations

# migrations/env.py — configuração básica
from sqlalchemy import engine_from_config
from alembic import context
from meus_modelos import Base

target_metadata = Base.metadata

# Gerando uma migração automaticamente
alembic revision --autogenerate -m "adicionar coluna telefone em alunos"

# Aplicando migrações
alembic upgrade head

# Revertendo
alembic downgrade -1

Exemplo Completo: Repositório com SQLAlchemy

from sqlalchemy import create_engine, select, func
from sqlalchemy.orm import Session
from typing import List, Optional


class AlunoRepositorio:
    """Encapsula todas as operações de banco para Aluno."""

    def __init__(self, engine):
        self._engine = engine

    def salvar(self, aluno: Aluno) -> Aluno:
        with Session(self._engine) as session:
            session.add(aluno)
            session.commit()
            session.refresh(aluno)
            return aluno

    def buscar_por_id(self, aluno_id: int) -> Optional[Aluno]:
        with Session(self._engine) as session:
            return session.get(Aluno, aluno_id)

    def buscar_por_email(self, email: str) -> Optional[Aluno]:
        with Session(self._engine) as session:
            stmt = select(Aluno).where(Aluno.email == email)
            return session.scalars(stmt).first()

    def listar_todos(self, apenas_ativos: bool = True) -> List[Aluno]:
        with Session(self._engine) as session:
            stmt = select(Aluno)
            if apenas_ativos:
                stmt = stmt.where(Aluno.ativo == True)
            return session.scalars(stmt).all()

    def media_notas(self, turma_id: int = None) -> float:
        with Session(self._engine) as session:
            stmt = select(func.avg(Aluno.nota)).where(Aluno.ativo == True)
            if turma_id:
                stmt = stmt.where(Aluno.turma_id == turma_id)
            resultado = session.scalar(stmt)
            return round(resultado or 0.0, 2)

    def top_alunos(self, n: int = 3) -> List[Aluno]:
        with Session(self._engine) as session:
            stmt = (
                select(Aluno)
                .where(Aluno.ativo == True)
                .order_by(Aluno.nota.desc())
                .limit(n)
            )
            return session.scalars(stmt).all()


repo = AlunoRepositorio(engine)

print(f"\nMédia geral: {repo.media_notas()}")
print("\nTop 3 alunos:")
for aluno in repo.top_alunos(3):
    print(f"  {aluno.nome:15} — {aluno.nota}")

O SQLite resolve a maior parte do que se pede a um banco em desenvolvimento, testes e aplicações pequenas: um arquivo, nenhum servidor, SQL de verdade. O sqlite3 da biblioteca padrão é suficiente para usá-lo bem, desde que a conexão seja fechada por alguém — o with dela só cuida da transação — e que todo valor entre na consulta como parâmetro, nunca por f-string. Vale lembrar também que o SQLite é permissivo em pontos onde outros bancos são rígidos: chave estrangeira só é verificada com PRAGMA foreign_keys=ON, e fuso horário gravado numa coluna de data não volta na leitura.

O SQLAlchemy troca o SQL escrito à mão por consultas montadas com select() e por classes mapeadas para tabelas, e com isso o código passa a mudar de banco trocando a URL do engine. O preço é entender a Session. Um relacionamento é carregado sob demanda, e lê-lo depois que a sessão fechou levanta DetachedInstanceError; lê-lo dentro de um laço, com a sessão aberta, faz uma consulta por item, e o selectinload resolve os dois problemas. O commit expira os atributos, o que explica o refresh do repositório. E o esquema, quando o projeto for real, evolui por migrações do Alembic, revisadas antes de aplicar — o --autogenerate sugere, não decide.

Fontes e leituras recomendadas

• sqlite3 — documentação oficial — https://docs.python.org/3/library/sqlite3.html
• SQLAlchemy ORM — tutorial oficial — https://docs.sqlalchemy.org/en/20/orm/quickstart.html
• SQLAlchemy 2.0 — guia de migração — https://docs.sqlalchemy.org/en/20/changelog/migration_20.html
• Alembic — migrações de banco — https://alembic.sqlalchemy.org/en/latest/
• SQL Injection — OWASP — https://owasp.org/www-community/attacks/SQL_Injection
• MYERS, Jason; COPELAND, Rick. Essential SQLAlchemy. 2. ed. O'Reilly Media, 2016. — guia de referência do SQLAlchemy, cobrindo Core e ORM.
• PERCIVAL, Harry; GREGORY, Bob. Architecture Patterns with Python. O'Reilly Media, 2020. Cap. 2 — padrão Repository com SQLAlchemy em projetos reais.

Exercícios

Exercício 1

Um relatório permite ao usuário escolher a coluna de ordenação num menu, e o desenvolvedor, seguindo a regra de nunca montar SQL com f-string, escreveu conn.execute("SELECT nome FROM alunos ORDER BY ? DESC", (coluna,)). O código roda sem erro, mas a lista aparece sempre na mesma ordem, qualquer que seja a coluna escolhida. Explique e corrija sem abrir espaço para SQL Injection.

Exercício 2

Uma página lista as 200 turmas da escola com a quantidade de alunos de cada uma: for t in session.scalars(select(Turma)).all(): print(t.nome, len(t.alunos)). Com SQLite, na máquina do desenvolvedor, a página abre instantaneamente. Em produção, com o banco num servidor separado, ela leva vários segundos. Explique a diferença e proponha duas correções, dizendo quando usar cada uma.

Exercício 3

Um modelo grava criado_em com mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC)), seguindo a recomendação de usar datas aware. Os testes rodam em SQLite. Um teste compara o valor lido do banco com datetime.now(UTC) - timedelta(minutes=1) e falha com TypeError: can't compare offset-naive and offset-aware datetimes. O desenvolvedor jura que gravou com fuso. Explique.

Exercício 4

Um painel administrativo apaga turmas com session.delete(turma); session.commit(). A turma A tinha 20 alunos. Depois da exclusão, os alunos continuam no banco, mas somem de todos os relatórios por turma, e ninguém recebeu mensagem de erro. O modelo é o deste artigo, com ForeignKey("turmas.id") em Aluno.turma_id. O que aconteceu com os alunos, e que alternativas o modelo deveria considerar?

Exercício 5

No início do semestre, antes de qualquer matrícula, a tela de estatísticas da turma, que usa a função estatisticas_turma() deste artigo, derruba o sistema com TypeError: unsupported format string passed to NoneType.__format__. No semestre anterior, com alunos cadastrados, ela funcionou sempre. Explique, corrija e diga o que o caso ensina sobre funções de agregação em SQL.