Collecting Data from APIs, Files, and Object Storage

In plain terms

Data from outside the company arrives however the other company decided. A file dropped on a server every night. A web interface that hands over a page of records at a time and refuses to answer if you ask too often. A shared storage bucket that another team writes into. Each of these has its own way of being incomplete, late, or delivered twice, and a successful transfer alone does not reveal which case occurred.

External APIs can offer stable IDs, change feeds, or snapshot exports, but those guarantees must be verified rather than inferred from the transport. A cursor, manifest, or checksum answers only part of the problem: record identity, delivery integrity, source completeness, and replay are separate contracts. This article develops those contracts and tests selected failure cases with a local API and file-drop simulator.

Choose the source before the transport

Internal sources include sales, CRM, finance, HR, marketing operations, and archived records. External sources include suppliers, government portals, universities, associations, and commercial providers. Internal data can describe your own operations closely; external data can add market context or a comparison group. Neither origin guarantees relevance, quality, free access, or low preparation cost. Name the owner and obtain the required access before designing extraction.

Internal versus external describes an organizational boundary. Primary versus secondary describes how data relates to the current study: collected directly for its question or reused from an earlier purpose. An internal transaction archive can therefore be secondary data for a new analysis. Do not use the two classifications interchangeably.

A bike-share operator might combine its trip counts with public transport schedules to investigate unmet demand. Align location, time period, units, and coverage before joining them; a combined table alone does not establish the best station location. Open data permits reuse under applicable open terms and must also be accessible in a usable form. A free download or a government publisher alone does not establish openness. Record the licence, source version, definitions, and refresh schedule. The Open Definition explains those requirements.

What the external source does not give you

These are starting points, not equivalent replacements for database guarantees. A file checksum identifies bytes, not individual business records, and an overlap cannot recover hard deletes unless the provider exposes them. Confirm each source contract before choosing a repair strategy.

APIs

Pagination that does not drift

Offset pagination addresses row positions. Inserts or deletes ahead of a page boundary can repeat or skip records in a changing list. Keyset pagination addresses a stable ordering key, such as “IDs below the last returned ID,” and avoids that positional shift. A cursor is merely an API token: it may encode an offset, a keyset boundary, or a snapshot. It does not by itself freeze changing values, prevent deletes, or include later inserts. The lab compares live offset reads with a keyset crawl bounded by a starting maximum ID. All original IDs are returned only because the simulator inserts larger IDs and does not update or delete existing rows.

Follow the provider’s documented pagination semantics, preserve opaque tokens unchanged, and check expiration, repeated tokens, empty-page behavior, and upper-bound support. If only offsets exist, an immutable sort and deduplication reduce some errors but cannot restore rows skipped by deletion. Ask for a snapshot export or reconcile a bounded period when completeness matters. The toy API exposes a numeric keyset cursor; its starting ceiling is obtained from simulator state, not from a production snapshot endpoint.

Rate limits and the retry policy

Retry only operations and failures the contract allows, with bounded attempts, timeouts, and jitter. A 429 or 503 may carry Retry-After as integer seconds or an HTTP date; wait at least the valid requested interval, or reschedule if it exceeds the run’s budget. Do not truncate it and retry early. Some 4xx responses, such as 408, can be retryable; a 401 may permit one documented credential refresh, while a revoked token or forbidden request needs intervention. The lab has no refresh flow and stops on 401/403. Log sanitized status and request IDs rather than arbitrary response bodies or secret-bearing URLs. See HTTP Retry-After semantics.

Incremental parameters and the request ledger

If an API offers since or updated_after, check which changes that boundary captures. The partner decides what “updated” means, and it may omit changes you care about. The same defences apply, an overlap window and a merge on the record id, plus one question to the partner’s documentation: what does updated_at change on, and does a deleted record ever appear?

Record the endpoint, source period, pagination boundary, request identifiers, counts, and selected quota headers. Store resumable cursor tokens securely if they carry access rights. This ledger supports diagnosis, not exact reproduction of a mutable API response. Reproducibility requires retained permitted response bytes or a stable source snapshot/version. Commit progress only after durable output; an expired cursor may require a bounded restart and deduplication. Keep credentials out of URLs and logs, and use protected storage and the provider’s rotation process.

Files: SFTP drops and shared folders

Visible is not complete

An SFTP upload may be visible while incomplete, and a truncated CSV can still parse. Require a producer completion protocol: upload under a temporary name and use a supported same-filesystem atomic rename, or publish a manifest that references an immutable completed file. A .part suffix alone is insufficient unless removing it is part of that protocol. Bind the signal to an exact delivery revision; an old manifest beside a newly overwritten file is not proof of completion. A missing signal means the delivery is not eligible for loading yet.

A manifest can declare source period, revision, file name, byte size, row count, checksum, and a control total with explicit units. A checksum compares the received bytes with the declared bytes; it does not prove that the producer exported all intended rows. Counts or totals computed from the same incomplete export can agree with a bad file. Independent expected counts and reconciliation strengthen the check. A trusted delivery channel or authenticated manifest is needed when origin matters; an untrusted hash alone provides no authenticity.

The file format is part of the contract

The encoding article covered what can go wrong inside a text file; here it goes wrong at the boundary between two companies. A partner’s export tool starts writing a UTF-8 byte-order mark and an unhandled decode leaves a U+FEFF character in the first column name. Someone changes the delimiter to a semicolon for a European spreadsheet. A column is renamed, or two columns are swapped. These changes can produce parseable files with unexpected columns, or cause decoding and parsing failures. The collector therefore validates the header against the contract exactly, decodes with the encoding the contract names (utf-8-sig, which accepts and strips a byte-order mark, is the tolerant choice for UTF-8), and quarantines any file whose shape differs, with the actual header in the reason. Silent tolerance (“just detect the delimiter”) converts a contract violation into a column that is sometimes wrong.

Redelivery and restatement

Identify the dataset and business period separately from delivery revision and byte checksum. A changed checksum signals different bytes, not necessarily a newer correction; a delayed old file can arrive after the restatement. Require an ordered revision or an agreed acceptance policy, reject conflicting content for one revision, and validate before deduplication or publication. Replace a period only from a complete replacement snapshot, including an explicitly empty one. Publish data, revision progress, and audit metadata atomically in durable storage. Retain permitted raw versions and rejected inputs under the source’s access and retention policy.

Object storage

Object guarantees are provider- and operation-specific. For Amazon S3, writes to one key are atomic, but a multi-object dataset has no automatic all-or-nothing publication. Use a versioned manifest for the complete set. Creation notifications can be duplicated or reordered. An ETag is not universally a content hash: multipart uploads and encryption affect its meaning. Prefer immutable bucket/key/version identity and a separately verified checksum. See S3 consistency and object integrity checks.

Fetch the version named by the notification rather than whatever currently occupies the key. For a latest-state replica, reject older versions only after the newer version has successfully published; a history archive may need every version. S3 sequencers compare events for the same key only, and different-length hexadecimal values need zero-padding or numeric comparison. AWS documents sequencer comparison. A ledger alone does not coordinate concurrent workers: use durable claims and atomic or conditional updates, and acknowledge the message after completion. Listings remain useful for inventory and missed-notification reconciliation; folder location alone is insufficient workflow state.

Transfer success is not completeness

Every mechanism so far answers “is this thing I received correct?” None of them answers “did I receive everything?” A partner whose nightly job crashed sends no file, and a collector that only reacts to files sees nothing wrong. Completeness is a separate check, driven by a calendar rather than by arrivals: the contract says a file per business day by 06:00, so at 06:00 the collector asks, for each expected day, whether an accepted file exists, and reports the days that are missing, still uploading, or quarantined, with how overdue each is. The same check runs for an API pulled hourly (was each hour’s window fetched?) and for a bucket (did each expected key arrive?). It is the difference between an incident found at 06:10 by the platform and one found on Monday by finance. The lab’s sixth exercise is that report.

Two habits follow. Backfills and re-fetches are addressed by source period, the business date or the API window, never by the time the job happened to run, so that “re-fetch 4 March” is a well-defined request. And the reconciliation counts from the database article apply across the boundary: pages against the API’s own count if it offers one, rows against the manifest, control totals against a finance figure the partner reports separately.

Failure classes at the boundary

The partner ingestion contract

Everything above depends on knowing what the partner promised, and the promise is rarely written down until the platform writes it. The contract is one page per source, agreed with the partner’s technical contact, holding: the transport and endpoint or path; the schedule and deadline, with the time zone stated; the format, encoding, header, and delimiter, or the JSON shape; the completion signal; the identity of a record and of a delivery; whether and how restatements are sent; how much notice precedes a schema change; the retention on their side, which bounds how far back a re-fetch can go; the rate limit; the credential rotation schedule; and a named person on each side. The lab’s last exercise writes it for the file source. These delivery rules are part of a data contract and also apply between internal teams.

The subscription company’s external sources

A created-time filter retrieves new records, not necessarily later changes to older charges. Use the provider’s supported update feed, webhooks plus reconciliation, or targeted re-fetches for those changes. Reconcile reports only after aligning periods, currencies, and amount definitions. The table’s overlap periods are proposed contract settings, not guarantees established by this lab.

Across these transports, agree identity with the producer, verify the completion signal and the delivered bytes before publication, record accepted revisions, and check the expected calendar even when nothing arrives.

Anti-patterns

• Offset pagination over a live list. A crawl that returns duplicates at every page boundary and misses whatever was created while it ran.
• Retrying a 401. A revoked token retried in a loop until the partner blocks the account.
• Reading whatever is in the folder. Half a file loaded at 02:00, the other half never.
• Detecting the format. A delimiter sniffer that turns a contract change into a column that is sometimes wrong.
• Appending the redelivery. 3 March counted twice because the partner’s retry looked like new data.
• The prefix listing as a workflow. Two runs overlap and one of them processes the other’s files.
• Silence as success. No file, no error, no report; the gap is found at month end.
• The token in the URL. Copied into every log line and every ledger row.

Lab

The lab uses a loopback HTTP server and local files, not a real partner or cloud service. The API supports numeric keyset and offset pagination and deterministic 429/500 failures. Its valid Retry-After value is zero so the demo runs quickly. The shared GET helper adds bounded jitter and selected transport retries; no credential refresh, checkpoint store, or end-to-end job deadline is implemented. File and notification state is in memory with one caller, so process-crash durability and concurrent workers are outside the lab. Run setup once; exercises may be run in order, and the completeness report inspects files rather than the publication ledger.

import io, re, socket, math
from email.utils import parsedate_to_datetime
from urllib.parse import parse_qs, urlencode
import csv, hashlib, json, os, random, tempfile, threading, time, urllib.error, urllib.request
from datetime import date, datetime, timedelta, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

rng = random.Random(0)
ORDERS = [{"id": i, "amount_cents": rng.randint(500, 9900), "updated_at": f"2026-03-01T{i % 24:02d}:00:00Z"} for i in range(1, 501)]
COUNTERS = {"requests": 0, "inserted": 0}

class PartnerAPI(BaseHTTPRequestHandler):
    """A partner's orders API: newest first, 50 per page, a 429 every fourth request, a 500 now and then."""
    protocol_version = "HTTP/1.1"

    def log_message(self, *args):
        pass

    def reply(self, status, body, headers=()):
        self.send_response(status)
        self.send_header("Content-Length", str(len(body)))
        for name, value in headers:
            self.send_header(name, value)
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self):
        COUNTERS["requests"] += 1
        path, _, query = self.path.partition("?")
        params = {k: v[0] for k, v in parse_qs(query).items()}
        if self.headers.get("Authorization") != "Bearer partner-token":
            return self.reply(401, b'{"error": "bad token"}')
        if COUNTERS["requests"] % 4 == 0:
            return self.reply(429, b'{"error": "slow down"}', [("Retry-After", "0")])
        if COUNTERS["requests"] % 15 == 7:
            return self.reply(500, b'{"error": "upstream timeout"}')
        newest_first = sorted(ORDERS, key=lambda o: -o["id"])
        limit = int(params.get("limit", 50))
        if "offset" in params:
            offset = int(params["offset"])
            page = newest_first[offset:offset + limit]
            if offset and params.get("live") == "1":
                ORDERS.append({"id": max(o["id"] for o in ORDERS) + 1, "amount_cents": 1000, "updated_at": "2026-03-02T00:00:00Z"})
                COUNTERS["inserted"] += 1
        else:
            after = int(params.get("after", 10 ** 9))
            ceiling = int(params.get("ceiling", 10 ** 9))
            page = [o for o in newest_first if o["id"] < after and o["id"] <= ceiling][:limit]
            if page and params.get("live") == "1":
                ORDERS.append({"id": max(o["id"] for o in ORDERS) + 1, "amount_cents": 1000, "updated_at": "2026-03-02T00:00:00Z"})
                COUNTERS["inserted"] += 1
        self.reply(200, json.dumps({"data": page, "next_after": page[-1]["id"] if page else None}).encode())

server = ThreadingHTTPServer(("127.0.0.1", 0), PartnerAPI)
threading.Thread(target=server.serve_forever, daemon=True).start()
BASE = f"http://127.0.0.1:{server.server_address[1]}"

def retry_after(value, now):
    if value is None:
        return None
    if re.fullmatch(r"[0-9]+", value.strip()):
        return int(value)
    try:
        when = parsedate_to_datetime(value)
        if when.utcoffset() is None:
            return None
        return max(0.0, (when - now).total_seconds())
    except (ValueError, TypeError, OverflowError):
        return None

RETRYABLE = {408, 429, 500, 502, 503, 504}
retry_rng = random.Random(7)

def checked_page(body):
    """Validate this simulator's numeric, descending-ID page contract."""
    if not isinstance(body, dict) or not isinstance(body.get("data"), list) or "next_after" not in body:
        raise ValueError("invalid page envelope")
    rows = body["data"]
    for row in rows:
        if not isinstance(row, dict) or type(row.get("id")) is not int or row["id"] < 1:
            raise ValueError("invalid record id")
        if type(row.get("amount_cents")) is not int or row["amount_cents"] < 0:
            raise ValueError("invalid amount_cents")
        if not isinstance(row.get("updated_at"), str):
            raise ValueError("invalid updated_at")
    ids = [r["id"] for r in rows]
    if any(a <= b for a, b in zip(ids, ids[1:])):
        raise ValueError("page IDs must be unique and descending")
    cursor = body["next_after"]
    if rows:
        if type(cursor) is not int or cursor != ids[-1]:
            raise ValueError("cursor differs from last record")
    elif cursor is not None:
        raise ValueError("empty page must end this toy traversal")
    return body

def get(url, tries=6, outcomes=None, opener=None, sleeper=time.sleep, max_wait=30):
    """Bounded retries for this read-only API; no credential refresh is implemented."""
    if type(tries) is not int or tries < 1:
        raise ValueError("tries must be positive")
    if type(max_wait) not in (int, float) or not math.isfinite(max_wait) or max_wait < 0:
        raise ValueError("max_wait must be finite and non-negative")
    opener = opener or urllib.request.urlopen
    for attempt in range(tries):
        request = urllib.request.Request(url, headers={"Authorization": "Bearer partner-token"})
        advised = None
        try:
            with opener(request, timeout=5) as response:
                if outcomes is not None:
                    outcomes.append(response.status)
                return checked_page(json.loads(response.read()))
        except urllib.error.HTTPError as error:
            if outcomes is not None:
                outcomes.append(error.code)
            if error.code not in RETRYABLE:
                error.close()
                raise
            advised = retry_after(error.headers.get("Retry-After"), datetime.now(timezone.utc))
            error.close()
        except (TimeoutError, ConnectionResetError):
            if outcomes is not None:
                outcomes.append("transport")
        except urllib.error.URLError as error:
            if not isinstance(error.reason, (TimeoutError, ConnectionResetError)):
                raise
            if outcomes is not None:
                outcomes.append("transport")
        if attempt == tries - 1:
            break
        delay = max(advised or 0, retry_rng.uniform(0, min(2.0, 0.01 * 2 ** attempt)))
        if delay > max_wait:
            raise RuntimeError("server retry delay exceeds this run's wait budget; reschedule")
        sleeper(delay)
    raise RuntimeError(f"gave up after {tries} attempts")

print(get(BASE + "/orders?limit=2"))
print(COUNTERS)
# {'data': [{'id': 500, 'amount_cents': 5313, 'updated_at': '2026-03-01T20:00:00Z'}, {'id': 499, 'amount_cents': 6851, 'updated_at': '2026-03-01T19:00:00Z'}], 'next_after': 499}
# {'requests': 1, 'inserted': 0}

The second partner uploads a daily sales CSV and a manifest to a drop folder, and the simulator writes them the way partners actually do: one day clean, one with a byte-order mark, one with semicolons, one truncated by their job but described by a manifest for the full file, and one still uploading under a .part name. Day four never arrives.

drop_dir = tempfile.TemporaryDirectory()
DROP = drop_dir.name
raw_archive = {}
first_day = date(2026, 3, 1)
HEADER = "order_id,amount_cents,country"

def deliver(day, rows, finished=True, encoding="utf-8", delimiter=",", truncate=None, revision=1):
    """The partner's nightly upload: a CSV and a manifest, written the way partners actually write them."""
    name = f"sales_{day.isoformat()}.csv"
    lines = [HEADER.replace(",", delimiter)] + [delimiter.join(map(str, r)) for r in rows[:truncate]]
    payload = ("\n".join(lines) + "\n").encode(encoding)
    with open(os.path.join(DROP, name + ("" if finished else ".part")), "wb") as f:
        f.write(payload if finished else payload[: len(payload) // 2])
    if finished:
        manifest = {"file": name, "business_date": day.isoformat(), "revision": revision,
                    "bytes": len(payload), "rows": len(rows), "sha256": hashlib.sha256(payload).hexdigest(),
                    "total_amount_cents": sum(r[1] for r in rows)}
        with open(os.path.join(DROP, name + ".manifest.json"), "w") as f:
            json.dump(manifest, f)

def partner_rows(day, count):
    r = random.Random(day.toordinal())
    return [(f"{day:%Y%m%d}-{n:04d}", r.randint(500, 9900), r.choice(["US", "KR", "DE"])) for n in range(1, count + 1)]

deliver(first_day, partner_rows(first_day, 300))
deliver(first_day + timedelta(days=1), partner_rows(first_day + timedelta(days=1), 320), encoding="utf-8-sig")
deliver(first_day + timedelta(days=2), partner_rows(first_day + timedelta(days=2), 310), delimiter=";")
deliver(first_day + timedelta(days=4), partner_rows(first_day + timedelta(days=4), 290), truncate=200)
deliver(first_day + timedelta(days=5), partner_rows(first_day + timedelta(days=5), 305), finished=False)
print(sorted(os.listdir(DROP)))
# ['sales_2026-03-01.csv', 'sales_2026-03-01.csv.manifest.json', 'sales_2026-03-02.csv', 'sales_2026-03-02.csv.manifest.json', 'sales_2026-03-03.csv', 'sales_2026-03-03.csv.manifest.json', 'sales_2026-03-05.csv', 'sales_2026-03-05.csv.manifest.json', 'sales_2026-03-06.csv.part']

The validator below is the file contract as code: a manifest must exist, the checksum must match, the header must be exactly the agreed one after decoding, and the row count and control total must equal the manifest’s. Exercises 3 and 6 use it.

def inspect_file(name):
    """Read one candidate once; never publish a separately re-read payload."""
    if not re.fullmatch(r"sales_[0-9]{4}-[0-9]{2}-[0-9]{2}[.]csv", name):
        return "quarantine: invalid file name", None
    path = os.path.join(DROP, name)
    try:
        with open(path + ".manifest.json", "rb") as f:
            manifest_bytes = f.read()
    except FileNotFoundError:
        return "waiting: no manifest yet", None
    try:
        with open(path, "rb") as f:
            payload = f.read()
    except FileNotFoundError:
        return "waiting: manifest exists but data is missing", None
    digest = hashlib.sha256(payload).hexdigest()
    raw_archive[(name, digest, hashlib.sha256(manifest_bytes).hexdigest())] = (payload, manifest_bytes)
    try:
        manifest = json.loads(manifest_bytes)
        if not isinstance(manifest, dict):
            raise ValueError("manifest is not an object")
        day = date.fromisoformat(name[6:-4])
        if manifest.get("file") != name or manifest.get("business_date") != day.isoformat():
            raise ValueError("manifest file or business date differs")
        if any(type(manifest.get(k)) is not int or manifest[k] < 0 for k in ("rows", "bytes", "total_amount_cents")):
            raise ValueError("invalid manifest counts")
        if type(manifest.get("revision")) is not int or manifest["revision"] < 1:
            raise ValueError("invalid revision")
        if digest != manifest.get("sha256"):
            raise ValueError("checksum does not match manifest")
        if len(payload) != manifest["bytes"]:
            raise ValueError("byte size differs")
        text = payload.decode("utf-8-sig")
        reader = csv.reader(io.StringIO(text, newline=""), strict=True)
        header = next(reader, None)
        if header != HEADER.split(","):
            first_line = text.splitlines()[0] if text else ""
            raise ValueError(f"header is {first_line!r}")
        rows, seen = [], set()
        for values in reader:
            if len(values) != 3:
                raise ValueError("row width differs")
            order_id, amount, country = values
            if not re.fullmatch(day.strftime("%Y%m%d") + r"-[0-9]{4,}", order_id) or order_id in seen:
                raise ValueError("invalid or duplicate order_id")
            if not re.fullmatch(r"[0-9]+", amount) or country not in {"US", "KR", "DE"}:
                raise ValueError("invalid amount or country")
            seen.add(order_id)
            rows.append(dict(order_id=order_id, amount_cents=int(amount), country=country))
        if len(rows) != manifest["rows"]:
            raise ValueError(f"{len(rows)} rows but manifest promises {manifest['rows']}")
        if sum(r["amount_cents"] for r in rows) != manifest["total_amount_cents"]:
            raise ValueError("control total differs")
        return f"accepted: {len(rows)} rows", (rows, manifest, payload)
    except (ValueError, UnicodeError, csv.Error) as error:
        return f"quarantine: {error}", None

def validate(name):
    return inspect_file(name)[0]

1. Crawl a source that moves. Fetch every order by offset pages from the quiet source, then again in live mode while the source inserts a record after each page, then by cursor. Compare rows fetched, distinct ids, and the source’s size each time.

def crawl_by_offset(live):
    fetched, offset = [], 0
    for _ in range(1000):
        page = get(f"{BASE}/orders?offset={offset}&limit=50&live={int(live)}")["data"]
        if not page:
            return fetched
        fetched += page
        offset += 50
    raise RuntimeError("offset page limit exceeded")

def crawl_by_cursor(ceiling):
    fetched, after, seen = [], None, set()
    for _ in range(1000):
        response = get(f"{BASE}/orders?limit=50&live=1&ceiling={ceiling}" + (f"&after={after}" if after else ""))
        if any(o["id"] > ceiling or (after is not None and o["id"] >= after) for o in response["data"]):
            raise ValueError("page crossed the requested keyset boundary")
        fetched += response["data"]
        after = response["next_after"]
        if after is None:
            return fetched
        if after in seen:
            raise RuntimeError("cursor did not advance")
        seen.add(after)
    raise RuntimeError("cursor page limit exceeded")

quiet = crawl_by_offset(live=False)
print("quiet source, offset pages:", len(quiet), "rows,", len({o["id"] for o in quiet}), "distinct, source has", len(ORDERS))
busy = crawl_by_offset(live=True)
print("busy source, offset pages: ", len(busy), "rows,", len({o["id"] for o in busy}), "distinct, source has", len(ORDERS),
      "after", COUNTERS["inserted"], "inserts during the crawl")
ceiling = max(o["id"] for o in ORDERS)
cursor = crawl_by_cursor(ceiling)
print("bounded live cursor crawl: ", len(cursor), "rows,", len({o["id"] for o in cursor}), "distinct, source has", len(ORDERS))

print("all IDs within the starting ceiling returned:", {o["id"] for o in cursor} == set(range(1, ceiling + 1)))
# quiet source, offset pages: 500 rows, 500 distinct, source has 500
# busy source, offset pages:  509 rows, 500 distinct, source has 511 after 11 inserts during the crawl
# bounded live cursor crawl:  511 rows, 511 distinct, source has 522
# all IDs within the starting ceiling returned: True

2. Count what the retry policy absorbed. Fetch eight cursor pages with a client that records every response code, and print the codes in order. Then call the API with a stale token and show what happens.

def fetch_with_stats(url, tries=6):
    outcomes = []
    return get(url, tries=tries, outcomes=outcomes), outcomes

COUNTERS["requests"] = 0
log, after = [], None
for page in range(8):
    response, outcomes = fetch_with_stats(f"{BASE}/orders?limit=50" + (f"&after={after}" if after else ""))
    log += outcomes
    after = response["next_after"]
print("8 pages took", len(log), "requests:", {code: log.count(code) for code in sorted(set(log))})
print("in order:", log)

try:
    urllib.request.urlopen(urllib.request.Request(BASE + "/orders", headers={"Authorization": "Bearer stale-token"}))
except urllib.error.HTTPError as error:
    print("with a stale token:", error.code, "and no blind retry")
    error.close()
# 8 pages took 11 requests: {200: 8, 429: 2, 500: 1}
# in order: [200, 200, 200, 429, 200, 200, 500, 429, 200, 200, 200]
# with a stale token: 401 and no blind retry

3. Validate the drop folder. Run the validator over every file in the drop and print each verdict. Then look at the first bytes of the second day’s file and explain what a plain UTF-8 decode would have made of the header.

for name in sorted(os.listdir(DROP)):
    if name.endswith(".csv"):
        print(f"{name}: {validate(name)}")
    elif name.endswith(".part"):
        print(f"{name}: waiting: upload still in progress")
print("day 2 starts with bytes", open(os.path.join(DROP, "sales_2026-03-02.csv"), "rb").read(5),
      "which utf-8 would read as", open(os.path.join(DROP, "sales_2026-03-02.csv"), "rb").read(5).decode("utf-8")[:1].encode("unicode_escape"))
# sales_2026-03-01.csv: accepted: 300 rows
# sales_2026-03-02.csv: accepted: 320 rows
# sales_2026-03-03.csv: quarantine: header is 'order_id;amount_cents;country'
# sales_2026-03-05.csv: quarantine: 200 rows but manifest promises 290
# sales_2026-03-06.csv.part: waiting: upload still in progress
# day 2 starts with bytes b'\xef\xbb\xbfor' which utf-8 would read as b'\\ufeff'

4. Redelivery and restatement. Load days one and two through a validated ledger keyed by business date with revision and checksum. Have the partner deliver day one again unchanged, and day two again with two rows fewer, and load both again.

store = {"ledger": {}, "published": {}, "audit": []}

def load(name, fail_before_publish=False):
    global store
    status, candidate = inspect_file(name)
    if candidate is None:
        return status
    rows, manifest, payload = candidate
    day, revision, digest = manifest["business_date"], manifest["revision"], manifest["sha256"]
    old = store["ledger"].get(day)
    if old and revision < old[0]:
        return f"{day}: stale revision, skipped"
    if old and revision == old[0]:
        if digest != old[1]:
            return f"{day}: quarantine: conflicting content for one revision"
        return f"{day}: same revision and checksum, skipped"
    verdict = (f"{day}: restatement, partition replaced ({len(store['published'][day])} rows became {len(rows)})"
               if old else f"{day}: new, {len(rows)} rows loaded")
    candidate_store = {"ledger": dict(store["ledger"], **{day: (revision, digest)}),
                       "published": dict(store["published"], **{day: rows}),
                       "audit": store["audit"] + [(day, revision, digest)]}
    if fail_before_publish:
        raise RuntimeError("simulated publication failure")
    store = candidate_store
    return verdict

print(load("sales_2026-03-01.csv"))
print(load("sales_2026-03-02.csv"))
deliver(first_day, partner_rows(first_day, 300))
print(load("sales_2026-03-01.csv"))
deliver(first_day + timedelta(days=1), partner_rows(first_day + timedelta(days=1), 318), revision=2)
print(load("sales_2026-03-02.csv"))
print("published:", {day: len(rows) for day, rows in sorted(store["published"].items())})
# 2026-03-01: new, 300 rows loaded
# 2026-03-02: new, 320 rows loaded
# 2026-03-01: same revision and checksum, skipped
# 2026-03-02: restatement, partition replaced (320 rows became 318)
# published: {'2026-03-01': 300, '2026-03-02': 318}

5. Select notification versions. Process a stream of object-created notifications that contains a duplicate, a newer version announced before an older one, and a repeat of the newer version, using immutable version IDs and numeric hexadecimal sequencers for a single simulated bucket.

notifications = [
    {"key": "landing/day1.csv", "version": "v1", "sequencer": "1"},
    {"key": "landing/day1.csv", "version": "v1", "sequencer": "01"},
    {"key": "landing/day2.csv", "version": "v2", "sequencer": "10"},
    {"key": "landing/day2.csv", "version": "v1", "sequencer": "f"},
    {"key": "landing/day3.csv", "version": "v1", "sequencer": "5"},
    {"key": "landing/day2.csv", "version": "v2", "sequencer": "10"},
]
state, latest = {}, {}

def handle_notice(notice, fail=False):
    key, version = notice["key"], notice["version"]
    if not re.fullmatch(r"[0-9a-fA-F]+", notice["sequencer"]):
        raise ValueError("invalid sequencer")
    seq = int(notice["sequencer"], 16)
    pair = (key, version)
    if pair in state:
        if state[pair] != seq:
            raise ValueError("one version has conflicting sequencers")
        return "duplicate notification, ignored"
    if key in latest and seq < latest[key]:
        return "older than the selected version, ignored"
    if key in latest and seq == latest[key]:
        raise ValueError("equal sequencer with different version")
    if fail:
        raise RuntimeError("simulated processing failure")
    # Selection simulation only: no object fetch, validation, or publication occurs.
    state[pair], latest[key] = seq, seq
    return "version selected"

for notice in notifications:
    print(notice["key"], notice["version"] + ":", handle_notice(notice))
print(len(notifications), "notifications,", len(state), "versions selected")
# landing/day1.csv v1: version selected
# landing/day1.csv v1: duplicate notification, ignored
# landing/day2.csv v2: version selected
# landing/day2.csv v1: older than the selected version, ignored
# landing/day3.csv v1: version selected
# landing/day2.csv v2: duplicate notification, ignored
# 6 notifications, 3 versions selected

6. The completeness report. It is 08:00 on 7 March and the contract promises each day’s file by 06:00 the next morning. For each of the seven expected days, report whether an accepted file exists, and for every day that is missing, in progress, or quarantined, say how overdue it is.

now = datetime(2026, 3, 7, 8, 0, tzinfo=timezone.utc)
expected = [first_day + timedelta(days=k) for k in range(7)]

def completeness(days, now):
    if not isinstance(now, datetime) or now.utcoffset() is None:
        raise ValueError("now must be timezone-aware")
    counts = {"accepted": 0, "overdue": 0, "not yet due": 0}
    report = []
    for day in days:
        due = datetime.combine(day + timedelta(days=1), datetime.min.time(), tzinfo=timezone.utc) + timedelta(hours=6)
        name = f"sales_{day.isoformat()}.csv"
        if os.path.exists(os.path.join(DROP, name + ".manifest.json")):
            status = validate(name)
        elif os.path.exists(os.path.join(DROP, name)):
            status = validate(name)
        elif os.path.exists(os.path.join(DROP, name + ".part")):
            status = "upload in progress"
        else:
            status = "not delivered"
        if status.startswith("accepted"):
            counts["accepted"] += 1
        elif now >= due:
            counts["overdue"] += 1
            report.append(f"{day}: {status}; was due {due:%m-%d %H:%M}, {int((now-due).total_seconds() // 3600)} hours ago")
        else:
            counts["not yet due"] += 1
    return counts, report

counts, report = completeness(expected, now)
print(len(expected), "days expected:", counts)
for line in report:
    print(" ", line)
# 7 days expected: {'accepted': 2, 'overdue': 4, 'not yet due': 1}
#   2026-03-03: quarantine: header is 'order_id;amount_cents;country'; was due 03-04 06:00, 74 hours ago
#   2026-03-04: not delivered; was due 03-05 06:00, 50 hours ago
#   2026-03-05: quarantine: 200 rows but manifest promises 290; was due 03-06 06:00, 26 hours ago
#   2026-03-06: upload in progress; was due 03-07 06:00, 2 hours ago

server.shutdown()
server.server_close()
drop_dir.cleanup()

7. Write the partner contract. For the sales file source, write the one-page contract from the section above, filling every line from what the simulator and the exercises revealed.

The final code block stops the server and removes the temporary folder. If you stop earlier, call server.shutdown(), server.server_close(), and drop_dir.cleanup() to stop the loopback server and remove the temporary drop folder.