Build a Metron Bronze Ingestor

Build a rate-aware Python client that writes Metron data to Bronze JSONL.

Introduction

30 Second Summary

A data download can finish cleanly while quietly missing later pages. It can also overload a service when it ignores the usage limits attached to each request.

In this project, you will build a rate-aware local Python ingestion command that retrieves a bounded set of issues from the Metron API through sequential pagination. Each run leaves behind a validated Bronze dataset stored as JSON Lines with the original source payloads intact.

What You'll Build

Picture your terminal ending a rate-aware Metron run with a timestamped file containing exactly the issue count you requested.

By the end of this project, you'll have:

  • A bounded ingestion command that follows Metron's ready-to-use next URLs until the requested issue count is on disk.
  • A timestamped Bronze dataset stored as independently parseable JSON lines. Each original issue remains unchanged under payload with its ingestion metadata.
  • Visible rate limiting for both burst and sustained windows. A local simulation proves that the client honors Retry-After while preserving your live quota. The final summary reports the operational result of each run.
  • Secret Mission: Generalize the issues-only command so the same ingestion engine can fetch characters into separate validated Bronze JSONL files.

Are there any prerequisites?

You'll need an existing Metron account with an API token. Your Mac also needs Python 3.11 or newer alongside VS Code and Git.

Before We Start

This checkpoint locks in your goal: a rate-aware local Metron Bronze ingestion client. It also connects trustworthy source data to the later ComicDB Platform stages.

Set Up the Local Ingestion Project

Your Metron client needs stable dependencies before it can collect trustworthy data. A virtual environment isolates this project's Python packages from the rest of your Mac.

Your API token also needs protection from source control. A session-only environment variable gives Python access to the token without storing it in a project file.

In this step, get ready to:
  • Create a local Git repository for the ingestion project.
  • Install Requests version 2.34.2 inside an isolated environment.
  • Protect the Metron token within the current terminal session.
Create the repository in VS Code

A dedicated folder keeps the code under one predictable path. Initializing Git there prepares the project for version history.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop

What does this command do?

The cd command changes the terminal's current location. Starting on your Desktop makes the new repository easy to find.

  • Create the project folder and initialize its repository by running these commands:
mkdir comicdb-metron-ingestor
cd comicdb-metron-ingestor
git init

What did these commands do?

  • The first command creates the comicdb-metron-ingestor folder on your Desktop.
  • The second command moves the terminal into that folder.
  • The final command creates the repository metadata that Git uses to track changes.

Your terminal confirms that Git initialized an empty repository. That repository is now the local home for your ingestion client.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to return to your search bar.
  • Type VS Code and press Enter to open VS Code.
  • Click File in the top menu bar.
  • Click Open Folder.
  • Select the comicdb-metron-ingestor folder on your Desktop.
  • Click Open to load the repository.

You'll see the empty comicdb-metron-ingestor folder in the Explorer sidebar. This confirms that VS Code is working inside the repository you initialized.

Build the isolated Python environment

The virtual environment keeps version 2.34.2 of Requests attached to this repository. Activating it also makes later Python commands use the isolated package set.

  • Click Terminal in the VS Code menu bar.
  • Click New Terminal to start a zsh terminal inside comicdb-metron-ingestor.
  • Create and activate the virtual environment by running these commands:
python3 -m venv .venv
source .venv/bin/activate

How does the environment stay isolated?

The first command creates the .venv directory with its own Python package location. The activation command updates the current shell so python uses that location.

You'll see (.venv) at the start of the terminal prompt. That prefix confirms the isolated environment is active.

  • Create the three starter files by running this command:
touch .gitignore requirements.txt ingest.py

What did this command create?

The command creates empty files for dependency protection, dependency pinning, and the Python entry point. You'll see .gitignore, requirements.txt, and ingest.py in the Explorer sidebar.

  • Select requirements.txt in the Explorer sidebar.
  • Add the pinned dependency by copying this line into the file:
requests==2.34.2

Why pin the dependency?

The exact version keeps every installation on the same Requests release. This removes dependency drift from later ingestion tests.

  • Save requirements.txt.
  • Install its pinned dependency by running this command:
python -m pip install -r requirements.txt

What does this installation use?

The command asks the active Python environment to install every entry in requirements.txt. The file pins Requests to version 2.34.2.

The terminal finishes with a successful installation message for Requests 2.34.2. Your repository now has its isolated HTTP dependency.

Having trouble installing Requests?

Confirm that your terminal prompt starts with (.venv). Return to the activation command above if the prefix is missing.

Check that requirements.txt contains the pinned line exactly. A spelling error prevents the package installer from resolving the dependency.

Ask for help with the installation output: Help me diagnose why Requests 2.34.2 is not installing in my active virtual environment.

Protect credentials and verify the setup

Generated environments, Python caches, and ingestion output do not belong in source history. The .gitignore file tells Git to leave those paths untracked.

  • Select .gitignore in the Explorer sidebar.
  • Add the repository exclusions by copying this content into the file:
.venv/
__pycache__/
*.pyc
data/

What stays out of Git?

  • The .venv/ rule excludes the local virtual environment.
  • The __pycache__/ rule excludes generated Python cache directories.
  • The *.pyc rule excludes compiled Python files.
  • The data/ rule excludes generated ingestion datasets.
  • Save .gitignore.
  • Click the Source Control icon in the left sidebar to check which files Git can see.

You'll see the three starter files as project changes. You should not see the .venv directory listed.

Is the virtual environment listed?

Check that the first line of .gitignore is exactly .venv/. Save the file before checking Source Control again.

Get help checking the ignore rule: Help me find out why Git can still see my .venv directory.

The starter Python module establishes the imports and resource identifiers used by the first live request. Keeping this foundation small makes the next result quick to test.

  • Return to the Explorer sidebar.
  • Select ingest.py.
  • Add the starter module by copying this code into the file:
import argparse
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

import requests

API_URL = "https://metron.cloud/api/issue/"
ENTITY_TYPE = "issue"

What does this starter module prepare?

  • The standard-library imports prepare argument parsing, JSON handling, environment lookup, timing, timestamps, and file paths.
  • The Requests import prepares the HTTP client used for the live Metron call.
  • The API_URL value identifies the issue list endpoint.
  • The ENTITY_TYPE value identifies records as issues inside the Bronze envelope.
  • Save ingest.py.

Typing a token into a shell can feel exposed. This export keeps it in the current zsh session without writing it to a project file.

  • Switch back to the integrated terminal from earlier.
  • Replace your-api-token locally with your Metron token before running this command:
export METRON_TOKEN='your-api-token'

Where does the token live?

The export makes METRON_TOKEN available to programs launched from this terminal session. A new terminal session does not inherit it.

The token stays outside ingest.py, requirements.txt, and .gitignore. This protects the credential from accidental commits.

  • Predict whether Python can see both the pinned Requests version and the session token.
  • Verify the environment by running this command:
python -c "import os, requests; print(requests.__version__, bool(os.getenv('METRON_TOKEN')))"

What should you see?

You'll see 2.34.2 True in the terminal. The version confirms the pinned package is active.

The final value confirms that Python can read the session token. The command prints only a Boolean value, so the credential remains hidden.

That's the local foundation complete. Your isolated repository can now import Requests and read the protected Metron token.

Seeing a different version or False?

If the version differs, confirm the prompt starts with (.venv). Repeat the installation inside that active environment.

If the final value is False, repeat the token export in the same terminal used for the verification.

Ask for help with your setup result: Help me diagnose why my setup check does not print 2.34.2 True.

Use these tabs to compare every starter file before moving on.

✔️ My starter files match

Great. Save all three files before completing the checkpoint.

ⓧ I'd like to double check the full code

Check .gitignore
.venv/
__pycache__/
*.pyc
data/
Check requirements.txt
requests==2.34.2
Check ingest.py
import argparse
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

import requests

API_URL = "https://metron.cloud/api/issue/"
ENTITY_TYPE = "issue"

Your local ingestion workspace is ready. Next up, you'll fetch one authenticated Metron page and write its issue records as Bronze JSONL.

Fetch One Page and Write Bronze JSONL

Your local repository can now import Requests without exposing your Metron token. That foundation is ready for its first live data request.

This step sends one authenticated request to the Metron API. The response becomes a timestamped JSON Lines file that you can inspect on disk.

In this step, get ready to:
  • Parse a positive record boundary from the command line.
  • Fetch one authenticated issue page from Metron.
  • Write the response as timestamped Bronze JSONL.
Parse the run boundary and token

The command needs a record boundary so each run stays controlled. The token remains in the METRON_TOKEN environment variable from the previous step.

  • In ingest.py, place your cursor below the ENTITY_TYPE constant.
  • Add the timeout setting plus the argument and token functions by pasting this code:
TIMEOUT_SECONDS = 30

def parse_args():
    # Define the record boundary accepted by each ingestion run
    parser = argparse.ArgumentParser(
        description="Ingest a bounded set of Metron issues into Bronze JSONL."
    )
    parser.add_argument(
        "--max-records",
        type=int,
        default=150,
        help="maximum number of records to write",
    )
    args = parser.parse_args()
    if args.max_records < 1:
        parser.error("--max-records must be at least 1")
    return args

def require_token():
    # Read the credential from the current shell without storing it in code
    token = os.getenv("METRON_TOKEN")
    if not token:
        raise RuntimeError(
            "METRON_TOKEN is not set. Export it in the current terminal session."
        )
    return token

What does this code do?

  • The --max-records argument controls the maximum number of issue records written by a run.
  • The positive-value check rejects boundaries below 1.
  • The require_token() function reads the credential from the current shell.
  • The TIMEOUT_SECONDS constant limits how long the request waits for server data.
  • Save ingest.py.
  • Check the editor for Python syntax warnings around parse_args() or require_token().

Seeing a syntax warning?

Check that both functions start at the left edge of ingest.py. Confirm that the closing parentheses beneath each multi-line call line up with the statement that opened them.

If the warning remains, help me compare my argument and token functions with the expected Python indentation.

Fetch one authenticated page

A Bearer token identifies your request without placing the credential in the URL. A session applies that authorization header to the issue request.

Each source object goes into a Bronze envelope under payload. The surrounding fields record where the object came from and when you ingested it.

  • In ingest.py, place your cursor below the require_token() function.
  • Add the output-path and one-page ingestion functions by pasting this code:
def make_output_path(entity_type):
    # Give every run its own UTC-stamped data artifact
    data_directory = Path("data")
    data_directory.mkdir(parents=True, exist_ok=True)
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    return data_directory / f"metron_{entity_type}_{run_id}.jsonl"

def ingest(token, max_records):
    # Fetch one live page and preserve each issue inside a Bronze envelope
    output_path = make_output_path(ENTITY_TYPE)

    with requests.Session() as session:
        session.headers.update({"Authorization": f"Bearer {token}"})
        response = session.get(API_URL, timeout=TIMEOUT_SECONDS)
        response.raise_for_status()
        page = response.json()
        results = page.get("results")
        ingested_at = datetime.now(timezone.utc).isoformat()

        with output_path.open("x", encoding="utf-8") as destination:
            for issue in results[:max_records]:
                envelope = {"source": "metron", "entity_type": ENTITY_TYPE, "ingested_at": ingested_at, "source_endpoint": response.url, "payload": issue}
                destination.write(json.dumps(envelope, ensure_ascii=False) + "\n")

        next_url = page.get("next")
        print(f"Page 1: {min(len(results), max_records)}/{max_records} records written")
        print(f"Next page available: {bool(next_url)}")
        return output_path, min(len(results), max_records)

How does the first page become Bronze data?

  • The requests.Session() object carries the authorization header into the request.
  • The request uses API_URL as the issue-list endpoint.
  • The UTC timestamp gives every run a unique file under data/.
  • The payload field receives the complete issue object returned by Metron.
  • The next value records whether Metron has another page ready.
  • Save ingest.py.
  • Check that the new functions appear below require_token() with no indentation warnings.

Seeing an indentation warning in ingest?

Keep the request and file-writing code indented inside the with requests.Session() block. Keep the envelope and write call indented inside the for issue loop.

For a closer check, help me repair the indentation in my one-page ingest function.

Run and inspect the Bronze file

The remaining function connects command-line input to the ingestion run. Its error boundary turns common failures into a readable terminal message.

  • In ingest.py, place your cursor below the ingest() function.
  • Add the command entry point by pasting this code:
def main():
    # Connect the command-line boundary to the one-page ingestion run
    args = parse_args()
    token = require_token()

    output_path, records_written = ingest(
        token=token,
        max_records=args.max_records,
    )

    print("\nRun complete")
    print(f"Records written: {records_written}")
    print(f"Output path: {output_path}")

if __name__ == "__main__":
    try:
        main()
    except (
        OSError,
        RuntimeError,
        ValueError,
        KeyError,
        requests.exceptions.RequestException,
    ) as exc:
        print(f"Error: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

What does the entry point do?

  • The main() function collects the parsed boundary and session token.
  • The call to ingest() starts the live request.
  • The completion output identifies the number of records written and the unique artifact path.
  • The exception boundary prints operational failures without printing the token.

✔️ Awesome, I've got everything!

Your command-line interface, one-page request, Bronze writer, and entry point are ready for the live check.

ⓧ I'd like to double check the full code

import argparse
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

import requests

API_URL = "https://metron.cloud/api/issue/"
ENTITY_TYPE = "issue"
TIMEOUT_SECONDS = 30

def parse_args():
    # Define the record boundary accepted by each ingestion run
    parser = argparse.ArgumentParser(
        description="Ingest a bounded set of Metron issues into Bronze JSONL."
    )
    parser.add_argument(
        "--max-records",
        type=int,
        default=150,
        help="maximum number of records to write",
    )
    args = parser.parse_args()
    if args.max_records < 1:
        parser.error("--max-records must be at least 1")
    return args

def require_token():
    # Read the credential from the current shell without storing it in code
    token = os.getenv("METRON_TOKEN")
    if not token:
        raise RuntimeError(
            "METRON_TOKEN is not set. Export it in the current terminal session."
        )
    return token

def make_output_path(entity_type):
    # Give every run its own UTC-stamped data artifact
    data_directory = Path("data")
    data_directory.mkdir(parents=True, exist_ok=True)
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    return data_directory / f"metron_{entity_type}_{run_id}.jsonl"

def ingest(token, max_records):
    # Fetch one live page and preserve each issue inside a Bronze envelope
    output_path = make_output_path(ENTITY_TYPE)

    with requests.Session() as session:
        session.headers.update({"Authorization": f"Bearer {token}"})
        response = session.get(API_URL, timeout=TIMEOUT_SECONDS)
        response.raise_for_status()
        page = response.json()
        results = page.get("results")
        ingested_at = datetime.now(timezone.utc).isoformat()

        with output_path.open("x", encoding="utf-8") as destination:
            for issue in results[:max_records]:
                envelope = {"source": "metron", "entity_type": ENTITY_TYPE, "ingested_at": ingested_at, "source_endpoint": response.url, "payload": issue}
                destination.write(json.dumps(envelope, ensure_ascii=False) + "\n")

        next_url = page.get("next")
        print(f"Page 1: {min(len(results), max_records)}/{max_records} records written")
        print(f"Next page available: {bool(next_url)}")
        return output_path, min(len(results), max_records)

def main():
    # Connect the command-line boundary to the one-page ingestion run
    args = parse_args()
    token = require_token()

    output_path, records_written = ingest(
        token=token,
        max_records=args.max_records,
    )

    print("\nRun complete")
    print(f"Records written: {records_written}")
    print(f"Output path: {output_path}")

if __name__ == "__main__":
    try:
        main()
    except (
        OSError,
        RuntimeError,
        ValueError,
        KeyError,
        requests.exceptions.RequestException,
    ) as exc:
        print(f"Error: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc
  • Save ingest.py.

Before you run the command, do you think a boundary of 150 can be satisfied by this one-page client?

  • Run the one-page ingestion command in the terminal from earlier:
python ingest.py --max-records 150

What should I see?

You should see a page progress line followed by Next page available: True. The final output also gives you a unique path under data/.

The record count reflects only the first response page. That shortfall is intentional because the current client records Metron's next URL without following it.

You have your first live Bronze artifact on disk. It contains real issue payloads plus the metadata needed to trace every line back to its source.

  • In the VS Code file sidebar, expand the data/ folder.
  • Select the newest metron_issue_ JSONL file.
  • Inspect the first line for source, entity_type, ingested_at, source_endpoint, and payload.
  • Confirm that payload contains a complete issue object.

Did the live request fail?

If the command says the token is missing, confirm that you are using the same terminal session from the previous step. A new shell does not inherit the session-only token.

If the request fails for another reason, help me diagnose my Metron request without exposing my API token.

Your authenticated client can now turn one Metron response into a traceable Bronze file. Next, you will follow the API's ready-to-use pagination URLs until the requested record boundary is satisfied.

Follow Bounded Pagination

Your first live run proved that the authenticated Metron request can produce a real Bronze JSON Lines dataset. However, stopping after one page leaves most available issues behind.

Metron provides the next page as a ready-to-use URL. In this step, your client follows those URLs through sequential pagination while enforcing the requested record boundary.

In this step, get ready to:
  • Follow each returned page URL sequentially.
  • Slice the final page at the requested record boundary.
  • Confirm each envelope records its actual source page.
Follow Metron's returned page URLs

The response's next field contains the URL for the following page. The client can stop when that value becomes empty or when the requested record count is reached.

  • In ingest.py, find the beginning of ingest().
  • Replace the current one-page state with the following pagination state:
    metrics = {
        "pages": 0,
        "records": 0,
    }
    next_url = API_URL

What does this state track?

  • The metrics dictionary tracks completed pages and written records.
  • The next_url value begins at API_URL.
  • Each response later replaces next_url with Metron's returned URL.
  • Inside the existing output file context in ingest(), select the one-page request and write block.
  • Replace the selected block with this bounded pagination loop:
            while next_url and metrics["records"] < max_records:
                response = session.get(next_url, timeout=TIMEOUT_SECONDS)
                response.raise_for_status()
                page = response.json()
                results = page.get("results")

                remaining_slots = max_records - metrics["records"]
                ingested_at = datetime.now(timezone.utc).isoformat()
                for issue in results[:remaining_slots]:
                    envelope = {
                        "source": "metron",
                        "entity_type": ENTITY_TYPE,
                        "ingested_at": ingested_at,
                        "source_endpoint": response.url,
                        "payload": issue,
                    }
                    destination.write(
                        json.dumps(envelope, ensure_ascii=False) + "\n"
                    )

                metrics["pages"] += 1
                metrics["records"] += min(len(results), remaining_slots)
                next_url = page.get("next")
                print(
                    f"Page {metrics['pages']}: "
                    f"{metrics['records']}/{max_records} records written"
                )

    return output_path, metrics

How does bounded pagination work?

  • The while condition requires another page URL and room for more records.
  • The remaining_slots calculation measures how many records still fit.
  • The results[:remaining_slots] slice prevents the final page from exceeding the boundary.
  • The source_endpoint field stores response.url for the page that supplied each issue.
  • The page.get("next") value supplies the following request URL without constructed page numbers.
  • Save ingest.py.

Before you run the updated client, do you expect it to stop after one page or continue until it reaches the record boundary?

  • Test the pagination loop by running:
python ingest.py --max-records 150

When the API has enough issues, you'll see progress for at least two pages. The final progress message shows 150/150 records written.

Did the pagination run stop early?

Confirm that the active terminal still has METRON_TOKEN exported. A new terminal session does not inherit the session-only token.

Check that next_url receives page.get("next") inside the loop. Moving that assignment outside the loop prevents the URL from advancing.

Ask for help with the exact terminal output: Help me debug my bounded pagination loop

Check the page-specific envelopes

Each Bronze envelope needs traceability back to the page that supplied its payload. Using response.url preserves the actual URL after the request completes.

  • Return to the Explorer sidebar in VS Code.
  • Expand the data/ folder.
  • Open the newest metron_issue_ JSONL file.
  • Inspect an early record's source_endpoint value.
  • Inspect a record near the end of the file for its source_endpoint value.

You'll see complete Metron issue objects under payload. Records from later pages preserve the later page URL in source_endpoint.

Why keep the source page URL?

The page URL provides lineage for each stored issue. If a later pipeline stage finds an unexpected record, you can trace that record back to the exact API page used during ingestion.

Report and verify the bounded run

The progress messages prove that the loop advances through pages. A short completion summary makes the final boundary and output location easy to verify after the run.

  • In main(), select the current ingest() assignment and the old one-page status message.
  • Replace the selected block with this completion summary:
    output_path, metrics = ingest(
        token=token,
        max_records=args.max_records,
    )

    print("\nRun complete")
    print(f"Pages fetched: {metrics['pages']}")
    print(f"Records written: {metrics['records']}")
    print(f"Output path: {output_path}")

What does the summary prove?

  • The page count confirms that the client followed more than one response when required.
  • The record count confirms that the final page slice respected max_records.
  • The output path identifies the unique JSONL artifact created by this run.
  • Save ingest.py.

Before the final run, what page count do you expect for a boundary larger than Metron's first page?

  • Verify the finished bounded pagination behavior by running:
python ingest.py --max-records 150

You'll see at least two pages fetched and exactly 150 records written when the API has enough issues. The output path points to a new timestamped file under data/.

That is a major reliability step completed. Your client now follows Metron's navigation while keeping every run inside a deliberate record boundary.

Does the summary exceed the boundary?

Make sure the write loop iterates over results[:remaining_slots]. Iterating over all of results allows the final page to exceed the requested count.

Confirm that the record metric uses the smaller value from the page length and the remaining slots. This keeps the summary aligned with the lines written.

Share the count and the relevant function without sharing credentials: Help me fix a pagination boundary mismatch

✔️ Awesome, I've got everything!

Great. Double check that ingest.py is saved before moving on.

ⓧ I'd like to double check the full code

import argparse
import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

import requests

API_URL = "https://metron.cloud/api/issue/"
ENTITY_TYPE = "issue"
TIMEOUT_SECONDS = 30


def parse_args():
    parser = argparse.ArgumentParser(
        description="Ingest a bounded set of Metron issues into Bronze JSONL."
    )
    parser.add_argument(
        "--max-records",
        type=int,
        default=150,
        help="maximum number of records to write",
    )
    args = parser.parse_args()
    if args.max_records < 1:
        parser.error("--max-records must be at least 1")
    return args


def require_token():
    token = os.getenv("METRON_TOKEN")
    if not token:
        raise RuntimeError(
            "METRON_TOKEN is not set. Export it in the current terminal session."
        )
    return token


def make_output_path(entity_type):
    data_directory = Path("data")
    data_directory.mkdir(parents=True, exist_ok=True)
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    return data_directory / f"metron_{entity_type}_{run_id}.jsonl"


def ingest(token, max_records):
    output_path = make_output_path(ENTITY_TYPE)
    metrics = {
        "pages": 0,
        "records": 0,
    }
    next_url = API_URL

    with requests.Session() as session:
        session.headers.update(
            {
                "Authorization": f"Bearer {token}",
                "Accept": "application/json",
                "User-Agent": "comicdb-platform-ingestor/1.0",
            }
        )

        with output_path.open("x", encoding="utf-8") as destination:
            while next_url and metrics["records"] < max_records:
                response = session.get(next_url, timeout=TIMEOUT_SECONDS)
                response.raise_for_status()
                page = response.json()
                results = page.get("results")

                remaining_slots = max_records - metrics["records"]
                ingested_at = datetime.now(timezone.utc).isoformat()
                for issue in results[:remaining_slots]:
                    envelope = {
                        "source": "metron",
                        "entity_type": ENTITY_TYPE,
                        "ingested_at": ingested_at,
                        "source_endpoint": response.url,
                        "payload": issue,
                    }
                    destination.write(
                        json.dumps(envelope, ensure_ascii=False) + "\n"
                    )

                metrics["pages"] += 1
                metrics["records"] += min(len(results), remaining_slots)
                next_url = page.get("next")
                print(
                    f"Page {metrics['pages']}: "
                    f"{metrics['records']}/{max_records} records written"
                )

    return output_path, metrics


def main():
    args = parse_args()
    token = require_token()

    output_path, metrics = ingest(
        token=token,
        max_records=args.max_records,
    )

    print("\nRun complete")
    print(f"Pages fetched: {metrics['pages']}")
    print(f"Records written: {metrics['records']}")
    print(f"Output path: {output_path}")


if __name__ == "__main__":
    try:
        main()
    except (
        OSError,
        RuntimeError,
        ValueError,
        KeyError,
        requests.exceptions.RequestException,
    ) as exc:
        print(f"Error: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

Your ingestor can now retrieve a bounded dataset across multiple pages. Next, you'll protect those sequential requests with rate-limit monitoring and bounded retries.

Add Rate Limits and Safe Retries

Your bounded paginator now follows the Metron API across multiple pages. An unattended ingestion run also needs to stay reliable when the service slows down or temporarily rejects a request.

This step adds rate limiting for Metron's sustained and burst windows. It also gives each HTTP failure the recovery behavior that fits it.

In this step, get ready to:
  • Route every request through a timeout and bounded retry policy.
  • Calculate adaptive waits from both Metron rate-limit windows.
  • Simulate a safe 429 recovery before the first live request.
Build the bounded retry path

Temporary network failures can succeed on a later attempt. Bounded exponential backoff gives the service more time between attempts while keeping the ingestion run from retrying forever.

  • In ingest.py, add the retry settings directly below ENTITY_TYPE by pasting this code:
RATE_LIMIT_SCOPES = ("Sustained", "Burst")
TIMEOUT_SECONDS = 30
MAX_RETRIES = 3
BACKOFF_CAP_SECONDS = 8

What do these settings control?

  • RATE_LIMIT_SCOPES keeps the sustained window first so the client evaluates it before the burst window.
  • TIMEOUT_SECONDS prevents a request from waiting indefinitely for server data.
  • MAX_RETRIES allows three retries after the first attempt.
  • BACKOFF_CAP_SECONDS places an upper bound on calculated backoff delays.
  • Below require_token(), add the header parser and backoff calculator by pasting this code:
def retry_after_seconds(headers):
    value = headers.get("Retry-After")
    if value is None:
        raise RuntimeError("A 429 response did not include Retry-After.")
    try:
        return max(0, int(value))
    except ValueError as exc:
        raise RuntimeError("Retry-After was not an integer number of seconds.") from exc


def bounded_backoff(attempt):
    return min(2**attempt, BACKOFF_CAP_SECONDS)

How are the waits calculated?

retry_after_seconds() requires the server-provided Retry-After value for a 429 response. It rejects missing or non-integer values so the client never invents a throttling delay.

bounded_backoff() produces waits of 1, 2, and 4 seconds for the three retries. The cap keeps later calculations bounded.

  • Save ingest.py.
  • Confirm the existing paginator still runs by executing this command:
python ingest.py --max-records 150

What should I see?

You should still see sequential page progress ending at 150 records. This confirms the new constants and helpers compile without changing bounded pagination.

Seeing a Python error?

Check that the constants sit at the top level of ingest.py. Confirm both helper functions also start at the left edge of the file.

Help me fix my retry helper syntax

  • Inside parse_args(), add the simulation argument immediately before args = parser.parse_args() by pasting this code:
    parser.add_argument(
        "--simulate-429",
        action="store_true",
        help="simulate one Retry-After response before the first live request",
    )

What does the simulation flag do?

The --simulate-429 flag lets the client exercise its retry parser with a local header dictionary. The simulation waits once before the first live request, so it demonstrates recovery without exhausting API quota.

  • Add request_page() below bounded_backoff() by pasting this first working version:
def request_page(session, url, metrics, simulate_429):
    if simulate_429 and not metrics["simulation_used"]:
        wait = retry_after_seconds({"Retry-After": "1"})
        metrics["simulation_used"] = True
        metrics["retries"] += 1
        print(f"Simulated 429: honoring Retry-After and waiting {wait}s")
        time.sleep(wait)

    for attempt in range(MAX_RETRIES + 1):
        try:
            response = session.get(url, timeout=TIMEOUT_SECONDS)
        except requests.exceptions.RequestException as exc:
            if attempt == MAX_RETRIES:
                raise
            wait = bounded_backoff(attempt)
            metrics["retries"] += 1
            print(f"Network failure: {exc}. Retrying in {wait}s")
            time.sleep(wait)
            continue

        response.raise_for_status()
        return response

    raise RuntimeError("Request attempts were exhausted without a response.")

What does this request wrapper do?

  • The simulation block parses a local Retry-After value once per run.
  • The request loop gives Requests one initial attempt plus three retries.
  • The timeout argument limits every request to 30 seconds.
  • RequestException covers network failures raised through the Requests exception hierarchy.
  • Update the ingest() definition to accept the simulation setting:
def ingest(token, max_records, simulate_429):

Why pass the flag into ingest?

The command-line layer owns the learner's choice to simulate a response. Passing that choice into ingest() keeps the network behavior explicit.

  • Replace the existing main() function with this version:
def main():
    args = parse_args()
    token = require_token()

    output_path, metrics = ingest(
        token=token,
        max_records=args.max_records,
        simulate_429=args.simulate_429,
    )

    print("\nRun complete")
    print(f"Pages fetched: {metrics['pages']}")
    print(f"Records written: {metrics['records']}")
    print(f"Retries: {metrics['retries']}")
    print(f"Output path: {output_path}")

What changed in main?

main() now passes the simulation choice into ingest(). The run summary also reports the retry count so recovery behavior remains visible.

  • Save ingest.py.
  • Confirm the updated command-line interface still completes by running:
python ingest.py --max-records 150

What should I see now?

You should see the usual page progress followed by a run summary. The summary now includes a retry count of zero when the live requests succeed on their first attempts.

  • Inside the metrics dictionary in ingest(), add this simulation state:
        "simulation_used": False,

Why track simulation state?

simulation_used makes the local 429 exercise run exactly once. Later page requests skip it even when pagination continues.

  • Inside the pagination loop, replace the direct session.get() call with this request wrapper call:
                response = request_page(
                    session,
                    next_url,
                    metrics,
                    simulate_429,
                )

Why route every page through one function?

Every sequential page now receives the same timeout and retry policy. Centralizing network access also prevents later pagination changes from bypassing the safety controls.

  • Save ingest.py.

Before you run this check, consider whether the simulated retry lets the live ingestion continue.

  • Exercise the local 429 simulation by running:
python ingest.py --max-records 150 --simulate-429

What should I see?

You should see Simulated 429: honoring Retry-After and waiting 1s before the first live page. The run should then continue to 150 records with one retry in the summary.

Simulation missing from the log?

Confirm that simulate_429=args.simulate_429 appears in main(). Check that the pagination loop calls request_page() instead of calling the session directly.

Help me trace the simulated 429 flag

Separate 429s from transient failures

A 429 response carries a server-selected wait in Retry-After. Network failures and 5xx responses use bounded exponential backoff because they do not provide that throttling instruction.

  • Inside request_page(), place these response branches immediately before response.raise_for_status():
        if response.status_code == 429:
            if attempt == MAX_RETRIES:
                response.raise_for_status()
            wait = retry_after_seconds(response.headers)
            metrics["retries"] += 1
            print(f"HTTP 429: honoring Retry-After and waiting {wait}s")
            time.sleep(wait)
            continue

        if 500 <= response.status_code < 600:
            if attempt == MAX_RETRIES:
                response.raise_for_status()
            wait = bounded_backoff(attempt)
            metrics["retries"] += 1
            print(f"HTTP {response.status_code}: retrying in {wait}s")
            time.sleep(wait)
            continue

How does each response behave?

  • A 429 response waits for the integer value supplied through Retry-After.
  • A 5xx response waits for 1, 2, or 4 seconds according to the current attempt.
  • A network exception follows the same bounded backoff schedule.
  • Another 4xx response reaches raise_for_status() immediately because repeating the same request will not correct it.
  • Save ingest.py.
  • Confirm the complete response policy still permits a successful run by executing:
python ingest.py --max-records 150 --simulate-429

What does this check prove?

The one-second local simulation should complete before the live requests begin. The remaining requests should still follow Metron's sequential next-page URLs until 150 records are written.

Run stops after the simulated wait?

Check that each retry branch ends with continue. Confirm response.raise_for_status() remains below both status branches.

Help me debug the request retry loop

Read both rate-limit windows

Metron returns separate limit, remaining, and reset headers for sustained and burst traffic. The client must require all six values because either window can require the longer pause.

  • Below request_page(), add the header contract and initial wait calculation by pasting this function:
def rate_limit_wait(headers):
    required_headers = []
    for scope in RATE_LIMIT_SCOPES:
        required_headers.extend(
            [
                f"X-RateLimit-{scope}-Limit",
                f"X-RateLimit-{scope}-Remaining",
                f"X-RateLimit-{scope}-Reset",
            ]
        )

    missing = [name for name in required_headers if headers.get(name) is None]
    if missing:
        raise RuntimeError(
            "Missing required Metron rate-limit headers: " + ", ".join(missing)
        )

    wait = 0
    now = int(time.time())

    return wait

Why require every header?

The function builds the six expected header names from RATE_LIMIT_SCOPES. A missing value stops the run because an incomplete quota picture could lead the next request into avoidable throttling.

now captures the current Unix timestamp. The reset calculations use it to turn each reset timestamp into a delay.

  • Inside rate_limit_wait(), insert this scope loop immediately before return wait:
    for scope in RATE_LIMIT_SCOPES:
        try:
            limit = int(headers[f"X-RateLimit-{scope}-Limit"])
            remaining = int(headers[f"X-RateLimit-{scope}-Remaining"])
            reset = int(headers[f"X-RateLimit-{scope}-Reset"])
        except ValueError as exc:
            raise RuntimeError("A Metron rate-limit header was not an integer.") from exc

        print(
            f"{scope} limit: {remaining}/{limit} remaining; "
            f"reset timestamp: {reset}"
        )

        if remaining <= 0:
            wait = max(wait, max(0, reset - now))

How is the adaptive wait chosen?

The loop parses and logs Sustained before Burst. When either remaining counter reaches zero, max() keeps the longest delay required by the two reset timestamps.

A positive delay protects both windows with one sequential pause. A zero value allows the next page request to proceed.

  • Save ingest.py.
  • Confirm the completed helper compiles by running the ingestor:
python ingest.py --max-records 150

What should happen?

The existing ingestion should still finish at 150 records. This run confirms the complete rate-limit helper compiles before it becomes part of the page loop.

  • Inside the pagination loop, add this rate-limit check immediately after the page progress message:
                wait = rate_limit_wait(response.headers)
                print(f"Calculated adaptive wait: {wait}s")

                if next_url and metrics["records"] < max_records and wait > 0:
                    time.sleep(wait)

Why check after every page?

Each successful response carries the current state of both quota windows. The client calculates the next safe request time after writing that page, then sleeps only when another page is needed.

  • Save ingest.py.

Before you run the final check, consider which wait message should appear before the first live page.

  • Verify the complete retry and rate-limit path by running:
python ingest.py --max-records 150 --simulate-429

What should I see?

You should first see a one-second simulated Retry-After wait. Each live page should then print Sustained counters, Burst counters, and a calculated adaptive wait.

The completed run should report at least two pages when Metron has enough issues. It should also report exactly 150 records and one simulated retry.

That is the resilience layer working: your paginator now recovers safely while respecting both live quota windows.

Missing rate-limit counters?

Confirm that rate_limit_wait(response.headers) runs after every successful page. Check each header name against its Sustained or Burst scope if the client reports a missing header.

Help me debug the Metron rate-limit headers

Use these tabs to compare your finished ingest.py with the complete rate-aware version.

✔️ Awesome, I've got everything!

Great. Double check that you saved ingest.py before moving on.

ⓧ I'd like to double check the full code

import argparse
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

import requests

API_URL = "https://metron.cloud/api/issue/"
ENTITY_TYPE = "issue"
RATE_LIMIT_SCOPES = ("Sustained", "Burst")
TIMEOUT_SECONDS = 30
MAX_RETRIES = 3
BACKOFF_CAP_SECONDS = 8


def parse_args():
    parser = argparse.ArgumentParser(
        description="Ingest a bounded set of Metron issues into Bronze JSONL."
    )
    parser.add_argument(
        "--max-records",
        type=int,
        default=150,
        help="maximum number of records to write",
    )
    parser.add_argument(
        "--simulate-429",
        action="store_true",
        help="simulate one Retry-After response before the first live request",
    )
    args = parser.parse_args()
    if args.max_records < 1:
        parser.error("--max-records must be at least 1")
    return args


def require_token():
    token = os.getenv("METRON_TOKEN")
    if not token:
        raise RuntimeError(
            "METRON_TOKEN is not set. Export it in the current terminal session."
        )
    return token


def retry_after_seconds(headers):
    value = headers.get("Retry-After")
    if value is None:
        raise RuntimeError("A 429 response did not include Retry-After.")
    try:
        return max(0, int(value))
    except ValueError as exc:
        raise RuntimeError("Retry-After was not an integer number of seconds.") from exc


def bounded_backoff(attempt):
    return min(2**attempt, BACKOFF_CAP_SECONDS)


def request_page(session, url, metrics, simulate_429):
    if simulate_429 and not metrics["simulation_used"]:
        wait = retry_after_seconds({"Retry-After": "1"})
        metrics["simulation_used"] = True
        metrics["retries"] += 1
        print(f"Simulated 429: honoring Retry-After and waiting {wait}s")
        time.sleep(wait)

    for attempt in range(MAX_RETRIES + 1):
        try:
            response = session.get(url, timeout=TIMEOUT_SECONDS)
        except requests.exceptions.RequestException as exc:
            if attempt == MAX_RETRIES:
                raise
            wait = bounded_backoff(attempt)
            metrics["retries"] += 1
            print(f"Network failure: {exc}. Retrying in {wait}s")
            time.sleep(wait)
            continue

        if response.status_code == 429:
            if attempt == MAX_RETRIES:
                response.raise_for_status()
            wait = retry_after_seconds(response.headers)
            metrics["retries"] += 1
            print(f"HTTP 429: honoring Retry-After and waiting {wait}s")
            time.sleep(wait)
            continue

        if 500 <= response.status_code < 600:
            if attempt == MAX_RETRIES:
                response.raise_for_status()
            wait = bounded_backoff(attempt)
            metrics["retries"] += 1
            print(f"HTTP {response.status_code}: retrying in {wait}s")
            time.sleep(wait)
            continue

        response.raise_for_status()
        return response

    raise RuntimeError("Request attempts were exhausted without a response.")


def rate_limit_wait(headers):
    required_headers = []
    for scope in RATE_LIMIT_SCOPES:
        required_headers.extend(
            [
                f"X-RateLimit-{scope}-Limit",
                f"X-RateLimit-{scope}-Remaining",
                f"X-RateLimit-{scope}-Reset",
            ]
        )

    missing = [name for name in required_headers if headers.get(name) is None]
    if missing:
        raise RuntimeError(
            "Missing required Metron rate-limit headers: " + ", ".join(missing)
        )

    wait = 0
    now = int(time.time())

    for scope in RATE_LIMIT_SCOPES:
        try:
            limit = int(headers[f"X-RateLimit-{scope}-Limit"])
            remaining = int(headers[f"X-RateLimit-{scope}-Remaining"])
            reset = int(headers[f"X-RateLimit-{scope}-Reset"])
        except ValueError as exc:
            raise RuntimeError("A Metron rate-limit header was not an integer.") from exc

        print(
            f"{scope} limit: {remaining}/{limit} remaining; "
            f"reset timestamp: {reset}"
        )

        if remaining <= 0:
            wait = max(wait, max(0, reset - now))

    return wait


def make_output_path(entity_type):
    data_directory = Path("data")
    data_directory.mkdir(parents=True, exist_ok=True)
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    return data_directory / f"metron_{entity_type}_{run_id}.jsonl"


def ingest(token, max_records, simulate_429):
    output_path = make_output_path(ENTITY_TYPE)
    metrics = {
        "pages": 0,
        "records": 0,
        "retries": 0,
        "simulation_used": False,
    }
    next_url = API_URL

    with requests.Session() as session:
        session.headers.update(
            {
                "Authorization": f"Bearer {token}",
                "Accept": "application/json",
                "User-Agent": "comicdb-platform-ingestor/1.0",
            }
        )

        with output_path.open("x", encoding="utf-8") as destination:
            while next_url and metrics["records"] < max_records:
                response = request_page(
                    session,
                    next_url,
                    metrics,
                    simulate_429,
                )
                page = response.json()
                results = page.get("results")

                if not isinstance(results, list):
                    raise RuntimeError("Metron response did not contain a results list.")

                remaining_slots = max_records - metrics["records"]
                ingested_at = datetime.now(timezone.utc).isoformat()

                for issue in results[:remaining_slots]:
                    envelope = {
                        "source": "metron",
                        "entity_type": ENTITY_TYPE,
                        "ingested_at": ingested_at,
                        "source_endpoint": response.url,
                        "payload": issue,
                    }
                    destination.write(
                        json.dumps(envelope, ensure_ascii=False) + "\n"
                    )

                metrics["pages"] += 1
                metrics["records"] += min(len(results), remaining_slots)
                next_url = page.get("next")

                print(
                    f"Page {metrics['pages']}: "
                    f"{metrics['records']}/{max_records} records written"
                )

                wait = rate_limit_wait(response.headers)
                print(f"Calculated adaptive wait: {wait}s")

                if next_url and metrics["records"] < max_records and wait > 0:
                    time.sleep(wait)

    return output_path, metrics


def main():
    args = parse_args()
    token = require_token()

    output_path, metrics = ingest(
        token=token,
        max_records=args.max_records,
        simulate_429=args.simulate_429,
    )

    print("\nRun complete")
    print(f"Pages fetched: {metrics['pages']}")
    print(f"Records written: {metrics['records']}")
    print(f"Retries: {metrics['retries']}")
    print(f"Output path: {output_path}")


if __name__ == "__main__":
    try:
        main()
    except (
        OSError,
        RuntimeError,
        ValueError,
        KeyError,
        requests.exceptions.RequestException,
    ) as exc:
        print(f"Error: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

Your ingestor can now paginate without overwhelming the API or retrying permanent failures. Next up, you will validate every JSONL record and turn the run metrics into a complete operational summary.

Validate and Report the Run

Your rate-aware Metron ingestor now follows sequential pages. It also handles rate limits without exposing your token.

Writing a file only proves that the client produced an artifact. A dependable data pipeline must prove that every line is parseable and complete.

This step adds line-by-line validation to the generated dataset. It also prints an operational summary that explains the completed run.

In this step, get ready to:
  • Define the required fields for every Bronze envelope.
  • Validate each JSON Lines record independently.
  • Report the final run metrics and inspect the generated dataset.
Define the Bronze validation contract

A validation contract lists the fields that make every source record useful to later pipeline stages. The validator uses this shared set to reject incomplete envelopes.

  • In ingest.py, find BACKOFF_CAP_SECONDS = 8.
  • Add the required envelope fields directly below that line by copying this block:
REQUIRED_ENVELOPE_FIELDS = {
    "source",
    "entity_type",
    "ingested_at",
    "source_endpoint",
    "payload",
}

What does this contract protect?

  • The source field identifies Metron as the system that supplied the record.
  • The entity_type field identifies the payload as an issue.
  • The ingested_at field records when the client received the issue.
  • The source_endpoint field preserves the page URL used for the request.
  • The payload field keeps the complete source object.
  • Save ingest.py.
  • Confirm the five field names appear directly below BACKOFF_CAP_SECONDS.

Seeing a syntax warning near the field set?

Check that every field name has matching quotation marks. Confirm the opening brace has a matching closing brace.

Compare each field name with the block above because the validator requires exact spelling.

Help me fix the required envelope field set in ingest.py.

Each JSONL line must stand on its own. The validator opens the finished file and checks one line at a time.

  • In ingest.py, locate def ingest(token, max_records, simulate_429):.
  • Add the validation function immediately above ingest() by copying this block:
def validate_jsonl(path):
    valid_lines = 0
    with path.open("r", encoding="utf-8") as source:
        for line_number, line in enumerate(source, start=1):
            if not line.strip():
                raise ValueError(f"Blank line found at line {line_number}")
            record = json.loads(line)
            if not isinstance(record, dict):
                raise ValueError(f"Line {line_number} is not a JSON object")
            missing = REQUIRED_ENVELOPE_FIELDS.difference(record)
            if missing:
                raise ValueError(f"Line {line_number} is missing fields: {sorted(missing)}")
            if not isinstance(record["payload"], dict):
                raise ValueError(f"Line {line_number} payload is not a JSON object")
            valid_lines += 1
    return valid_lines

What does this validator check?

  • The blank-line check rejects empty records before parsing begins.
  • The json.loads() call parses each line independently.
  • The dictionary check requires each parsed line to be a JSON object.
  • The set comparison finds any missing envelope fields.
  • The final type check confirms that payload still contains a complete JSON object.
  • The returned count records how many lines passed every check.
  • Save ingest.py.
  • Predict whether the existing ingestion flow still completes with the new validator defined.
  • Check that the existing ingestion flow still completes by running:
python ingest.py --max-records 150

What should you see?

You should see the existing page progress and rate-limit logs complete successfully. The printed output path identifies a new timestamped file under data/.

This run confirms that the new contract and function contain valid Python syntax. The next change connects validation to every completed run.

Does the ingestion stop before creating a file?

Check the indentation inside validate_jsonl(). Each check must remain inside the file-reading loop.

Confirm validate_jsonl() ends before ingest() begins.

Help me debug the validate_jsonl function.

Report validated run metrics

The run summary turns a generated file into an explainable pipeline result. It connects the validated line count with the records written during ingestion.

Elapsed time adds operational context. The unique output path makes the artifact easy to locate after each run.

  • In ingest.py, locate the main() function.
  • Replace the complete main() function with this version:
def main():
    args = parse_args()
    token = require_token()
    started_at = time.perf_counter()
    output_path, metrics = ingest(token=token, max_records=args.max_records, simulate_429=args.simulate_429)
    validated_lines = validate_jsonl(output_path)
    if validated_lines != metrics["records"]:
        raise RuntimeError("Validated line count did not match records written.")
    elapsed = time.perf_counter() - started_at
    print("\nRun complete")
    print(f"Pages fetched: {metrics['pages']}")
    print(f"Records written: {metrics['records']}")
    print(f"Retries: {metrics['retries']}")
    print(f"Validated lines: {validated_lines}")
    print(f"Elapsed seconds: {elapsed:.2f}")
    print(f"Output path: {output_path}")

How does the final report work?

  • The first time.perf_counter() call starts the run timer.
  • The validate_jsonl() call checks the completed output file.
  • The count comparison detects any mismatch between written records and valid lines.
  • The second timer reading calculates the total elapsed time.
  • The final print statements expose the metrics needed to explain the run.
  • Save ingest.py.
  • Predict which two summary counts should match after a successful bounded run.
  • Validate the dataset and report a bounded run by running:
python ingest.py --max-records 150

What should you see?

  • The page progress should show at least two pages fetched when the API has enough issues.
  • The Records written value should be 150.
  • The Retries value should report the retries used during this run.
  • The Validated lines value should be 150.
  • The Elapsed seconds value should report the complete run duration.
  • The Output path value should point to a new timestamped file under data/.
  • The terminal output should contain no API token or Authorization header.

That is the proof your pipeline needed. The client now validates its own output and reports the result of every run.

Did validation stop the run?

A blank-line error means the generated file contains an empty record. A missing-fields error identifies the exact line with an incomplete envelope.

A payload type error means the source record was stored in a different shape. Compare the named line with the envelope creation inside ingest().

Help me diagnose a JSONL validation failure.

Inspect the finished Bronze dataset

Independent line validation proves the file is machine-readable. A manual spot check confirms that each envelope still preserves a complete issue under payload.

  • Switch back to VS Code.
  • Use the file sidebar to select the JSONL file shown beside Output path.
  • Inspect the first line in the generated file.
  • Inspect a non-adjacent line later in the same file.
  • Confirm each inspected line contains source, entity_type, ingested_at, source_endpoint, and payload.
  • Confirm each payload contains a complete issue object.
  • Review the inspected lines for any API token or Authorization header.

What does this inspection prove?

Each line is an independently parseable Bronze envelope with ingestion metadata around the source issue. The source payload remains available for later pipeline stages.

The timestamped filename separates each run from earlier datasets. This prevents silent record accumulation across reruns.

Use these tabs to confirm your final ingest.py file.

✔️ Awesome, I've got everything!

Your file is saved. Your final run now validates every line and prints the complete operational summary.

ⓧ I'd like to double check the full code

import argparse
import json
import os
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

import requests

API_URL = "https://metron.cloud/api/issue/"
ENTITY_TYPE = "issue"
RATE_LIMIT_SCOPES = ("Sustained", "Burst")
TIMEOUT_SECONDS = 30
MAX_RETRIES = 3
BACKOFF_CAP_SECONDS = 8
REQUIRED_ENVELOPE_FIELDS = {
    "source",
    "entity_type",
    "ingested_at",
    "source_endpoint",
    "payload",
}


def parse_args():
    parser = argparse.ArgumentParser(description="Ingest a bounded set of Metron issues into Bronze JSONL.")
    parser.add_argument("--max-records", type=int, default=150, help="maximum number of records to write")
    parser.add_argument("--simulate-429", action="store_true", help="simulate one Retry-After response before the first live request")
    args = parser.parse_args()
    if args.max_records < 1:
        parser.error("--max-records must be at least 1")
    return args


def require_token():
    token = os.getenv("METRON_TOKEN")
    if not token:
        raise RuntimeError("METRON_TOKEN is not set. Export it in the current terminal session.")
    return token


def retry_after_seconds(headers):
    value = headers.get("Retry-After")
    if value is None:
        raise RuntimeError("A 429 response did not include Retry-After.")
    try:
        return max(0, int(value))
    except ValueError as exc:
        raise RuntimeError("Retry-After was not an integer number of seconds.") from exc


def bounded_backoff(attempt):
    return min(2**attempt, BACKOFF_CAP_SECONDS)


def request_page(session, url, metrics, simulate_429):
    if simulate_429 and not metrics["simulation_used"]:
        wait = retry_after_seconds({"Retry-After": "1"})
        metrics["simulation_used"] = True
        metrics["retries"] += 1
        print(f"Simulated 429: honoring Retry-After and waiting {wait}s")
        time.sleep(wait)
    for attempt in range(MAX_RETRIES + 1):
        try:
            response = session.get(url, timeout=TIMEOUT_SECONDS)
        except requests.exceptions.RequestException as exc:
            if attempt == MAX_RETRIES:
                raise
            wait = bounded_backoff(attempt)
            metrics["retries"] += 1
            print(f"Network failure: {exc}. Retrying in {wait}s")
            time.sleep(wait)
            continue
        if response.status_code == 429:
            if attempt == MAX_RETRIES:
                response.raise_for_status()
            wait = retry_after_seconds(response.headers)
            metrics["retries"] += 1
            print(f"HTTP 429: honoring Retry-After and waiting {wait}s")
            time.sleep(wait)
            continue
        if 500 <= response.status_code < 600:
            if attempt == MAX_RETRIES:
                response.raise_for_status()
            wait = bounded_backoff(attempt)
            metrics["retries"] += 1
            print(f"HTTP {response.status_code}: retrying in {wait}s")
            time.sleep(wait)
            continue
        response.raise_for_status()
        return response
    raise RuntimeError("Request attempts were exhausted without a response.")


def rate_limit_wait(headers):
    required_headers = []
    for scope in RATE_LIMIT_SCOPES:
        required_headers.extend([f"X-RateLimit-{scope}-Limit", f"X-RateLimit-{scope}-Remaining", f"X-RateLimit-{scope}-Reset"])
    missing = [name for name in required_headers if headers.get(name) is None]
    if missing:
        raise RuntimeError("Missing required Metron rate-limit headers: " + ", ".join(missing))
    wait = 0
    now = int(time.time())
    for scope in RATE_LIMIT_SCOPES:
        try:
            limit = int(headers[f"X-RateLimit-{scope}-Limit"])
            remaining = int(headers[f"X-RateLimit-{scope}-Remaining"])
            reset = int(headers[f"X-RateLimit-{scope}-Reset"])
        except ValueError as exc:
            raise RuntimeError("A Metron rate-limit header was not an integer.") from exc
        print(f"{scope} limit: {remaining}/{limit} remaining; reset timestamp: {reset}")
        if remaining <= 0:
            wait = max(wait, max(0, reset - now))
    return wait


def make_output_path(entity_type):
    data_directory = Path("data")
    data_directory.mkdir(parents=True, exist_ok=True)
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    return data_directory / f"metron_{entity_type}_{run_id}.jsonl"


def validate_jsonl(path):
    valid_lines = 0
    with path.open("r", encoding="utf-8") as source:
        for line_number, line in enumerate(source, start=1):
            if not line.strip():
                raise ValueError(f"Blank line found at line {line_number}")
            record = json.loads(line)
            if not isinstance(record, dict):
                raise ValueError(f"Line {line_number} is not a JSON object")
            missing = REQUIRED_ENVELOPE_FIELDS.difference(record)
            if missing:
                raise ValueError(f"Line {line_number} is missing fields: {sorted(missing)}")
            if not isinstance(record["payload"], dict):
                raise ValueError(f"Line {line_number} payload is not a JSON object")
            valid_lines += 1
    return valid_lines


def ingest(token, max_records, simulate_429):
    output_path = make_output_path(ENTITY_TYPE)
    metrics = {"pages": 0, "records": 0, "retries": 0, "simulation_used": False}
    next_url = API_URL
    with requests.Session() as session:
        session.headers.update({"Authorization": f"Bearer {token}", "Accept": "application/json", "User-Agent": "comicdb-platform-ingestor/1.0"})
        with output_path.open("x", encoding="utf-8") as destination:
            while next_url and metrics["records"] < max_records:
                response = request_page(session, next_url, metrics, simulate_429)
                page = response.json()
                results = page.get("results")
                if not isinstance(results, list):
                    raise RuntimeError("Metron response did not contain a results list.")
                remaining_slots = max_records - metrics["records"]
                ingested_at = datetime.now(timezone.utc).isoformat()
                for issue in results[:remaining_slots]:
                    envelope = {"source": "metron", "entity_type": ENTITY_TYPE, "ingested_at": ingested_at, "source_endpoint": response.url, "payload": issue}
                    destination.write(json.dumps(envelope, ensure_ascii=False) + "\n")
                metrics["pages"] += 1
                metrics["records"] += min(len(results), remaining_slots)
                next_url = page.get("next")
                print(f"Page {metrics['pages']}: {metrics['records']}/{max_records} records written")
                wait = rate_limit_wait(response.headers)
                print(f"Calculated adaptive wait: {wait}s")
                if next_url and metrics["records"] < max_records and wait > 0:
                    time.sleep(wait)
    return output_path, metrics


def main():
    args = parse_args()
    token = require_token()
    started_at = time.perf_counter()
    output_path, metrics = ingest(token=token, max_records=args.max_records, simulate_429=args.simulate_429)
    validated_lines = validate_jsonl(output_path)
    if validated_lines != metrics["records"]:
        raise RuntimeError("Validated line count did not match records written.")
    elapsed = time.perf_counter() - started_at
    print("\nRun complete")
    print(f"Pages fetched: {metrics['pages']}")
    print(f"Records written: {metrics['records']}")
    print(f"Retries: {metrics['retries']}")
    print(f"Validated lines: {validated_lines}")
    print(f"Elapsed seconds: {elapsed:.2f}")
    print(f"Output path: {output_path}")


if __name__ == "__main__":
    try:
        main()
    except (OSError, RuntimeError, ValueError, KeyError, requests.exceptions.RequestException) as exc:
        print(f"Error: {exc}", file=sys.stderr)
        raise SystemExit(1) from exc

Secret mission

Generalize the Client for Characters

Your issues ingestor already handles pagination, rate limits, retries, validation, and run reporting. Generalize it so one command can ingest either issues or characters without duplicating the reliability logic.

Clean Up Your Resources

Clean Up Your Resources

This part adds no ongoing cloud costs because the ingestor runs on your Mac with your existing Metron account. Choose the cleanup option that fits whether you are continuing the series.

Resources you used:

  • The local Git repository currently open in VS Code.
  • The .venv environment with Requests 2.34.2 installed.
  • The timestamped issue Bronze JSONL files under data/.
  • The timestamped character JSONL files under data/.
  • The session-only METRON_TOKEN value in the current zsh shell.

Keep everything running

No action is needed. Choose this if you plan to continue into Part 2.

  • Keep the repository available in VS Code for the next project.
  • Retain .venv so the pinned Requests dependency stays isolated.
  • Retain the Bronze datasets under data/ as sample ingestion artifacts.
  • Leave the terminal window open only while you are actively using the session-only token.

Pause - I'll come back to this later

Stop the active local session while leaving every project file available for later.

  • Switch back to the terminal from earlier.
  • Deactivate the active .venv by running this command:
deactivate

What does this command do?

The deactivate command returns this shell to your system Python. The repository remains unchanged.

  • Close the terminal window from earlier.

The current zsh session ends. METRON_TOKEN is unavailable in a new shell.

Delete - I don't want to use this again

Remove the local project resources if you do not plan to continue the ComicDB Platform series.

  • Switch back to the terminal from earlier.
  • Deactivate the active .venv by running this command:
deactivate

What does this command do?

The deactivate command stops this shell from using packages inside .venv.

  • Close the terminal window from earlier.
  • Close the VS Code window from earlier.
  • Use Finder to delete the repository folder that was open in VS Code.
  • Open your Metron account page.
  • Select API Tokens.
  • Revoke the dedicated token if you do not plan to continue the series.

What gets removed?

Deleting the repository removes its source files plus Git history. The nested .venv disappears with it.

Every Bronze dataset under data/ also disappears. Closing the terminal ends the current zsh session.

Revoking the dedicated token blocks future API access with that credential.

Nice Work!

Nice Work!

Mission accomplished! Your reusable Metron client now retrieves issues or characters into validated run-specific files. This completes Part 1 of the nine-part ComicDB Platform series.

What you learned:

  • Built a client with bearer token authentication that follows the API's ready-to-use next URLs to a precise record boundary.
  • Preserved unchanged source payloads inside validated Bronze envelopes. Used JSON Lines so every record remains independently parseable.
  • Protected live requests with dual-window rate limiting across burst counters. The same protection covers sustained counters. Your client honors Retry-After for 429 responses. It applies bounded exponential backoff to network failures. It applies the same backoff to 5xx responses.
  • Secret Mission: Generalized the same ingestion engine to characters without duplicating its pagination or reliability logic.

Ready to quiz yourself?