Acesso ao Banco via Python

Autor

Douglas Braga

O SQL não é uma linguagem de programação completa: não suporta laços, entrada de usuário, chamadas de rede ou lógica de apresentação. Para construir uma aplicação real, é preciso combinar SQL com uma linguagem hospedeira — aqui, Python.

A abordagem padrão é usar uma biblioteca de acesso a banco de dados (database API) que:

  1. Abre uma conexão com o SGBD
  2. Envia instruções SQL pelo canal aberto
  3. Recebe os resultados (linhas, contagens, erros) de volta
  4. Confirma ou desfaz transações

Em Python, a biblioteca padrão para PostgreSQL é o psycopg2, que implementa a especificação DB-API 2.0 (PEP 249) — a mesma interface usada por sqlite3, pymysql e outras.

Instalação

pip install psycopg2-binary

Conectando ao PostgreSQL

import psycopg2
import os

conn = psycopg2.connect(
    dbname   = "curso",
    host     = "localhost",
    port     = 5432,
    user     = "postgres",
    password = os.getenv("POSTGRES_PASSWORD", "postgres")
)
Objeto Papel
conn Representa a sessão com o banco; mantém estado de transação
cur = conn.cursor() Canal pelo qual se enviam SQL e se recebem resultados

Executando Consultas SELECT

cur = conn.cursor()
cur.execute("SELECT nome, salario FROM professor ORDER BY salario DESC")

for nome, salario in cur.fetchall():
    print(f"{nome}: R$ {salario:.2f}")

cur.close()
Helena Carvalho: R$ 6700.00
Carla Pinto: R$ 6600.00
Bruno Teixeira: R$ 6500.00
Felipe Araujo: R$ 6400.00
Gustavo Costa: R$ 3300.00
Eduarda Souza: R$ 3200.00
Igor Melo: R$ 3100.00

Alternativas de fetch:

Método Retorna
fetchone() Próxima linha (ou None)
fetchmany(n) Até n linhas
fetchall() Todas as linhas restantes

SQL Injection — O Perigo das Strings Concatenadas

Aviso

SQL Injection é uma das vulnerabilidades mais críticas (OWASP Top 10). Ocorre quando dados do usuário são inseridos diretamente em uma string SQL — permitindo que um atacante altere a estrutura da consulta.

Exemplo inseguro

O código abaixo concatena o input do usuário diretamente na string SQL:

# VULNERÁVEL — nunca faça isso
escola_id = input("ID da escola: ")

sql = "SELECT nome FROM professor WHERE id_escola = " + escola_id
cur.execute(sql)

Quando o usuário digita 31 OR 1=1 --, a variável sql fica:

SELECT nome FROM professor WHERE id_escola = 31 OR 1=1 --

O fragmento OR 1=1 é sempre verdadeiro — o filtro é completamente ignorado. Executando exatamente essa query no banco:

cur = conn.cursor()

# Simulando o código vulnerável com input malicioso
escola_id_malicioso = "31 OR 1=1 --"
sql_injetado = "SELECT nome FROM professor WHERE id_escola = " + escola_id_malicioso

cur.execute(sql_injetado)
rows = cur.fetchall()

print(f"Professores retornados com input '{escola_id_malicioso}':")
print(f"  → {len(rows)} linha(s) — deveria retornar apenas os da escola 31\n")
for (nome,) in rows:
    print(f"  {nome}")

cur.close()
Professores retornados com input '31 OR 1=1 --':
  → 7 linha(s) — deveria retornar apenas os da escola 31

  Eduarda Souza
  Felipe Araujo
  Gustavo Costa
  Helena Carvalho
  Igor Melo
  Bruno Teixeira
  Carla Pinto

Todos os professores foram retornados — independente da escola pedida. Além do OR 1=1, um atacante pode usar UNION SELECT para exfiltrar dados de outras tabelas ou ; DROP TABLE para destruir dados.

Solução: Consultas Parametrizadas

Com consultas parametrizadas, o valor do usuário é enviado separado da query — nunca como parte do texto SQL:

# SEGURO — use sempre parâmetros
escola_id = input("ID da escola: ")

cur.execute(
    "SELECT nome FROM professor WHERE id_escola = %s",
    (escola_id,)   # psycopg2 envia como parâmetro $1, nunca concatena no SQL
)

Quando escola_id = "31 OR 1=1 --", psycopg2 envia ao PostgreSQL:

  • Query template: SELECT nome FROM professor WHERE id_escola = $1
  • Parâmetro $1: a string literal 31 OR 1=1 --

O PostgreSQL recebe um texto e tenta compará-lo com a coluna id_escola (tipo INTEGER). Como "31 OR 1=1 --" não é um inteiro válido, o banco rejeita e lança um erro — que o Python captura normalmente:

import psycopg2.errors

cur = conn.cursor()
escola_id_malicioso = "31 OR 1=1 --"

try:
    cur.execute(
        "SELECT nome FROM professor WHERE id_escola = %s",
        (escola_id_malicioso,)
    )
    rows = cur.fetchall()
    print(f"{len(rows)} linha(s) retornada(s)")
except psycopg2.errors.InvalidTextRepresentation as e:
    conn.rollback()  # limpa o estado de erro da conexão
    print("PostgreSQL bloqueou a operação:")
    print(f"  {e.diag.message_primary}")
    print(f"\n  → O input foi tratado como DADO, não como SQL.")
    print(f"  → A estrutura da query permanece intacta.")

cur.close()
PostgreSQL bloqueou a operação:
  invalid input syntax for type integer: "31 OR 1=1 --"

  → O input foi tratado como DADO, não como SQL.
  → A estrutura da query permanece intacta.

Mesmo que o atacante forneça um input que seja um inteiro válido (como "999"), a query simplesmente retorna 0 linhas — nenhum SQL extra é executado:

cur = conn.cursor()
escola_id_invalido = "999"

cur.execute(
    "SELECT nome FROM professor WHERE id_escola = %s",
    (int(escola_id_invalido),)
)
rows = cur.fetchall()
print(f"Escola {escola_id_invalido}: {len(rows)} professor(es) — escola inexistente, 0 linhas.")
cur.close()
Escola 999: 0 professor(es) — escola inexistente, 0 linhas.

Em psycopg2, o marcador de posição é %s para qualquer tipo. A biblioteca envia query e parâmetros separadamente ao servidor via protocolo wire — o SQL enviado nunca contém os dados do usuário concatenados como texto.

Compare com JDBC (Java), onde o marcador é ? em PreparedStatement — o mesmo princípio.

Operações DML (INSERT, UPDATE, DELETE)

cur = conn.cursor()

# INSERT com parâmetros
cur.execute(
    """
    INSERT INTO evento_academico
        (nome, tipo, id_escola, data_inicio, data_fim, hora_inicio, hora_fim)
    VALUES (%s, %s, %s, %s, %s, %s, %s)
    """,
    ("Hackathon UnDF", "Workshop", 31, "2026-09-15", "2026-09-16", "08:00", "22:00")
)

# UPDATE
cur.execute(
    "UPDATE professor SET salario = %s WHERE id_prof = %s",
    (6900.00, 31400002)
)
print(f"Linhas afetadas: {cur.rowcount}")

conn.commit()
cur.close()

Controle de Transações

Por padrão, psycopg2 inicia uma transação implicitamente na primeira instrução. Para confirmar ou desfazer:

cur = conn.cursor()
try:
    # As duas instruções são atômicas: confirmam juntas ou nenhuma confirma
    cur.execute("UPDATE professor SET salario = salario * 1.10 WHERE id_escola = 31")
    cur.execute("UPDATE professor SET salario = salario * 0.95 WHERE id_escola = 21")
    conn.commit()
    print("Transação confirmada.")
except Exception as e:
    conn.rollback()
    print(f"Erro — rollback realizado: {e}")
finally:
    cur.close()

Para operações de leitura simples onde transação explícita não é necessária:

conn.autocommit = True   # cada instrução é confirmada imediatamente

Padrão Completo com Context Manager

import psycopg2, os
from contextlib import closing

dsn = dict(dbname="curso", host="localhost", port=5432,
           user="postgres", password=os.getenv("POSTGRES_PASSWORD","postgres"))

with psycopg2.connect(**dsn) as conn:
    with closing(conn.cursor()) as cur:
        cur.execute("""
            SELECT e.sigla, COUNT(p.id_prof) AS professores
            FROM   escola e
            LEFT JOIN professor p ON p.id_escola = e.id_escola
            GROUP BY e.sigla
            ORDER BY professores DESC
        """)
        for sigla, total in cur.fetchall():
            print(f"{sigla}: {total} professor(es)")
    conn.commit()
Nota

O with psycopg2.connect(...) chama conn.commit() ao sair sem exceção, ou conn.rollback() se uma exceção ocorrer.

Para Praticar

Os trechos abaixo são exercícios para executar localmente (requerem Python + psycopg2):

# Exercício 1: listar alunos de um curso específico
import psycopg2, os

def alunos_do_curso(id_curso: int):
    conn = psycopg2.connect(
        dbname="curso", host="localhost", port=5432,
        user="postgres", password=os.getenv("POSTGRES_PASSWORD","postgres")
    )
    cur = conn.cursor()
    cur.execute("""
        SELECT a.nome, a.ano_ingresso
        FROM   aluno a
        WHERE  a.id_curso = %s
        ORDER BY a.nome
    """, (id_curso,))
    for nome, ano in cur.fetchall():
        print(f"  {nome} ({ano})")
    cur.close()
    conn.close()

alunos_do_curso(311)   # Engenharia de Software
# Exercício 2: matrícula segura — trata violações de FK e UNIQUE
def matricular(conn, id_aluno, id_disciplina, ano, semestre, turma):
    cur = conn.cursor()
    try:
        cur.execute("""
            INSERT INTO matricula_disciplina
                (id_disciplina, ano, semestre, turma, id_aluno)
            VALUES (%s, %s, %s, %s, %s)
        """, (id_disciplina, ano, semestre, turma, id_aluno))
        conn.commit()
        print("Matrícula realizada.")
    except psycopg2.errors.ForeignKeyViolation:
        conn.rollback()
        print("Turma não encontrada — matrícula cancelada.")
    except psycopg2.errors.UniqueViolation:
        conn.rollback()
        print("Aluno já matriculado nesta turma.")
    finally:
        cur.close()