Collect API Data Into a Database

Turn a one-off API call into a tracker that remembers. Flatten a nested JSON response into clean records, give them a shape with a dataclass, store them safely in SQLite without ever duplicating a row, and query the history back out.

Advanced 60–75 minutes

About this project

Unit 7 taught you to fetch data. Unit 8 taught you to store it. Nobody shows you the join, and the join is where the useful programs live: a script you can run again tomorrow that pulls fresh data, keeps only what it needs, and does not lose or duplicate what it already had.

Why it is worth building: this is the shape of most real automation. Monitoring, price tracking, dashboards, and reporting pipelines are all the same four moves: fetch, reshape, store, ask. The interesting part is not any one of them; it is making them safe to run twice. By the end you will have a script you could genuinely put on a schedule.

The API call itself is the one piece you cannot run on this page, so a real saved response ships with the project and every other step works from it.

One API response, flattened into records, stored without duplicates, then queried back as history.

Build it step by step

We will work from a saved response so every step runs, then swap in the real network call at the very end. That order is deliberate: it is also the fastest way to build this kind of script for real.

Step 0: The response you are working with

Here is what the API hands back. It is nested, which is normal and is the whole difficulty: the readings live inside a list, inside a key, and the station name lives somewhere else entirely. Save it as sample_response.json.

Example · sample_response.json 
{
  "station": {"id": "OXF", "name": "Oxford"},
  "collected_at": "2026-07-29T09:00:00Z",
  "observations": [
    {"metric": "temperature", "value": 18.2, "unit": "C"},
    {"metric": "humidity",    "value": 71,   "unit": "%"},
    {"metric": "wind",        "value": 12.5, "unit": "km/h"}
  ]
}
One reply, three readings, and the station name a level up from them.

Step 1: Flatten it into records you control

The first real move is refusing to let the API's shape leak into the rest of your program. Write one function that takes the response and returns a flat list of rows, and every later step gets to be simple.

Note .get() with defaults. A field that is present in every response you have seen is not the same as a field that is always present.

Example 
import json

with open("sample_response.json", encoding="utf-8") as f:
    payload = json.load(f)


def to_records(payload):
    """Flatten one API reply into a list of tidy rows."""
    station = payload.get("station", {})
    when = payload.get("collected_at", "")
    rows = []
    for observation in payload.get("observations", []):
        rows.append({
            "station_id": station.get("id", "?"),
            "station_name": station.get("name", "unknown"),
            "collected_at": when,
            "metric": observation.get("metric", "unknown"),
            "value": float(observation.get("value", 0)),
        })
    return rows


for row in to_records(payload):
    print(row)
Output 
{'station_id': 'OXF', 'station_name': 'Oxford', 'collected_at': '2026-07-29T09:00:00Z', 'metric': 'temperature', 'value': 18.2}
{'station_id': 'OXF', 'station_name': 'Oxford', 'collected_at': '2026-07-29T09:00:00Z', 'metric': 'humidity', 'value': 71.0}
{'station_id': 'OXF', 'station_name': 'Oxford', 'collected_at': '2026-07-29T09:00:00Z', 'metric': 'wind', 'value': 12.5}
Three flat rows out of one nested reply.

Step 2: Give the record a shape

A dictionary will happily accept row["statoin_id"] and hand you a KeyError at 3am. A dataclass names the fields once, so a typo is caught where it is written.

frozen=True is a deliberate choice: a reading is a measurement that already happened, and nothing downstream should be editing it.

Example 
from dataclasses import dataclass, astuple


@dataclass(frozen=True)
class Reading:
    station_id: str
    station_name: str
    collected_at: str
    metric: str
    value: float


def to_readings(payload):
    return [Reading(**row) for row in to_records(payload)]


readings = to_readings(payload)
print(readings[0])
print("fields in order:", astuple(readings[1]))
print("count:", len(readings))
Output 
Reading(station_id='OXF', station_name='Oxford', collected_at='2026-07-29T09:00:00Z', metric='temperature', value=18.2)
fields in order: ('OXF', 'Oxford', '2026-07-29T09:00:00Z', 'humidity', 71.0)
count: 3
The same data, now with a name and a fixed set of fields.

Step 3: Create the table, and insert safely

Now the storage. Two rules, both non-negotiable.

Use ? placeholders. Never build SQL with an f-string. A station named O'Brien is enough to break a string-built query, and the same hole is how data gets destroyed on purpose.

Commit. Without connection.commit() the rows vanish when the script ends, and nothing warns you.

Example 
import sqlite3

connection = sqlite3.connect("readings.db")
connection.execute("""
    CREATE TABLE IF NOT EXISTS readings (
        station_id   TEXT NOT NULL,
        station_name TEXT NOT NULL,
        collected_at TEXT NOT NULL,
        metric       TEXT NOT NULL,
        value        REAL NOT NULL
    )
""")

connection.executemany(
    "INSERT INTO readings VALUES (?, ?, ?, ?, ?)",
    [astuple(r) for r in readings],
)
connection.commit()

print("rows stored:", connection.execute("SELECT COUNT(*) FROM readings").fetchone()[0])
Output 
rows stored: 3
Five columns, five question marks.

Step 4: Make running it twice harmless

Run the script again right now and you get six rows: three real ones and three duplicates. A scheduled script runs hundreds of times, so this is the difference between a tracker and a mess.

The fix is to let the database enforce it, not your code. A UNIQUE constraint says which combination of columns identifies a reading; INSERT OR IGNORE then quietly skips anything that would collide.

Example 
fresh = sqlite3.connect(":memory:")
fresh.execute("""
    CREATE TABLE readings (
        station_id   TEXT NOT NULL,
        station_name TEXT NOT NULL,
        collected_at TEXT NOT NULL,
        metric       TEXT NOT NULL,
        value        REAL NOT NULL,
        UNIQUE (station_id, collected_at, metric)
    )
""")


def store(conn, readings):
    """Insert readings, skipping any we already have. Returns rows added."""
    before = conn.total_changes
    conn.executemany(
        "INSERT OR IGNORE INTO readings VALUES (?, ?, ?, ?, ?)",
        [astuple(r) for r in readings],
    )
    conn.commit()
    return conn.total_changes - before


print("first run added: ", store(fresh, readings))
print("second run added:", store(fresh, readings))
print("total rows:      ", fresh.execute("SELECT COUNT(*) FROM readings").fetchone()[0])
Output 
first run added:  3
second run added: 0
total rows:       3
The second run adds nothing. That is the whole point of the step.

Step 5: Ask the history a question

Here is the payoff. A single response could only ever tell you now. A table of them can be asked about trends, averages, and extremes — questions the API never offered.

Example 
rows = [
    Reading("OXF", "Oxford", "2026-07-29T09:00:00Z", "temperature", 18.2),
    Reading("OXF", "Oxford", "2026-07-29T12:00:00Z", "temperature", 22.4),
    Reading("OXF", "Oxford", "2026-07-29T15:00:00Z", "temperature", 21.0),
    Reading("LDS", "Leeds",  "2026-07-29T09:00:00Z", "temperature", 15.9),
    Reading("LDS", "Leeds",  "2026-07-29T12:00:00Z", "temperature", 17.3),
]
store(fresh, rows)

for name, samples, avg, high in fresh.execute("""
    SELECT station_name, COUNT(*), ROUND(AVG(value), 1), MAX(value)
    FROM readings
    WHERE metric = ?
    GROUP BY station_name
    ORDER BY AVG(value) DESC
""", ("temperature",)):
    print(f"{name:8s} {samples} samples  avg {avg:5.1f}  peak {high:5.1f}")
Output 
Oxford   3 samples  avg  20.5  peak  22.4
Leeds    2 samples  avg  16.6  peak  17.3
A question the raw API response could not answer.

Step 6: Fetch it for real, and survive a bad night

The last piece is the network call, and it is the one that fails. A scheduled script meets timeouts, 500s, and rate limits, so it needs two things: a small retry, and a log that will tell you tomorrow what happened while you were asleep.

This block needs the network, so it cannot run on this page, but it is the exact code in the finished tracker.

Example 
import logging
import os
import time

import requests

log = logging.getLogger(__name__)


def fetch(url, attempts=3):
    """GET a JSON payload, retrying briefly on a failure worth retrying."""
    for attempt in range(1, attempts + 1):
        try:
            response = requests.get(
                url,
                params={"station": "OXF"},
                headers={"Authorization": f"Bearer {os.environ['WEATHER_TOKEN']}"},
                timeout=10,
            )
            response.raise_for_status()
            return response.json()
        except requests.RequestException as err:
            log.warning("attempt %d/%d failed: %s", attempt, attempts, err)
            if attempt == attempts:
                raise
            time.sleep(2 ** attempt)


if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(message)s",
        filename="tracker.log",
    )
    added = store(connection, to_readings(fetch("https://api.example.com/v1/current")))
    log.info("stored %d new reading(s)", added)
The real fetch: a token from the environment, a timeout, a retry, and a log.

The finished tracker

Everything assembled, with a small command-line front door so it can be scheduled. --once loads the saved sample so you can try it offline; --report asks the stored history for a summary.

Example · tracker.py 
"""Collect readings from an API into a local SQLite database."""
import argparse
import json
import logging
import os
import sqlite3
import time
from dataclasses import dataclass, astuple

import requests

DB_PATH = "readings.db"
API_URL = "https://api.example.com/v1/current"
log = logging.getLogger("tracker")


@dataclass(frozen=True)
class Reading:
    station_id: str
    station_name: str
    collected_at: str
    metric: str
    value: float


def to_records(payload):
    station = payload.get("station", {})
    when = payload.get("collected_at", "")
    return [{"station_id": station.get("id", "?"),
             "station_name": station.get("name", "unknown"),
             "collected_at": when,
             "metric": o.get("metric", "unknown"),
             "value": float(o.get("value", 0))}
            for o in payload.get("observations", [])]


def connect(path=DB_PATH):
    conn = sqlite3.connect(path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS readings (
            station_id TEXT NOT NULL, station_name TEXT NOT NULL,
            collected_at TEXT NOT NULL, metric TEXT NOT NULL, value REAL NOT NULL,
            UNIQUE (station_id, collected_at, metric))
    """)
    return conn


def store(conn, readings):
    before = conn.total_changes
    conn.executemany("INSERT OR IGNORE INTO readings VALUES (?, ?, ?, ?, ?)",
                     [astuple(r) for r in readings])
    conn.commit()
    return conn.total_changes - before


def fetch(url, attempts=3):
    for attempt in range(1, attempts + 1):
        try:
            response = requests.get(
                url, timeout=10,
                headers={"Authorization": f"Bearer {os.environ['WEATHER_TOKEN']}"})
            response.raise_for_status()
            return response.json()
        except requests.RequestException as err:
            log.warning("attempt %d/%d failed: %s", attempt, attempts, err)
            if attempt == attempts:
                raise
            time.sleep(2 ** attempt)


def report(conn):
    for name, n, avg, high in conn.execute("""
        SELECT station_name, COUNT(*), ROUND(AVG(value), 1), MAX(value)
        FROM readings WHERE metric = ? GROUP BY station_name ORDER BY AVG(value) DESC
    """, ("temperature",)):
        print(f"{name:10s} {n:3d} samples  avg {avg:5.1f}  peak {high:5.1f}")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--once", action="store_true",
                        help="load sample_response.json instead of calling the API")
    parser.add_argument("--report", action="store_true", help="summarize what is stored")
    args = parser.parse_args()

    logging.basicConfig(level=logging.INFO,
                        format="%(asctime)s %(levelname)s %(message)s")
    conn = connect()

    if args.report:
        report(conn)
        return

    if args.once:
        with open("sample_response.json", encoding="utf-8") as f:
            payload = json.load(f)
    else:
        payload = fetch(API_URL)

    added = store(conn, [Reading(**row) for row in to_records(payload)])
    log.info("stored %d new reading(s) of %d received", added, len(to_records(payload)))


if __name__ == "__main__":
    main()
Output 
$ python tracker.py --once
2026-07-29 09:00:04 INFO stored 3 new reading(s) of 3 received

$ python tracker.py --once          # the same response again
2026-07-29 09:01:11 INFO stored 0 new reading(s) of 3 received

$ python tracker.py --report
Oxford       3 samples  avg  18.2  peak  18.2
Run it twice and the second run stores nothing. Run it daily and you have history.

Keep going, make it your own

You have a tracker. These turn it into something you would actually schedule.

Track more than one station

Loop over a list of station ids and call fetch for each. The table already has station_id, and the UNIQUE constraint already includes it, so nothing else has to change, which is a good sign the schema was right.

Example 
for station in ["OXF", "LDS", "MAN"]:
    payload = fetch(API_URL, station=station)
    added = store(conn, [Reading(**row) for row in to_records(payload)])
    log.info("%s: %d new", station, added)

Fetch the stations concurrently

Ten stations one at a time is ten round trips of waiting. A thread pool fetches them at once, but keep the pool small, because twenty simultaneous requests is a fine way to earn a 429.

Example 
from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=4) as pool:
    payloads = list(pool.map(lambda s: fetch(API_URL, station=s), STATIONS))

Add an index once the table gets big

Every query in the report filters on metric and groups by station. Once there are a hundred thousand rows, an index on the columns you filter by turns a scan into a lookup.

Example 
conn.execute("CREATE INDEX IF NOT EXISTS idx_metric_station ON readings (metric, station_id)")

Export the history for analysis

A SELECT straight into pandas is one line, and from there you are in report territory: charts, rolling averages, and a one-page summary of a month of collection.

Example 
import pandas as pd

frame = pd.read_sql_query("SELECT * FROM readings", conn)
print(frame.groupby("station_name")["value"].describe())

Download the files

The finished tracker plus a saved API response, so you can build the whole thing offline. Run python tracker.py --once to load the sample, then python tracker.py --report.

Mini exercise (medium)

Write the flattening function on your own, against a response shaped a little differently from the one above, because APIs never agree with each other. to_rows(payload) should return one row per observation, carrying the sensor id and label down onto each, and it must not fall over when a field is missing.

Have a go. Finish the code and press Run to see the result immediately, right on this page.

PAYLOAD = {
    "sensor": {"ref": "S-9", "label": "Loading bay"},
    "taken": "2026-07-29T06:00:00Z",
    "points": [
        {"kind": "temp", "reading": 4.5, "unit": "C"},
        {"kind": "door", "reading": 1},
    ],
}

def to_rows(payload):
    """One flat row per point, carrying the sensor details down."""
    return []   # TODO

for row in to_rows(PAYLOAD):
    print(row)

print("missing sensor is survivable:", to_rows({"points": [{"kind": "x"}]}))

Where to go next

You have joined the two halves of the Advanced track: fetching from Unit 7 and storing from Unit 8, with the durability habits of Unit 9 holding it together.

The natural next build is to make the collected data say something: From Messy Export to One-Page Report takes a table and turns it into a summary and a chart. If instead you want this tool installable so it can be scheduled properly, go to Package and Ship a Command-Line Tool.