Fetching data is only step one. A useful program fetches, transforms into a shape it controls, and stores the result — combining this module with Module 6.

The pipeline

import json
import requests

def fetch_users(url):
    """Fetch raw user data from the API."""
    response = requests.get(url, timeout=10)
    response.raise_for_status()
    return response.json()

def transform(raw):
    """Keep only the fields we care about."""
    return [
        {
            "name": u["name"],
            "email": u.get("email", "unknown"),
            "city": u.get("address", {}).get("city", "unknown"),
        }
        for u in raw
    ]

def save(records, path):
    """Write records to a JSON file."""
    with open(path, "w") as f:
        json.dump(records, f, indent=2)

def run(url, path):
    raw = fetch_users(url)
    records = transform(raw)
    save(records, path)
    return len(records)

Each function does one job, so each can be tested on its own — the Module 7 structure lesson, applied. transform in particular needs no network at all.

Caching

Re-fetching unchanged data is slow and rude to the API. Cache it: check for a fresh local copy first, and only hit the network when you have to.

import json
import os
import time

CACHE_FILE = "cache.json"
MAX_AGE = 3600

def is_fresh(path, max_age):
    """True if the cache exists and is younger than max_age seconds."""
    if not os.path.exists(path):
        return False
    return (time.time() - os.path.getmtime(path)) < max_age

def load_cache(path):
    with open(path) as f:
        return json.load(f)

with open(CACHE_FILE, "w") as f:
    json.dump({"cached": True}, f)

print(is_fresh(CACHE_FILE, MAX_AGE))
print(is_fresh("no-such-file.json", MAX_AGE))
print(load_cache(CACHE_FILE))

os.path.getmtime gives the last-modified time; comparing it to time.time() gives the age.

The real function then reads:

def get_data(url, path=CACHE_FILE, max_age=MAX_AGE):
    if is_fresh(path, max_age):
        print("Using cache")
        return load_cache(path)

    print("Fetching from API")
    data = requests.get(url, timeout=10).json()
    with open(path, "w") as f:
        json.dump(data, f)
    return data

Write safely

A crash halfway through json.dump leaves a truncated, unreadable file — and you've destroyed the good copy that was there before. Write to a temp file, then rename:

import json
import os

def save_atomic(data, path):
    """Write to a temp file, then replace — never leaves a half-written file."""
    tmp = path + ".tmp"
    with open(tmp, "w") as f:
        json.dump(data, f, indent=2)
    os.replace(tmp, path)

save_atomic({"safe": True}, "out.json")

with open("out.json") as f:
    print(json.load(f))

os.replace() is atomic on every major platform: the file is either the old version or the new one, never a mixture.

Appending over time

For data you collect repeatedly — a daily reading, a log — load, append, save:

import json
import os

def append_record(record, path="history.json"):
    history = []
    if os.path.exists(path):
        with open(path) as f:
            history = json.load(f)
    history.append(record)
    with open(path, "w") as f:
        json.dump(history, f, indent=2)
    return len(history)

print(append_record({"temp": 21}))
print(append_record({"temp": 23}))
print(append_record({"temp": 19}))

For large volumes you'd use a database, but JSON is perfectly good for thousands of records.

Try it yourself

The whole pipeline, with a simulated API.

Checkpoint

• Split the pipeline into fetch, transform, and save — each testable alone.
• Transform into your own shape so API changes hit one function.
• Cache with a freshness check to avoid needless requests.
• Write to a temp file and os.replace() so a crash can't corrupt good data.
• Load-append-save builds history files over time.