Python Tutorial / Project: Collect API Data Without Duplicate Storage

Python Tutorial

Validate a whole API response before storing records by a stable external identifier. This project connects JSON retrieval to an SQLite transaction and updates the name and amount for an existing ID on subsequent runs.

Define identity and rerun semantics

Upsert requires a contract that one external id keeps identifying the same transaction. Equal names and amounts alone do not establish duplicate identity. Within one response, this example lets the last occurrence of an ID win; reject repeated IDs before writing if the application requires a different policy.

All rows are validated before any database writes begin. The database writes use one transaction, but that does not turn the network request and local storage into a distributed transaction.

For the relevant APIs and platform requirements, see Project: Collect API Data Without Duplicate Storage documentation.

Prepare the local response

Use a fresh folder. Save expenses.json:

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

Start the local server in the first terminal:

python -m http.server 8000 --bind 127.0.0.1

Keep it running while using the client in a second terminal. Use the same actual port in both commands.

Run the full collector

Save collect.py:

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")
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 amount < 0:
        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 >= 0))")
        con.executemany("INSERT INTO expenses VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET name=excluded.name, amount=excluded.amount", records)
    count, total = con.execute("SELECT COUNT(*), COALESCE(SUM(amount),0) FROM expenses").fetchone()
    print("stored:", count, "total:", total)
finally:
    con.close()

With the project environment selected, run:

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

Expect stored: 2 total: 11500. Running it again should retain two records instead of creating four. On a failure, inspect the traceback and exit status before starting dependent processing.

Check updates, failures and collection completeness

Process the same response twice and compare row counts and totals.
Change e2 to 9000 and verify two rows totaling 12500.
Make an amount a string and verify rejection while the prior database content remains.
Restore the response and stop the server with Ctrl+C.

An external service also needs authentication, pagination, rate-limit handling and rules for changed or deleted records. A few successful pages are not necessarily a complete collection. Do not turn every error into an empty list or add unlimited retries, both of which can hide missing data.

The HTTP client guide explains request failures and limits; the SQLite guide explains the local transaction boundary.

Python Tutorial