Curso de Python / Proyecto de recopilación de una API con almacenamiento SQLite

Curso de Python

Para recopilar una API, valida toda la respuesta antes de guardar registros identificados de forma estable. Este proyecto conecta una petición JSON con una transacción SQLite y actualiza el nombre e importe de un registro existente cuando vuelve a recibir su identificador.

Definir identidad y comportamiento al repetir

El identificador externo debe seguir representando la misma transacción. Tener el mismo nombre e importe no basta para considerar dos gastos duplicados. Dentro de una misma respuesta, este programa permite que gane la última aparición de un identificador. Si tu aplicación exige rechazar duplicados, compruébalos antes de escribir.

La validación de todo el lote precede a cualquier escritura. Las escrituras comparten una transacción local, pero la descarga y la base no forman una transacción distribuida. Si el servicio cambia mientras recopilas varias páginas, necesitas una política adicional de consistencia y de colección completa.

Preparar el servicio local

Utiliza una carpeta nueva y Python 3.12 o posterior. Guarda expenses.json con este contenido:

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

En la primera terminal, desde esa carpeta, inicia el servidor local:

python -m http.server 8000 --bind 127.0.0.1

Mantenlo abierto. Si cambias el puerto, utiliza el mismo en la URL del cliente. El cliente HTTP introductorio explica la dirección local y las distintas clases de errores de comunicación.

Ejecutar el recopilador completo

Guarda el programa como collect.py y utiliza una segunda terminal para ejecutarlo. Solo requiere la biblioteca estándar.

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()

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

La primera ejecución debe mostrar stored: 2 total: 11500. Repetir la misma respuesta mantiene dos filas, en lugar de crear cuatro. Los valores se vinculan como parámetros SQL y la cláusula de conflicto actualiza la fila con el mismo identificador.

Cada importe debe ser un entero entre cero y 1.000.000.000.000. El límite del lote es de 10000 registros y el de la respuesta, un millón de bytes. El total se acumula con enteros de Python: una suma SQL de enteros puede desbordarse incluso si cada fila cabe individualmente. No sustituyas importes exactos por una suma de coma flotante para ocultar ese problema.

Las restricciones de una tabla existente no cambian con CREATE TABLE IF NOT EXISTS. Para el ejercicio utiliza una base nueva; los datos reales requieren una migración respaldada. Las lecciones de SQLite y transacciones y la referencia de sum y total de SQLite explican estas condiciones.

Verificar actualizaciones y conservación ante fallos

Procesa la misma respuesta dos veces y compara número de filas y total.
Cambia el importe de e2 a 9000: deben mantenerse dos filas y aparecer un total de 12500.
Cambia ese importe a una cadena: la ejecución debe fallar antes de escribir y conservar los valores anteriores.
Restaura la respuesta y detén el servidor con Ctrl+C.

Inspecciona el traceback y el estado de salida antes de iniciar procesos dependientes. Un servicio real necesita autenticación, paginación, límites de uso y reglas para cambios o eliminaciones. No conviertas cada error en una lista vacía ni añadas reintentos ilimitados, porque ocultarían información ausente. La referencia de urlopen detalla el alcance de la petición y sus esperas.

Curso de Python