Python Conditional JSON Fetching: ETag, 304, and a Local Cache

python3 --version
mkdir etag-demo
cd etag-demo
curl -fS https://www.bobobk.com/downloads/http-etag-cache/fetch_cached.py -o fetch_cached.py
curl -fS https://www.bobobk.com/downloads/http-etag-cache/demo_server.py -o demo_server.py
curl -fS https://www.bobobk.com/downloads/http-etag-cache/check_cache.py -o check_cache.py
python3 fetch_cached.py --help
printf '{"count": 1}\n' > data.json
python3 demo_server.py --port 8765

"""Revalidate one small public JSON endpoint. Python 3.10+, one writer."""
import argparse
from http.client import HTTPException
import json
import math
import os
from pathlib import Path
import sys
import tempfile
from urllib.error import HTTPError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen

LIMIT = 1024 * 1024


def read_limited(stream, limit):
    raw = stream.read(limit + 1)
    if len(raw) > limit:
        raise ValueError("document exceeds size limit")
    return raw


def save_cache(path, record):
    text = json.dumps(record, ensure_ascii=False, allow_nan=False) + "\n"
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(
            mode="w", encoding="utf-8", dir=path.parent, delete=False
        ) as stream:
            temporary = Path(stream.name)
            stream.write(text)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(temporary, path)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


def fetch(url, path, timeout):
    cached = None
    try:
        with path.open("rb") as stream:
            cached = json.loads(read_limited(stream, 8 * LIMIT))
    except FileNotFoundError:
        pass
    if cached is not None:
        if (not isinstance(cached, dict) or cached.get("url") != url
                or "data" not in cached or "etag" not in cached
                or not (cached["etag"] is None or isinstance(cached["etag"], str))):
            raise ValueError("cache schema or URL mismatch; use a separate cache")
    elif path.exists():
        raise ValueError("cache must be an object, not null")

    headers = {"Accept": "application/json", "Accept-Encoding": "identity"}
    if cached is not None and cached["etag"]:
        headers["If-None-Match"] = cached["etag"]
    try:
        with urlopen(Request(url, headers=headers), timeout=timeout) as response:
            if response.status != 200:
                raise ValueError(f"unexpected HTTP status: {response.status}")
            if response.headers.get("Content-Encoding", "identity") != "identity":
                raise ValueError("compressed response not supported")
            data = json.loads(read_limited(response, LIMIT).decode("utf-8"))
            record = {"url": url, "etag": response.headers.get("ETag"), "data": data}
    except HTTPError as exc:
        try:
            if exc.code != 304:
                raise
            if "If-None-Match" not in headers:
                raise ValueError("304 without a cached validator") from exc
            return {"status": "not_modified", "data": cached["data"]}
        finally:
            exc.close()
    save_cache(path, record)
    return {"status": "updated", "data": data}


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("url")
    parser.add_argument("cache", type=Path, help="dedicated JSON cache; parent must exist")
    parser.add_argument("--timeout", type=float, default=5)
    args = parser.parse_args()
    parts = urlsplit(args.url)
    if parts.scheme not in ("http", "https") or not parts.hostname:
        parser.error("URL must use http or https")
    if not math.isfinite(args.timeout) or args.timeout <= 0:
        parser.error("--timeout must be positive and finite")
    try:
        result = fetch(args.url, args.cache, args.timeout)
        print(json.dumps(result, ensure_ascii=False, allow_nan=False))
    except (OSError, ValueError, HTTPException, RecursionError) as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

export no_proxy=127.0.0.1,localhost
export NO_PROXY=127.0.0.1,localhost
python3 fetch_cached.py http://127.0.0.1:8765/data.json cache.json
python3 fetch_cached.py http://127.0.0.1:8765/data.json cache.json

{"status": "updated", "data": {"count": 1}}
{"status": "not_modified", "data": {"count": 1}}

printf '{"count": 2}\n' > data.json
python3 fetch_cached.py http://127.0.0.1:8765/data.json cache.json
python3 -m json.tool cache.json

if python3 fetch_cached.py http://127.0.0.1:8765/data.json cache.json > result.json; then
    python3 -m json.tool result.json
else
    printf 'Fetch failed; do not consume the old cache as a fresh result\n' >&2
fi

python3 check_cache.py

PASS: 200 stores; 304 reuses without rewriting; weak ETag preserved
PASS: changed ETag replaces data and validator together
PASS: 503, invalid JSON, oversized body, timeout preserve old cache
PASS: no ETag means another unconditional GET
PASS: unsolicited 304 fails without creating a cache
PASS: corrupt cache, URL mismatch, invalid timeout rejected
OK: all conditional-fetch checks passed