APIs and Async

In PyLearn the imports are fake_requests and fake_httpx, because Python in the browser has no sockets. Everything below is the shape you would write against the real requests and httpx.

Making a request

import fake_requests as requests

r = requests.get("https://api.pylearn.dev/weather", params={"city": "Oslo"})
r = requests.post(url, json={"name": "Ada"}, headers={"X-Trace": "1"})

Pass params rather than gluing values onto the URL — the library escapes them.

Reading the response

r.status_code      # 200
r.ok               # True for 2xx and 3xx
r.headers["retry-after"]
r.text             # the body as a string
r.json()           # parsed into dicts and lists
r.raise_for_status()   # raises on 4xx / 5xx, silent otherwise

Status codes

Code	Meaning	Retry?
200 / 201	OK / Created	—
400	your request was malformed	no
401	it does not know who you are	no
403	it knows, and you may not	no
404	not there	no
429	too many requests	yes, after `Retry-After`
500 / 503	the server broke	yes

The rule: 4xx will fail again; 5xx and 429 might not.

Parsing defensively

data = r.json()
name  = data.get("name") or "unknown"     # covers missing AND null
tags  = data.get("tags") or []
count = int(data.get("count", 0))

data.get("tags", []) still returns None when the key exists and holds null — which is the case that actually breaks things.

Authentication

headers = {"Authorization": f"Bearer {token}"}
r = requests.get(url, headers=headers)

Read the token from the environment, never from source, and never put it in a query string — URLs end up in logs, history and referrer headers.

Retries with backoff

def should_retry(status):
    return status == 429 or 500 <= status < 600

for attempt in range(max_attempts):
    r = requests.get(url)
    if r.ok:
        return r.json()
    if not should_retry(r.status_code):
        break
    wait = float(r.headers.get("retry-after", 2 ** attempt))
    time.sleep(wait)

A ceiling, exponential backoff, and only retrying what might succeed. All three matter.

async and await

import asyncio

async def greet(name):
    await asyncio.sleep(0.1)
    return f"Hello, {name}"

asyncio.run(greet("Ada"))

async def defines a coroutine function; calling it runs nothing.
await runs a coroutine and waits for its value — only legal inside async def.
asyncio.run(main()) starts the loop once, at the top.

Concurrency

# Serial — three 100ms fetches take 300ms.
for city in cities:
    results.append(await fetch(city))

# Concurrent — about 100ms.
results = await asyncio.gather(*(fetch(c) for c in cities))

gather returns results in the order passed in, not the order they finished. If one raises, the exception propagates; return_exceptions=True returns it as a result instead.

async with httpx.AsyncClient(base_url=BASE) as client:
    pages = await asyncio.gather(*(client.get(u) for u in urls))

Async iteration

async def pages(last):            # an async generator
    for n in range(1, last + 1):
        await asyncio.sleep(0)
        yield n

async for page in pages(3):
    print(page)

Written out longhand it is __aiter__ plus an __anext__ that raises StopAsyncIteration at the end.

Timeouts and cancellation

try:
    result = await asyncio.wait_for(slow(), timeout=2.0)
except asyncio.TimeoutError:
    result = None

A timeout cancels the coroutine: CancelledError is raised inside it at its current await, so finally cleanup still runs.

try:
    await something()
except asyncio.CancelledError:
    await tidy_up()
    raise            # always re-raise

A client class

class WeatherClient:
    def __init__(self, http, base_url, token=None):
        self.http = http          # injected, so a test can pass a stub
        self.base_url = base_url
        self.token = token

    def _headers(self):
        return {"Authorization": f"Bearer {self.token}"} if self.token else {}

    def get_weather(self, city):
        r = self.http.get(self.base_url + "/weather",
                          params={"city": city}, headers=self._headers())
        r.raise_for_status()
        return r.json()

Convert the API's shape into your own at this boundary, so a renamed field upstream is one edit rather than fifty.

Pagination

page, users = 1, []
while page <= max_pages:
    data = requests.get(url, params={"page": page}).json()
    users.extend(data["users"])
    if not data["has_next"]:
        break
    page += 1

Stop on the server's signal, and keep a ceiling so a broken API cannot loop forever.

Testing without a network

class FakeHttp:
    def __init__(self, temps):
        self.temps, self.calls = temps, []

    def get(self, path, params=None):
        self.calls.append((path, params))
        return FakeResponse({"temp_c": self.temps[params["city"]]})

client = WeatherClient(FakeHttp({"Oslo": 3}), BASE)

Fast, deterministic, credential-free, and able to reproduce a 500 or a timeout on demand — which is the part a live service will never do for you on request.

This cheatsheet is the summary. If you want to build it yourself, the APIs and Async course walks you through it in the browser — the first lesson is free.