Python

Banco de Dados com SQLite e SQLAlchemy Já leu

• • 8 min de leitura

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 sol

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 — removendo dados
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.

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


# 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=datetime.utcnow)
    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)

CRUD com SQLAlchemy 2.0

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

# 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)
            .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}")

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}")

Resumo

  • SQLite é ideal para desenvolvimento e aplicações leves — sem servidor, arquivo único
  • Sempre use parâmetros (?) em SQL bruto — nunca concatene strings para evitar SQL Injection
  • row_factory = sqlite3.Row retorna resultados como objetos acessíveis por nome de coluna
  • SQLAlchemy ORM mapeia tabelas a classes Python — elimina SQL repetitivo
  • A API 2.0 do SQLAlchemy usa select(), Session e scalars() de forma consistente
  • O padrão Repository encapsula o acesso ao banco, facilitando testes e manutenção
  • Alembic gerencia migrações de esquema de forma versionada e reproduzível

Referências e Leituras Complementares

  • 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
  • BAYER, Mike. Essential SQLAlchemy. 2. ed. O'Reilly Media, 2016. — guia definitivo do SQLAlchemy escrito pelo seu criador.
  • PERCIVAL, Harry; GREGORY, Bob. Architecture Patterns with Python. O'Reilly Media, 2020. Cap. 2 — padrão Repository com SQLAlchemy em projetos reais.
Comentários

Mais em Python

Decoradores e Metaprogramação
Decoradores e Metaprogramação

Decoradores são um dos recursos mais elegantes do Python. Eles permite...

Artigo 39 — Visão Computacional com OpenCV e PIL
Artigo 39 — Visão Computacional com OpenCV e PIL

Visão computacional é a capacidade de máquinas interpretarem e processarem im...

Leitura e Escrita de Arquivos
Leitura e Escrita de Arquivos

Programas raramente existem de forma isolada — eles leem configurações, proce...