Ficha 01 · Drivers e SQL parametrizado · Ficha de trabalho

Aluno(a)

Turma

Nº

Data

Objectivos da ficha

Conectar BD via driver
Usar prepared statements
Gerir transações
Tratar erros

Parte I · Conexão

Exercício 1 · Configuração segura (10 pts)

Como deves armazenar a senha da BD na aplicação? Indica o que não fazer e o que fazer:

Não fazer: - Hardcode no código (password="123"). - Commitar .env ou config.py com credenciais. - Variável global visível em logs.

Fazer: - Variável de ambiente (os.environ["DB_PASSWORD"]). - .env local (no .gitignore) com python-dotenv. - Em produção: secrets manager (Vault, AWS Secrets Manager). - Conta de BD com privilégios mínimos (não usar root/admin para app).

Exercício 2 · Connect padrão (10 pts)

Escreve código Python (usando psycopg2) que: - Lê credenciais de variáveis de ambiente. - Abre conexão. - Executa query simples. - Fecha tudo correctamente.

import os
import psycopg2
from dotenv import load_dotenv

load_dotenv()

DB_CONFIG = {
    "host": os.environ.get("DB_HOST", "localhost"),
    "database": os.environ["DB_NAME"],
    "user": os.environ["DB_USER"],
    "password": os.environ["DB_PASSWORD"],
}

def listar_clientes():
    with psycopg2.connect(**DB_CONFIG) as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT id, nome FROM Cliente ORDER BY nome")
            for row in cur:
                print(row)

listar_clientes()

with fecha conexão e cursor automaticamente, mesmo com excepções.

Parte II · Prepared statements

Exercício 3 · Corrigir SQL injection (15 pts)

Estes 3 fragmentos têm SQL injection. Corrige cada um:

nome = request.form["nome"]
cur.execute(f"SELECT * FROM users WHERE nome = '{nome}'")

ids = "1,2,3"  # vem do user
cur.execute(f"SELECT * FROM Cliente WHERE id IN ({ids})")

nome = request.args["nome"]
cur.execute("INSERT INTO Cliente (nome) VALUES ('%s')" % nome)

a) Placeholder simples:

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

b) Para IN dinâmico, usar tantos placeholders quanto ids:

id_list = [int(x) for x in ids.split(",")]  # valida que são inteiros
placeholders = ",".join(["%s"] * len(id_list))
cur.execute(
    f"SELECT * FROM Cliente WHERE id IN ({placeholders})",
    id_list
)

c) O % aqui é Python string formatting, não placeholder de SQL. Continua vulnerável:

cur.execute("INSERT INTO Cliente (nome) VALUES (%s)", (nome,))

Em todos os casos, vírgula entre query e tuplo é fundamental — não confundir com % de formatting.

Exercício 4 · Múltiplos parâmetros (10 pts)

Escreve INSERT que insere um produto com nome, preço e stock vindo de input do utilizador.

def adicionar_produto(nome, preco, stock):
    with psycopg2.connect(**DB_CONFIG) as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO Produto (nome, preco, stock)
                VALUES (%s, %s, %s)
                RETURNING id
                """,
                (nome, preco, stock)
            )
            new_id = cur.fetchone()[0]
            conn.commit()
            return new_id

# Uso
novo_id = adicionar_produto("Caneta azul", 1.50, 100)
print(f"Produto criado com id {novo_id}")

RETURNING id (PostgreSQL) devolve o id gerado. Em MySQL usa-se cursor.lastrowid.

Parte III · Transações

Exercício 5 · Transferência (20 pts)

Escreve função transferir(origem, destino, qty) que move dinheiro entre 2 contas como transação atómica. Se algo correr mal, faz rollback.

import psycopg2

def transferir(origem, destino, qty):
    if qty <= 0:
        raise ValueError("Quantidade tem de ser positiva")
    if origem == destino:
        raise ValueError("Origem e destino diferentes")

    conn = psycopg2.connect(**DB_CONFIG)
    try:
        with conn.cursor() as cur:
            # Verificar saldo (LOCK para evitar race condition)
            cur.execute(
                "SELECT saldo FROM Conta WHERE id = %s FOR UPDATE",
                (origem,)
            )
            row = cur.fetchone()
            if row is None:
                raise ValueError(f"Conta {origem} não existe")
            if row[0] < qty:
                raise ValueError("Saldo insuficiente")

            cur.execute(
                "UPDATE Conta SET saldo = saldo - %s WHERE id = %s",
                (qty, origem)
            )
            cur.execute(
                "UPDATE Conta SET saldo = saldo + %s WHERE id = %s",
                (qty, destino)
            )
            cur.execute(
                """
                INSERT INTO Transferencia (origem_id, destino_id, qty, data)
                VALUES (%s, %s, %s, NOW())
                """,
                (origem, destino, qty)
            )
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

Notas: - FOR UPDATE bloqueia linha para que outra transação não interfira. - Try/except/rollback garante atomicidade. - Verificação de saldo dentro da transação (não antes).

Exercício 6 · Erros (15 pts)

Que excepções podem ocorrer ao executar query? Como tratar:

a) BD indisponível.

b) Constraint violada (ex.: UNIQUE).

c) Timeout.

import psycopg2
from psycopg2 import errors
from psycopg2.errors import UniqueViolation, OperationalError

def adicionar_cliente(nome, email):
    try:
        with psycopg2.connect(**DB_CONFIG, connect_timeout=5) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    "INSERT INTO Cliente (nome, email) VALUES (%s, %s)",
                    (nome, email)
                )
                conn.commit()
        return True

    except UniqueViolation:
        # Email já existe
        return False

    except OperationalError as e:
        # BD indisponível, timeout, problema de rede
        logger.error(f"BD indisponível: {e}")
        raise

    except Exception as e:
        logger.exception("Erro inesperado")
        raise

Parte IV · Cenário

Exercício 7 · Padrão de acesso (20 pts)

Escreve uma classe ClienteRepository que encapsula acesso à tabela Cliente. Métodos:

get(id) — devolve cliente ou None.
get_by_email(email) — idem.
list_all() — todos os clientes.
create(nome, email) — devolve id.
update(id, **kwargs) — actualiza colunas.
delete(id) — apaga.

Usa prepared statements em tudo. Aceita conn no construtor (dependency injection).

import psycopg2

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

    def get(self, id):
        with self.conn.cursor() as cur:
            cur.execute(
                "SELECT id, nome, email FROM Cliente WHERE id = %s",
                (id,)
            )
            row = cur.fetchone()
            if row is None:
                return None
            return {"id": row[0], "nome": row[1], "email": row[2]}

    def get_by_email(self, email):
        with self.conn.cursor() as cur:
            cur.execute(
                "SELECT id, nome, email FROM Cliente WHERE email = %s",
                (email,)
            )
            row = cur.fetchone()
            if row is None:
                return None
            return {"id": row[0], "nome": row[1], "email": row[2]}

    def list_all(self):
        with self.conn.cursor() as cur:
            cur.execute("SELECT id, nome, email FROM Cliente ORDER BY nome")
            return [
                {"id": r[0], "nome": r[1], "email": r[2]}
                for r in cur
            ]

    def create(self, nome, email):
        with self.conn.cursor() as cur:
            cur.execute(
                "INSERT INTO Cliente (nome, email) VALUES (%s, %s) RETURNING id",
                (nome, email)
            )
            new_id = cur.fetchone()[0]
            self.conn.commit()
            return new_id

    def update(self, id, **kwargs):
        if not kwargs:
            return
        # Construir SET dinamicamente, mas com whitelist de colunas
        allowed = {"nome", "email"}
        sets = []
        params = []
        for col, val in kwargs.items():
            if col not in allowed:
                raise ValueError(f"Coluna não permitida: {col}")
            sets.append(f"{col} = %s")
            params.append(val)
        params.append(id)
        query = f"UPDATE Cliente SET {', '.join(sets)} WHERE id = %s"
        with self.conn.cursor() as cur:
            cur.execute(query, params)
            self.conn.commit()

    def delete(self, id):
        with self.conn.cursor() as cur:
            cur.execute("DELETE FROM Cliente WHERE id = %s", (id,))
            self.conn.commit()

# Uso
conn = psycopg2.connect(**DB_CONFIG)
repo = ClienteRepository(conn)

print(repo.get(1))
new_id = repo.create("Ana", "ana@ex.pt")
repo.update(new_id, nome="Ana Silva")
repo.delete(new_id)

Note a whitelist em update() — sem ela, atacante poderia injectar nomes de colunas perigosos.