P15 Paginated extraction · P16 Memoize, LRU cache

FreeLesson 65 of 78

P15Paginated extractionDEMEDIUM

Sounds like: "pull every page from this API", "the endpoint returns 100 rows and a next link"

Worked question: Extract every record from a cursor-paginated API (simulated, so it runs here) as one stream of rows.

Approach: 1. call, yield the rows, read the next cursor 2. stop when there is no cursor 3. expose it as a generator

SOLUTIONPYTHON 3.12

SERVER = list(range(101, 116))        # 15 orders on the "server"

def fake_api(cursor=None, limit=4):
    start = int(cursor or 0)
    page = SERVER[start:start + limit]
    nxt = start + limit
    return {"data": page,
            "next_cursor": str(nxt) if nxt < len(SERVER) else None}

def extract_all(limit=4):
    cursor, n = None, 0
    while True:
        resp = fake_api(cursor, limit)
        n += 1
        print(f"  page {n}: {len(resp['data'])} rows,"
              f" next={resp['next_cursor']}")
        yield from resp["data"]
        cursor = resp["next_cursor"]
        if cursor is None:
            break

ids = list(extract_all())
print(len(ids), ids[:3], "...", ids[-1])

Output

  page 1: 4 rows, next=4
  page 2: 4 rows, next=8
  page 3: 4 rows, next=12
  page 4: 3 rows, next=None
15 [101, 102, 103] ... 115

THE REAL THINGREQUESTS · NOT RUN

import requests

s = requests.Session()
s.headers["Authorization"] = f"Bearer {token}"

def pages(url, params=None):
    while url:
        r = s.get(url, params=params, timeout=30)
        r.raise_for_status()   # 4xx/5xx raise
        body = r.json()
        yield from body["data"]
        url, params = body.get("next"), None

P16Memoize, LRU cacheDEMEDIUM

Sounds like: "avoid repeat API lookups", "cache dimension lookups", "design an LRU cache"

Worked question: Cache pincode lookups so repeats cost nothing, then build an LRU cache from scratch.

Approach: 1. same input, same output? then cache it 2. bound the cache size 3. evict the least recently used

SOLUTIONPYTHON 3.12

from collections import OrderedDict
from functools import lru_cache

calls = 0
@lru_cache(maxsize=128)
def city_for_pincode(pin):            # pretend: a slow API call
    global calls
    calls += 1
    return {"560001": "Bengaluru", "400001": "Mumbai"}.get(pin, "?")

for pin in ["560001", "400001", "560001", "560001", "999999"]:
    city_for_pincode(pin)
print("real calls:", calls, city_for_pincode.cache_info())

class LRU:                       # build it yourself
    def __init__(self, cap):
        self.cap, self.d = cap, OrderedDict()
    def get(self, k):
        if k not in self.d:
            return None
        self.d.move_to_end(k)         # mark as recently used
        return self.d[k]
    def put(self, k, v):
        self.d[k] = v
        self.d.move_to_end(k)
        if len(self.d) > self.cap:
            self.d.popitem(last=False)   # evict least recent

c = LRU(2)
c.put("a", 1); c.put("b", 2); c.get("a"); c.put("c", 3)
print(list(c.d))                      # "b" was evicted

Output

real calls: 3 CacheInfo(hits=2, misses=3, maxsize=128, currsize=3)
['a', 'c']

Practice before you move on

Solve the two practice questions for P15 and P16 in Coding Practice, under Python.
Make the simulated API send one cursor twice and check that your extractor stops.

In Coding Practice

Solve this pattern on fresh data, under Python. Hidden tests grade it.