Capstone: a bounded, complete API report

pythonapi-clientsautomationindependent-practice

Saved on this device

Full lesson · 4 learning steps Open any exercise for focused practice

Step 1 of 4 · intro card

Capstone: a bounded, complete API report

Concept and contract

Compose controlled HTTP, explicit pagination and source identity into one country-scoped report that refuses incomplete evidence.

This capstone connects the earlier boundaries into a small fictional API client. It requests a declared country from a caller-authorized base URL, reads bounded JSON pages and publishes a result only after explicit completion.

The lab supplies its own loopback server and invented companies. The function is a learning client for one declared page contract, not a production ApplyDjinn importer or a general-purpose crawler for arbitrary untrusted endpoints.

Mechanism and worked example

The client owns one opener, a visited-cursor set and a mapping keyed by normalized source identifier.

The client owns one opener, a visited-cursor set and a mapping keyed by normalized source identifier. Every received record is validated before the country view is selected.

Equal repeated records count once; a conflicting record with the same identifier fails even if one version would later be filtered out. After the traversal completes, identifiers and distinct company counts are calculated from the same active country population, preventing mismatched denominators.

Complete worked example

python

from urllib.request import build_opener, HTTPRedirectHandler, ProxyHandler

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, request, file, code, message, headers, new_url):
        return None

import json
from urllib.parse import urlencode

def invalid_constant(value):
    raise ValueError("non-finite JSON constant")

def client_report(base_url, country, max_pages=3):
    if not isinstance(country, str) or country not in {"US", "IL", "UA", "CA"}:
        raise ValueError("supported country required")
    if type(max_pages) is not int or not 1 <= max_pages <= 5:
        raise ValueError("bounded page count required")
    opener = build_opener(ProxyHandler({}), NoRedirect())
    cursor, visited, records = None, set(), {}
    for _ in range(max_pages):
        if cursor in visited:
            raise ValueError("cursor cycle")
        visited.add(cursor)
        query = {"country": country}
        if cursor is not None:
            query["cursor"] = cursor
        with opener.open(base_url + "/jobs?" + urlencode(query), timeout=2) as response:
            if response.headers.get_content_type() != "application/json":
                raise ValueError("JSON media type required")
            raw = response.read(4097)
        if len(raw) > 4096:
            raise ValueError("response exceeds budget")
        page = json.loads(raw.decode("utf-8"), parse_constant=invalid_constant)
        if not isinstance(page, dict) or "next" not in page or not isinstance(page.get("items"), list):
            raise ValueError("page shape required")
        for row in page["items"]:
            if not isinstance(row, dict) or type(row.get("active")) is not bool:
                raise ValueError("role object and boolean state required")
            if any(not isinstance(row.get(key), str) or not 1 <= len(row[key].strip()) <= 80 for key in ("id", "company", "country")):
                raise ValueError("bounded text fields required")
            normalized = {key: row[key].strip() for key in ("id", "company", "country")}
            normalized["country"] = normalized["country"].upper()
            if len(normalized["country"]) != 2 or not normalized["country"].isascii() or not normalized["country"].isalpha():
                raise ValueError("country code required")
            normalized["active"] = row["active"]
            previous = records.get(normalized["id"])
            if previous is not None and previous != normalized:
                raise ValueError("conflicting identity")
            records[normalized["id"]] = normalized
        cursor = page["next"]
        if cursor is None:
            break
        if not isinstance(cursor, str) or not cursor:
            raise ValueError("invalid next cursor")
    else:
        raise ValueError("incomplete collection")
    selected = [row for row in records.values() if row["active"] and row["country"] == country]
    return {"ids": sorted(row["id"] for row in selected), "companies": len({row["company"] for row in selected})}

Reasoning: Compose controlled HTTP, explicit pagination and source identity into one country-scoped report that refuses incomplete evidence. Compare the stated values and failure boundary before changing the implementation.

Counterexamples and common mistakes

The counterexample lets the last repeated identity overwrite the earlier one.

The counterexample lets the last repeated identity overwrite the earlier one. That hides inconsistent source evidence and can silently change which country receives a record.

Returning accumulated rows after exhausting the page budget is another completeness defect. HTTP overload errors propagate instead of triggering an unlimited retry loop inside the client. A caller can apply the earlier bounded retry policy while preserving the exact cursor and avoiding repeated publication of partial work.

Counterexample: a runnable policy defect

python

from urllib.request import build_opener, HTTPRedirectHandler, ProxyHandler

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, request, file, code, message, headers, new_url):
        return None

import json
from urllib.parse import urlencode

def invalid_constant(value):
    raise ValueError("non-finite JSON constant")

def client_report(base_url, country, max_pages=3):
    if not isinstance(country, str) or country not in {"US", "IL", "UA", "CA"}:
        raise ValueError("supported country required")
    if type(max_pages) is not int or not 1 <= max_pages <= 5:
        raise ValueError("bounded page count required")
    opener = build_opener(ProxyHandler({}), NoRedirect())
    cursor, visited, records = None, set(), {}
    for _ in range(max_pages):
        if cursor in visited:
            raise ValueError("cursor cycle")
        visited.add(cursor)
        query = {"country": country}
        if cursor is not None:
            query["cursor"] = cursor
        with opener.open(base_url + "/jobs?" + urlencode(query), timeout=2) as response:
            if response.headers.get_content_type() != "application/json":
                raise ValueError("JSON media type required")
            raw = response.read(4097)
        if len(raw) > 4096:
            raise ValueError("response exceeds budget")
        page = json.loads(raw.decode("utf-8"), parse_constant=invalid_constant)
        if not isinstance(page, dict) or "next" not in page or not isinstance(page.get("items"), list):
            raise ValueError("page shape required")
        for row in page["items"]:
            if not isinstance(row, dict) or type(row.get("active")) is not bool:
                raise ValueError("role object and boolean state required")
            if any(not isinstance(row.get(key), str) or not 1 <= len(row[key].strip()) <= 80 for key in ("id", "company", "country")):
                raise ValueError("bounded text fields required")
            normalized = {key: row[key].strip() for key in ("id", "company", "country")}
            normalized["country"] = normalized["country"].upper()
            if len(normalized["country"]) != 2 or not normalized["country"].isascii() or not normalized["country"].isalpha():
                raise ValueError("country code required")
            normalized["active"] = row["active"]
            previous = records.get(normalized["id"])
            records[normalized["id"]] = normalized
        cursor = page["next"]
        if cursor is None:
            break
        if not isinstance(cursor, str) or not cursor:
            raise ValueError("invalid next cursor")
    else:
        raise ValueError("incomplete collection")
    selected = [row for row in records.values() if row["active"] and row["country"] == country]
    return {"ids": sorted(row["id"] for row in selected), "companies": len({row["company"] for row in selected})}

Reasoning: This runnable counterexample violates the declared contract; the companion lab requires it to fail.

Independent practice: API client task

Add a by_company mapping whose values count distinct selected role identities per company.

Add a by_company mapping whose values count distinct selected role identities per company. Keep the existing sorted identifier list and distinct company count consistent with that mapping.

Start with two selected roles at one company and a role in another country. Then include an equal repeated source record. Reuse the already deduplicated selected population rather than counting raw page appearances a second time.

Step 1: Count selected distinct role identities per company. Step 2: Keep company counts and identifier counts on the same deduplicated population.

Practice starter: solve before reading the solution

python

from urllib.request import build_opener, HTTPRedirectHandler, ProxyHandler

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, request, file, code, message, headers, new_url):
        return None

import json
from urllib.parse import urlencode

def invalid_constant(value):
    raise ValueError("non-finite JSON constant")

def client_report(base_url, country, max_pages=3):
    # Remove this placeholder, then adapt the retained baseline:
    # Count selected distinct role identities per company.
    raise NotImplementedError
    if not isinstance(country, str) or country not in {"US", "IL", "UA", "CA"}:
        raise ValueError("supported country required")
    if type(max_pages) is not int or not 1 <= max_pages <= 5:
        raise ValueError("bounded page count required")
    opener = build_opener(ProxyHandler({}), NoRedirect())
    cursor, visited, records = None, set(), {}
    for _ in range(max_pages):
        if cursor in visited:
            raise ValueError("cursor cycle")
        visited.add(cursor)
        query = {"country": country}
        if cursor is not None:
            query["cursor"] = cursor
        with opener.open(base_url + "/jobs?" + urlencode(query), timeout=2) as response:
            if response.headers.get_content_type() != "application/json":
                raise ValueError("JSON media type required")
            raw = response.read(4097)
        if len(raw) > 4096:
            raise ValueError("response exceeds budget")
        page = json.loads(raw.decode("utf-8"), parse_constant=invalid_constant)
        if not isinstance(page, dict) or "next" not in page or not isinstance(page.get("items"), list):
            raise ValueError("page shape required")
        for row in page["items"]:
            if not isinstance(row, dict) or type(row.get("active")) is not bool:
                raise ValueError("role object and boolean state required")
            if any(not isinstance(row.get(key), str) or not 1 <= len(row[key].strip()) <= 80 for key in ("id", "company", "country")):
                raise ValueError("bounded text fields required")
            normalized = {key: row[key].strip() for key in ("id", "company", "country")}
            normalized["country"] = normalized["country"].upper()
            if len(normalized["country"]) != 2 or not normalized["country"].isascii() or not normalized["country"].isalpha():
                raise ValueError("country code required")
            normalized["active"] = row["active"]
            previous = records.get(normalized["id"])
            if previous is not None and previous != normalized:
                raise ValueError("conflicting identity")
            records[normalized["id"]] = normalized
        cursor = page["next"]
        if cursor is None:
            break
        if not isinstance(cursor, str) or not cursor:
            raise ValueError("invalid next cursor")
    else:
        raise ValueError("incomplete collection")
    selected = [row for row in records.values() if row["active"] and row["country"] == country]
    return {"ids": sorted(row["id"] for row in selected), "companies": len({row["company"] for row in selected})}

Reasoning: Count selected distinct role identities per company. Keep company counts and identifier counts on the same deduplicated population.

Hint 1: Build the mapping from selected, after successful traversal.
Hint 2: Do not count repeated page appearances or records from another country.

Solution, tests and trade-offs

The fixture checks two pages, repeated identities, country filtering, conflicting identities, a cursor cycle, page-budget exhaustion and an oversized response.

The fixture checks two pages, repeated identities, country filtering, conflicting identities, a cursor cycle, page-budget exhaustion and an oversized response. It records requested paths so cycle detection and the page cap cannot silently overfetch.

Practice adds exact per-company counts. The output stays in memory; the earlier atomic-file or SQLite adapter should receive it only after successful completion, under its own publication boundary and trusted workspace policy.

Practice solution and acceptance checks

python

from urllib.request import build_opener, HTTPRedirectHandler, ProxyHandler

class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, request, file, code, message, headers, new_url):
        return None

import json
from urllib.parse import urlencode

def invalid_constant(value):
    raise ValueError("non-finite JSON constant")

def client_report(base_url, country, max_pages=3):
    if not isinstance(country, str) or country not in {"US", "IL", "UA", "CA"}:
        raise ValueError("supported country required")
    if type(max_pages) is not int or not 1 <= max_pages <= 5:
        raise ValueError("bounded page count required")
    opener = build_opener(ProxyHandler({}), NoRedirect())
    cursor, visited, records = None, set(), {}
    for _ in range(max_pages):
        if cursor in visited:
            raise ValueError("cursor cycle")
        visited.add(cursor)
        query = {"country": country}
        if cursor is not None:
            query["cursor"] = cursor
        with opener.open(base_url + "/jobs?" + urlencode(query), timeout=2) as response:
            if response.headers.get_content_type() != "application/json":
                raise ValueError("JSON media type required")
            raw = response.read(4097)
        if len(raw) > 4096:
            raise ValueError("response exceeds budget")
        page = json.loads(raw.decode("utf-8"), parse_constant=invalid_constant)
        if not isinstance(page, dict) or "next" not in page or not isinstance(page.get("items"), list):
            raise ValueError("page shape required")
        for row in page["items"]:
            if not isinstance(row, dict) or type(row.get("active")) is not bool:
                raise ValueError("role object and boolean state required")
            if any(not isinstance(row.get(key), str) or not 1 <= len(row[key].strip()) <= 80 for key in ("id", "company", "country")):
                raise ValueError("bounded text fields required")
            normalized = {key: row[key].strip() for key in ("id", "company", "country")}
            normalized["country"] = normalized["country"].upper()
            if len(normalized["country"]) != 2 or not normalized["country"].isascii() or not normalized["country"].isalpha():
                raise ValueError("country code required")
            normalized["active"] = row["active"]
            previous = records.get(normalized["id"])
            if previous is not None and previous != normalized:
                raise ValueError("conflicting identity")
            records[normalized["id"]] = normalized
        cursor = page["next"]
        if cursor is None:
            break
        if not isinstance(cursor, str) or not cursor:
            raise ValueError("invalid next cursor")
    else:
        raise ValueError("incomplete collection")
    selected = [row for row in records.values() if row["active"] and row["country"] == country]
    by_company = {}
    for row in selected:
        by_company[row["company"]] = by_company.get(row["company"], 0) + 1
    return {"ids": sorted(row["id"] for row in selected), "companies": len(by_company), "by_company": by_company}

Reasoning: Build the mapping from selected, after successful traversal. Do not count repeated page appearances or records from another country.

Interview reasoning and next step

Explain the client’s evidence limits before calling the report complete.

Explain the client’s evidence limits before calling the report complete. Completion means the owned page contract reached its explicit end within the declared bounds; it does not prove global market coverage or unrestricted source rights.

Production work also needs source-specific authorization, freshness, backoff, checkpointing and cross-source identity rules. The optional extension composes cached reads and a durable publication adapter while retaining these separate responsibilities and preserving old data on failure.

Optional stretch — after the core task works: Compose cached reads and one durable publication adapter without publishing partial results.

Computer practice: https://applydjinn.com/typorithm/python-automation-lab-20261105.zip. Run verify.py using your selected Python interpreter, then adapt the retained exercise baseline and check it with --practice followed by the lesson id. The standard-library lab uses owned temporary 127.0.0.1 loopback servers, fictional pages and disposable files; it never scans production feeds or opens the working ApplyDjinn database.

Compose controlled HTTP, explicit pagination and source identity into one country-scoped report that refuses incomplete evidence.