Curso de Python / Projeto: coletar dados de API e atualizar o SQLite sem duplicar

Valide toda a resposta de uma API antes de armazenar seus registros por um identificador externo estável. Neste projeto, uma nova execução atualiza o nome e a quantia do mesmo ID em SQLite, sem acrescentar outra cópia da mesma despesa.

Definir identidade e comportamento de repetição

A operação de inserir ou atualizar, chamada de upsert, depende de um contrato: um mesmo id externo precisa continuar identificando a mesma transação. Nome e quantia iguais não provam que duas despesas sejam a mesma. Não combine registros distintos apenas porque seus campos parecem repetidos.

Dentro de uma resposta, este exemplo deixa a última ocorrência válida de um ID prevalecer. Isso difere da política de rejeitar duplicatas usada no projeto CSV. Se sua aplicação exigir rejeição, detecte os IDs repetidos antes da escrita e altere o contrato explicitamente.

Todos os registros são verificados antes de abrir o banco para gravar. As escritas ficam em uma transação local, mas a requisição de rede e o armazenamento não formam uma transação distribuída. As aulas de requisições HTTP e validação de JSON e parâmetros e transações em SQLite explicam as duas partes.

Preparar a resposta e o servidor local

Use Python 3.12 ou posterior, necessário ao parâmetro autocommit do exemplo. Crie uma pasta nova, separada dos exercícios anteriores, e salve o arquivo completo expenses.json em UTF-8:

[{"id":"e1","name":"coffee","amount":3500},{"id":"e2","name":"lunch","amount":8000}]

No primeiro terminal, selecione o ambiente virtual, entre nessa pasta e inicie o servidor de teste:

python -m http.server 8000 --bind 127.0.0.1

Deixe o terminal aberto e use outro para executar o coletor. O servidor escuta apenas no endereço de loopback 127.0.0.1 do próprio computador. Se a porta 8000 estiver ocupada, escolha outra e atualize também a URL do cliente. Sirva somente a pasta de prática, pois outros arquivos desse diretório podem ficar acessíveis pelo servidor local. Esse servidor não representa uma implantação de produção.

Executar o coletor completo

Salve o programa completo abaixo como collect.py, em UTF-8, na pasta de prática.

import json
import sqlite3
import sys
from urllib.request import urlopen

with urlopen(sys.argv[1], timeout=5) as response:
    if response.status != 200:
        raise ValueError("unexpected status")
    raw = response.read(1_000_001)
    if len(raw) > 1_000_000:
        raise ValueError("response too large")
    data = json.loads(raw.decode("utf-8"))
if not isinstance(data, list) or len(data) > 10000:
    raise ValueError("invalid batch")
MAX_AMOUNT = 1_000_000_000_000
records = []
for row in data:
    if not isinstance(row, dict):
        raise ValueError("invalid row")
    key, name, amount = row.get("id"), row.get("name"), row.get("amount")
    if not isinstance(key, str) or not key.strip() or not isinstance(name, str) or not name.strip():
        raise ValueError("invalid identity")
    if type(amount) is not int or not 0 <= amount <= MAX_AMOUNT:
        raise ValueError("invalid amount")
    records.append((key, name, amount))
con = sqlite3.connect("collected.db", autocommit=False)
try:
    with con:
        con.execute("CREATE TABLE IF NOT EXISTS expenses (id TEXT PRIMARY KEY, name TEXT NOT NULL, amount INTEGER NOT NULL CHECK(amount BETWEEN 0 AND 1000000000000))")
        con.executemany("INSERT INTO expenses VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET name=excluded.name, amount=excluded.amount", records)
    count = con.execute("SELECT COUNT(*) FROM expenses").fetchone()[0]
    total = sum(row[0] for row in con.execute("SELECT amount FROM expenses"))
    print("stored:", count, "total:", total)
finally:
    con.close()

No segundo terminal, entre na mesma pasta, selecione o ambiente do projeto e execute:

python collect.py http://127.0.0.1:8000/expenses.json

No macOS ou Ubuntu, fora de um ambiente virtual, use python3 se esse for o comando do interpretador. Os módulos usados pertencem à biblioteca padrão. Na primeira execução, a saída esperada é stored: 2 total: 11500. O arquivo collected.db aparece na pasta atual do terminal.

A resposta precisa ter status 200, no máximo 1.000.000 bytes e uma lista com até 10.000 itens. Cada item precisa ser um dicionário com ID e nome em strings não vazias e uma quantia inteira entre zero e 1.000.000.000.000. Booleanos, strings numéricas e valores fora do intervalo são rejeitados. O uso de strip() verifica se o texto é vazio, mas não normaliza o valor salvo: "e1" e " e1 " continuam sendo identificadores diferentes nesta implementação.

O programa só passa os valores às posições ? depois dessa validação. executemany aplica os registros no mesmo bloco de transação, e o contexto confirma o conjunto ou o reverte se uma exceção sair do bloco. O finally fecha a conexão separadamente. Consulte o controle de transações de sqlite3.

stored e total descrevem todo o banco armazenado, não apenas a resposta atual. Uma lista vazia não apaga registros anteriores; IDs ausentes de uma resposta também não são excluídos. O esquema existente não é migrado por CREATE TABLE IF NOT EXISTS, portanto comece no diretório novo indicado.

Verificar atualizações, falhas e conclusão da coleta

1. Execute duas vezes com a resposta original. Ambas devem mostrar dois registros e total 11500, sem aumentar para quatro registros.
2. Altere somente a quantia de e2 para 9000 no JSON servido. Execute novamente e confirme stored: 2 total: 12500.
3. Troque essa quantia por uma string, como "9000". A validação deve gerar uma exceção antes da escrita. O banco continua com os dois registros e o total 12500 da execução válida anterior.
4. Restaure o número 8000 e execute para recuperar o total 11500. Encerre o servidor com Ctrl+C no primeiro terminal ao terminar.

O coletor deixa as exceções aparecerem como traceback e termina com código diferente de zero. Confira a mensagem e o estado do banco antes de iniciar tarefas dependentes. Uma falha HTTP, de conexão, de decodificação ou de validação não significa uma coleta bem-sucedida sem dados. O timeout de cinco segundos limita esperas de rede, sem garantir um prazo absoluto para todo o processo. Veja urllib.request.urlopen.

O total usa sum do Python sobre os inteiros consultados. Um SUM inteiro do SQLite pode ultrapassar seu limite em bases suficientemente grandes mesmo quando cada valor armazenado cabe no tipo; total() usa ponto flutuante e não é uma substituição adequada quando se exige exatidão inteira. Essa diferença geral não significa que a amostra de dois registros ultrapasse o limite. Consulte as regras de sum e total do SQLite.

Uma API externa também exige autenticação, paginação, limites de requisição e regras para alterações ou exclusões. Algumas páginas processadas não demonstram que toda a coleta terminou. Defina uma política limitada de tentativas e registre falhas; devolver uma lista vazia para qualquer erro ou repetir indefinidamente pode esconder dados ausentes.