Pular para o conteúdo

    Capítulo 55, Backend

    PostgreSQL e SQL

    O banco de dados é, quase sempre, a parte mais duradoura de um sistema. Eu ensino SQL com o SQLite, que vem com o Python e não exige instalar nada, e depois mostro o que muda no PostgreSQL.

    Código deste capítulo: backend/cap55_sql.py

    Por que SQL, e por que PostgreSQL

    SQL é a linguagem declarativa que descreve o que você quer, e o banco decide como buscar. O PostgreSQL é o banco relacional que eu escolho por padrão: é aberto, extremamente confiável, tem tipos ricos (JSON, arrays, datas com fuso) e transações sólidas. Os conceitos deste capítulo valem para qualquer banco relacional. O SQLite serve para aprender e para testes rápidos, e eu mostro onde o PostgreSQL difere.

    Tabelas e restrições

    As restrições (NOT NULL, UNIQUE, CHECK, REFERENCES) fazem o banco recusar dado inválido, mesmo que um bug no código tente gravá-lo. Eu sempre as declaro, porque o banco é a última linha de defesa:

    backend/cap55_sql.pylinhas 10 a 28
    import sqlite3

conexao = sqlite3.connect(":memory:")
conexao.row_factory = sqlite3.Row
conexao.execute("PRAGMA foreign_keys = ON")
conexao.executescript("""
CREATE TABLE clientes (
    id INTEGER PRIMARY KEY,
    nome TEXT NOT NULL,
    cidade TEXT
);
CREATE TABLE pedidos (
    id INTEGER PRIMARY KEY,
    cliente_id INTEGER NOT NULL REFERENCES clientes(id),
    total_centavos INTEGER NOT NULL CHECK (total_centavos > 0)
);
INSERT INTO clientes (nome, cidade) VALUES ('Ana', 'Recife'), ('Bia', 'Natal'), ('Caio', 'Recife');
INSERT INTO pedidos (cliente_id, total_centavos) VALUES (1, 5000), (1, 2500), (2, 10000);
""")

    Valores monetários ficam em centavos inteiros (INTEGER) ou em NUMERIC, nunca em FLOAT, pelo mesmo motivo do capítulo de tipos.

    Consultar: SELECT, JOIN e agregação

    backend/cap55_sql.pylinhas 33 a 42
    consulta = "SELECT nome FROM clientes WHERE cidade = 'Recife' ORDER BY nome"
print([linha["nome"] for linha in conexao.execute(consulta)])

juncao = """
SELECT c.nome, p.total_centavos
FROM pedidos p
JOIN clientes c ON c.id = p.cliente_id
ORDER BY p.id
"""
print([tuple(linha) for linha in conexao.execute(juncao)])
    Saída
    ['Ana', 'Caio']
[('Ana', 5000), ('Ana', 2500), ('Bia', 10000)]

    O JOIN (ou INNER JOIN) só devolve clientes que têm pedido. O LEFT JOIN mantém todos os clientes, e quem não tem pedido aparece com NULL. O GROUP BY agrupa linhas e o HAVING filtra os grupos (o WHERE filtra linhas antes de agrupar):

    backend/cap55_sql.pylinhas 44 a 59
    agregacao = """
SELECT c.nome, COALESCE(SUM(p.total_centavos), 0) AS total
FROM clientes c
LEFT JOIN pedidos p ON p.cliente_id = c.id
GROUP BY c.nome
ORDER BY c.nome
"""
print([tuple(linha) for linha in conexao.execute(agregacao)])

com_filtro = """
SELECT cliente_id, COUNT(*) AS quantidade
FROM pedidos
GROUP BY cliente_id
HAVING COUNT(*) > 1
"""
print([tuple(linha) for linha in conexao.execute(com_filtro)])
    Saída
    [('Ana', 7500), ('Bia', 10000), ('Caio', 0)]
[(1, 2)]

    O COALESCE troca o NULL do Caio (sem pedidos) por zero. Sem o LEFT JOIN, ele nem apareceria no resultado.

    Parâmetros: nunca monte SQL com texto

    Montar o SQL com f-string ou concatenação é a vulnerabilidade mais antiga e ainda mais comum: a injeção de SQL. A entrada do usuário vira parte do comando. Os parâmetros enviam o valor separado do comando, e o banco jamais o interpreta como SQL:

    backend/cap55_sql.pylinhas 64 a 67
    entrada = "x' OR '1'='1"
inseguro = conexao.execute(f"SELECT nome FROM clientes WHERE nome = '{entrada}'").fetchall()
seguro = conexao.execute("SELECT nome FROM clientes WHERE nome = ?", (entrada,)).fetchall()
print(len(inseguro), len(seguro))
    Saída
    3 0

    A versão com f-string devolveu todos os clientes (3), porque a entrada reescreveu a condição. A versão com parâmetro procurou por um nome literalmente estranho e achou zero. O marcador é ? no SQLite e %s no psycopg, mas a regra é a mesma em qualquer driver.

    Transações: tudo ou nada

    Uma transação agrupa operações que precisam acontecer juntas. Transferir dinheiro é o exemplo clássico: debitar de uma conta e creditar em outra não pode ficar pela metade. O with conexao confirma (commit) se o bloco termina bem e desfaz (rollback) se levantar exceção:

    backend/cap55_sql.pylinhas 72 a 88
    conexao.execute("CREATE TABLE contas (id INTEGER PRIMARY KEY, saldo INTEGER NOT NULL CHECK (saldo >= 0))")
conexao.executemany("INSERT INTO contas VALUES (?, ?)", [(1, 100), (2, 50)])
conexao.commit()


def transferir(origem, destino, valor):
    with conexao:
        conexao.execute("UPDATE contas SET saldo = saldo - ? WHERE id = ?", (valor, origem))
        conexao.execute("UPDATE contas SET saldo = saldo + ? WHERE id = ?", (valor, destino))


transferir(1, 2, 30)
try:
    transferir(1, 2, 500)
except sqlite3.IntegrityError as erro:
    print("recusado:", erro)
print([tuple(linha) for linha in conexao.execute("SELECT id, saldo FROM contas ORDER BY id")])
    Saída
    recusado: CHECK constraint failed: saldo >= 0
[(1, 70), (2, 80)]

    A segunda transferência falhou no débito (saldo negativo viola o CHECK), e o rollback garantiu que nada mudou: os saldos são 70 e 80, que somam os mesmos 150 de antes. Essas garantias (atomicidade, consistência, isolamento e durabilidade) são o ACID.

    Índices e o plano de consulta

    Sem índice, buscar por uma coluna obriga o banco a ler a tabela inteira. Um índice é uma estrutura ordenada auxiliar que torna a busca rápida, ao custo de espaço e de escrita um pouco mais lenta. O comando EXPLAIN mostra como o banco pretende executar a consulta:

    backend/cap55_sql.pylinhas 93 a 101
    def usa_indice(sql):
    plano = conexao.execute("EXPLAIN QUERY PLAN " + sql).fetchall()
    return any("INDEX" in linha["detail"] for linha in plano)


busca = "SELECT * FROM pedidos WHERE cliente_id = 1"
print("antes do índice:", usa_indice(busca))
conexao.execute("CREATE INDEX ix_pedidos_cliente ON pedidos (cliente_id)")
print("depois do índice:", usa_indice(busca))
    Saída
    antes do índice: False
depois do índice: True

    Crie índices para as colunas que aparecem em WHERE, em JOIN e em ORDER BY nas consultas frequentes. Não crie um para cada coluna: cada índice custa a cada INSERT.

    O que muda no PostgreSQL

    AssuntoSQLitePostgreSQL
    ServidorUm arquivo, sem processoUm servidor, com usuários e rede
    Chave automáticaINTEGER PRIMARY KEYINTEGER GENERATED ALWAYS AS IDENTITY
    Marcador de parâmetro?%s (no psycopg)
    TiposFlexíveis (tipagem fraca)Estritos, com JSONB, ARRAY, TIMESTAMPTZ
    ConcorrênciaUma escrita por vezMuitas escritas simultâneas (MVCC)
    Devolver a linha criadaLimitadoINSERT ... RETURNING

    Com o driver psycopg (versão 3), a conexão também é um gerenciador de contexto: sair do with sem erro confirma a transação, e sair com exceção a desfaz. Os blocos abaixo rodam contra um PostgreSQL de verdade (a variável DATABASE_URL aponta para ele):

    Com PostgreSQL (psycopg)
    import os

import psycopg
from psycopg.types.json import Jsonb

with psycopg.connect(os.environ["DATABASE_URL"]) as conexao:
    conexao.execute("DROP TABLE IF EXISTS produtos")
    conexao.execute("""
        CREATE TABLE produtos (
            id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
            nome TEXT UNIQUE NOT NULL,
            preco_centavos INTEGER NOT NULL,
            atributos JSONB NOT NULL DEFAULT '{}'
        )
    """)
    cursor = conexao.execute(
        "INSERT INTO produtos (nome, preco_centavos, atributos) VALUES (%s, %s, %s) RETURNING id",
        ("caneta", 350, Jsonb({"cor": "azul"})),
    )
    print("id criado:", cursor.fetchone()[0])
    Saída
    id criado: 1

    O RETURNING devolve a chave gerada na mesma operação, sem uma segunda consulta. O ON CONFLICT faz um upsert (insere, ou atualiza se a chave única já existir), e o operador ->> consulta dentro do JSON:

    Upsert e consulta em JSONB
    import os

import psycopg

with psycopg.connect(os.environ["DATABASE_URL"]) as conexao:
    conexao.execute(
        """INSERT INTO produtos (nome, preco_centavos) VALUES (%s, %s)
           ON CONFLICT (nome) DO UPDATE SET preco_centavos = EXCLUDED.preco_centavos""",
        ("caneta", 400),
    )
    print(conexao.execute("SELECT nome, preco_centavos FROM produtos").fetchall())
    print(conexao.execute("SELECT nome FROM produtos WHERE atributos->>'cor' = %s", ("azul",)).fetchall())

with psycopg.connect(os.environ["DATABASE_URL"]) as conexao:
    try:
        conexao.execute("INSERT INTO produtos (nome, preco_centavos) VALUES (%s, %s)", ("caneta", 1))
    except psycopg.errors.UniqueViolation as erro:
        print("violação de unicidade:", erro.diag.constraint_name)
    conexao.rollback()
    conexao.execute("DROP TABLE produtos")
    Saída
    [('caneta', 400)]
[('caneta',)]
violação de unicidade: produtos_nome_key

    Duas transações, dois mundos

    No PostgreSQL, o isolamento padrão (READ COMMITTED) significa que cada comando enxerga o que já foi confirmado por outras transações. Para evitar a corrida de "ler o saldo, calcular, gravar", use SELECT ... FOR UPDATE, que trava a linha até o fim da transação, ou uma restrição no banco. Concorrência de dados é problema do banco, e não do Python.

    Exercício 1

    Total por cliente, incluindo quem não comprou

    Escreva total_por_cliente() que use a conexão acima e devolva uma lista de (nome, total) ordenada por nome, com zero para quem não tem pedidos.