Use the API first: JSON, keys and quotas

api/api_first.py

"""All 240 products from the shop's API and from its HTML pages: requests, bytes, and what one product looks like."""
import json
import os
import urllib.request
from pathlib import Path
from urllib.parse import urlencode

import nimbu_lab as lab

for line in Path("lab_api.env").read_text(encoding="utf-8").splitlines():  # in the lab only: the shell sets it
    if line and not line.startswith("#"):
        os.environ.setdefault(*line.split("=", 1))
HEADERS = {"User-Agent": "NimbuLearner/1.0 (+https://example.org/my-bot; you@example.org)",
           "X-Api-Key": os.environ["NIMBU_API_KEY"], "Accept": "application/json"}
shop = lab.opener()


def next_link(response):
    link = response.headers.get("Link", "")
    return link[1:link.index(">")] if 'rel="next"' in link else None


url, items = f"{lab.SITE}/api/v2/products?{urlencode({'page': 1, 'per_page': 100})}", []
before = len(lab.LOG)
received = 0
while url:
    with shop.open(urllib.request.Request(url, headers=HEADERS), timeout=10) as response:
        body = response.read()
        received += len(body)
        items += json.loads(body)["items"]
        url = next_link(response)
print(f"API:  {len(lab.LOG) - before:2} requests, {received:6,} bytes, {len(items)} products")

before, received, url = len(lab.LOG), 0, lab.SITE + "/products/?page=1"
while url:
    with shop.open(urllib.request.Request(url, headers=HEADERS), timeout=10) as response:
        received += len(response.read())
        url = lab.SITE + next_link(response) if next_link(response) else None
print(f"HTML: {len(lab.LOG) - before:2} requests, {received:6,} bytes, the same products as text in table cells")
print()
print("one product from the API:", items[0])

API:   3 requests, 25,244 bytes, 240 products
HTML: 10 requests, 20,907 bytes, the same products as text in table cells

one product from the API: {'sku': 'NB-0001', 'name': 'Basmati rice 250 g', 'price_paise': 2500, 'currency': 'INR', 'in_stock': True}

nimbu_lab.py (the shop) · api/nimbu_lab.py

"""Nimbu Bazaar, the practice shop of this track. Every product, price and person in it is invented.

The shop is a WSGI application (PEP 3333): app() answers one request. opener() builds Python's own HTTP client,
urllib.request, with its usual handlers for redirects, errors and cookies, and one change: the last step hands each
request to app() instead of opening a socket. The client code is real, nothing leaves your program, and any host but
the shop's gets 421 Misdirected Request.
"""
import fnmatch
import http.client
import http.cookies
import io
import json
import math
import socket
import urllib.request
import urllib.response
from urllib.parse import parse_qs, urlsplit

HOST = "bazaar.nimbu.test"
SITE = f"https://{HOST}"
KINDS = ["Basmati rice", "Toor dal", "Green tea", "Jaggery", "Turmeric", "Mustard oil", "Poha", "Rock salt",
         "Cashews", "Cardamom", "Ragi flour", "Ghee", "Honey", "Peanuts", "Coffee", "Raisins", "Sesame seeds",
         "Tamarind", "Cumin", "Black pepper", "Rolled oats", "Cloves", "Fennel", "Moong dal", "Chana dal",
         "Wheat flour", "Semolina", "Coriander", "Red chilli", "Almonds", "Dates", "Coconut oil", "Brown rice",
         "Millet", "Flax seeds", "Cinnamon", "Ginger powder", "Vermicelli", "Masoor dal", "Sago"]
SIZES = [250, 500, 1000, 2000, 5000, 10000]  # grams
LOG = []  # what the shop saw: one entry per request


class Clock:
    """The shop's clock, in seconds. A slow page moves it on instead of waiting, so every example runs at once."""

    def __init__(self):
        self.seconds = 0.0

    def now(self):
        return self.seconds

    def sleep(self, seconds):
        self.seconds += seconds


clock = Clock()


def products():
    """The 240 products: 40 kinds in 6 pack sizes, each price in paise (100 paise = 1 rupee)."""
    out = []
    for k, kind in enumerate(KINDS):
        per_kg = 60 + (k * 37) % 900  # rupees a kilogram
        for grams in SIZES:
            size = f"{grams // 1000} kg" if grams >= 1000 else f"{grams} g"
            out.append({"sku": f"NB-{len(out) + 1:04d}", "name": f"{kind} {size}",
                        "paise": per_kg * grams // 10 + 1000})
    return out


def rupees(paise):
    return f"₹{paise // 100:,}.{paise % 100:02d}"


class Asked:
    """One request as the shop's pages see it."""

    def __init__(self, environ):
        self.method, self.path = environ["REQUEST_METHOD"], environ["PATH_INFO"]
        self.query = {k: v[-1] for k, v in parse_qs(environ.get("QUERY_STRING", "")).items()}
        self.headers = {k[5:].replace("_", "-").title(): v for k, v in environ.items() if k.startswith("HTTP_")}
        size = int(environ.get("CONTENT_LENGTH") or 0)
        self.form = {k: v[-1] for k, v in parse_qs(environ["wsgi.input"].read(size).decode()).items()}
        self.cookies = {k: m.value for k, m in http.cookies.SimpleCookie(environ.get("HTTP_COOKIE", "")).items()}
        self.wait = environ.get("lab.wait", clock.sleep)  # the shop takes this long before its next bytes


PAGES = {}  # path pattern ("/products/NB-*/" matches every product) -> function(asked) -> (status, headers, body)


def page(pattern):
    def register(fn):
        PAGES[pattern] = fn
        return fn
    return register


def find_page(path):
    exact = PAGES.get(path)
    return exact or next((fn for p, fn in PAGES.items() if "*" in p and fnmatch.fnmatchcase(path, p)), None)


def text(status, body, kind="text/plain; charset=utf-8", extra=()):
    data = body.encode() if isinstance(body, str) else body
    return status, [("Content-Type", kind), ("Content-Length", str(len(data))), *extra], data


def not_found(asked):
    return text("404 Not Found", "no such page\n")


def app(environ, start_response):
    """The shop: answers one request, and writes it into LOG first."""
    asked = Asked(environ)
    if environ.get("HTTP_HOST") != HOST:
        status, headers, body = text("421 Misdirected Request", "not a Nimbu Bazaar address\n")
    else:
        status, headers, body = (find_page(asked.path) or not_found)(asked)
    target = asked.path + ("?" + environ["QUERY_STRING"] if environ.get("QUERY_STRING") else "")
    LOG.append({"line": f"{asked.method} {target} HTTP/1.1", "headers": environ.get("lab.sent", []),
                "status": int(status[:3])})
    start_response(status, headers)
    return [body] if isinstance(body, bytes) else body


class Stream(io.RawIOBase):
    """A response body read chunk by chunk, as from a socket: a slow chunk can time out in the middle of read()."""

    def __init__(self, chunks):
        self.chunks, self.rest = iter(chunks), b""

    def readable(self):
        return True

    def readinto(self, buffer):
        while not self.rest:
            self.rest = next(self.chunks, None)
            if self.rest is None:
                self.rest = b""
                return 0
        n = min(len(buffer), len(self.rest))
        buffer[:n], self.rest = self.rest[:n], self.rest[n:]
        return n


class ShopTransport(urllib.request.AbstractHTTPHandler):
    """urllib's last step, with app() where the socket would be. Everything before it is urllib's own code."""

    handler_order = 100  # runs before any handler that would open a real connection
    http_request = https_request = urllib.request.AbstractHTTPHandler.do_request_

    def https_open(self, req):
        # The headers urllib's do_open() writes to the socket, and the one http.client adds by itself.
        headers = dict(req.unredirected_hdrs)
        headers.update({k: v for k, v in req.headers.items() if k not in headers})
        headers["Connection"] = "close"
        sent = [(k.title(), v) for k, v in headers.items()]
        if "Accept-encoding" not in headers:
            sent.insert(0, ("Accept-Encoding", "identity"))
        timeout = socket.getdefaulttimeout() if req.timeout is socket._GLOBAL_DEFAULT_TIMEOUT else req.timeout

        def wait(seconds):
            """A socket in timeout mode gives up when one wait for bytes lasts longer than the timeout."""
            if timeout is not None and seconds > timeout:
                clock.sleep(timeout)
                raise TimeoutError("timed out")
            clock.sleep(seconds)

        parts, data = urlsplit(req.full_url), req.data or b""
        environ = {"REQUEST_METHOD": req.get_method(), "PATH_INFO": parts.path or "/", "QUERY_STRING": parts.query,
                   "SERVER_NAME": parts.hostname or "", "SERVER_PORT": str(parts.port or 443),
                   "SERVER_PROTOCOL": "HTTP/1.1", "wsgi.url_scheme": parts.scheme, "wsgi.input": io.BytesIO(data),
                   "CONTENT_TYPE": headers.get("Content-type", ""), "CONTENT_LENGTH": str(len(data)),
                   "lab.wait": wait, "lab.sent": sent}
        environ.update({"HTTP_" + k.upper().replace("-", "_"): v for k, v in sent
                        if k not in ("Content-Type", "Content-Length")})
        answer = {}
        body = app(environ, lambda status, hdrs, exc_info=None: answer.update(status=status, headers=hdrs))
        raw = "".join(f"{k}: {v}\r\n" for k, v in answer["headers"]) + "\r\n"
        message = http.client.parse_headers(io.BytesIO(raw.encode("iso-8859-1")))
        response = urllib.response.addinfourl(io.BufferedReader(Stream(body)), message, req.full_url,
                                              int(answer["status"][:3]))
        response.reason = response.msg = answer["status"][4:]  # as urllib's own responses carry them
        return response

    http_open = https_open


DEFAULTS = [urllib.request.HTTPDefaultErrorHandler, urllib.request.HTTPRedirectHandler,
            urllib.request.HTTPErrorProcessor, urllib.request.UnknownHandler]


def opener(*handlers):
    """Like urllib.request.build_opener(*handlers), with the shop in place of the network: urllib's usual handlers,
    except those a handler you pass replaces (a subclass of HTTPRedirectHandler, say)."""
    director = urllib.request.OpenerDirector()
    kept = [d for d in DEFAULTS if not any(isinstance(h, d) for h in handlers)]
    for handler in [ShopTransport(), *(d() for d in kept), *handlers]:
        director.add_handler(handler)
    return director


# ---- The pages this lesson's examples ask for ----------------------------------------------------------------------
KEYS = {"nb_test_5f1c9a7e2d4b8c60": "learner@example.org"}  # the lab's one test key: invented, and worthless elsewhere
HOURLY = 100  # requests a key may make in each hour of the shop's clock
BURST = 5  # requests a key may make within one second of the shop's clock
USAGE = {}  # key -> {"window": hour number, "used": requests this hour, "recent": times of the last requests}
SCHEMA = {"version": "2"}  # an example may set "3" to see what an API change does to a collector


def listing(n):
    """Page n of the catalogue as HTML, as a browser would get it (24 products a page)."""
    rows = "".join(f'<tr><td><a href="/products/{p["sku"]}/">{p["name"]}</a></td><td>{rupees(p["paise"])}</td></tr>\n'
                   for p in products()[(n - 1) * 24:n * 24])
    return f"<!doctype html>\n<title>Products, page {n}</title>\n<table>\n{rows}</table>\n"


@page("/products/")
def catalogue(asked):
    n = int(asked.query.get("page", "1"))
    link = [("Link", f'</products/?page={n + 1}>; rel="next"')] if n < 10 else []
    return text("200 OK", listing(n), "text/html; charset=utf-8", link)


def item(p):
    if SCHEMA["version"] == "3":  # the shop's new release: a renamed field, and the price as text in rupees
        return {"sku": p["sku"], "name": p["name"], "price": f'{p["paise"] / 100:.2f}', "currency": "INR"}
    return {"sku": p["sku"], "name": p["name"], "price_paise": p["paise"], "currency": "INR", "in_stock": True}


def problem(status, detail, extra=()):
    return text(status, json.dumps({"error": detail}), "application/json", extra)


@page("/api/v2/products")
def api_products(asked):
    key = asked.headers.get("X-Api-Key") or asked.query.get("api_key")
    if not key:
        return problem("401 Unauthorized", "send your API key in the X-Api-Key header")
    if key not in KEYS:
        return problem("401 Unauthorized", "this API key is not valid")
    now = clock.now()
    use = USAGE.setdefault(key, {"window": -1, "used": 0, "recent": []})
    window = int(now // 3600)
    if window != use["window"]:
        use.update(window=window, used=0)
    reset = (window + 1) * 3600
    use["recent"] = [t for t in use["recent"] if now - t < 1] + [now]
    if len(use["recent"]) > BURST:
        return problem("429 Too Many Requests", "too many requests in one second", [("Retry-After", "2")])
    if use["used"] >= HOURLY:
        return problem("429 Too Many Requests", "hourly limit reached", [("Retry-After", str(math.ceil(reset - now)))])
    use["used"] += 1
    limits = [("X-RateLimit-Limit", str(HOURLY)), ("X-RateLimit-Remaining", str(HOURLY - use["used"])),
              ("X-RateLimit-Reset", str(reset))]
    n, size = int(asked.query.get("page", "1")), min(int(asked.query.get("per_page", "50")), 100)
    last = math.ceil(len(products()) / size)
    link = [("Link", f'<{SITE}/api/v2/products?page={n + 1}&per_page={size}>; rel="next"')] if n < last else []
    body = json.dumps({"items": [item(p) for p in products()[(n - 1) * size:n * size]]})
    return text("200 OK", body, "application/json", limits + link)

lab_api.env (the lab's test key) · api/lab_api.env

# The practice shop's test API key. Your real .env file holds your own key and is listed in .gitignore.
NIMBU_API_KEY=nb_test_5f1c9a7e2d4b8c60