PythonMastery
intermediate 20 min read · lesson 2 of 12 in Python How-To

Consuming REST APIs

1 · The lesson

read

If a site has an API, use it. Scraping HTML for data that's already available as JSON is doing twice the work for half the reliability. A REST API gives you a contract: documented endpoints, stable schemas, real status codes, rate-limit headers that mean something.

This lesson is the toolkit for being a good API client — auth, pagination, retries, timeouts, the lot. We'll use requests for the synchronous examples and introduce httpx for when you graduate to async.


1. HTTP Verbs — What Each One Implies

The verb you pick communicates intent. APIs care.

VerbMeansIdempotent?Has a body?
GETFetch a resource. Pure read.YesNo
POSTCreate a new resource (or run an action).NoYes
PUTReplace a resource entirely.YesYes
PATCHPartially update a resource.UsuallyYes
DELETERemove a resource.YesNo

"Idempotent" means calling it twice has the same effect as calling it once — important for retry logic. You can safely retry GET, PUT, DELETE. You should not retry POST without a deduplication key, or you'll create two of whatever it was meant to create.

python
import requests

r = requests.get("https://api.example.com/users/42")
r = requests.post("https://api.example.com/users", json={"name": "Bob"})
r = requests.put("https://api.example.com/users/42", json={"name": "Bob", "email": "s@x.com"})
r = requests.patch("https://api.example.com/users/42", json={"email": "new@x.com"})
r = requests.delete("https://api.example.com/users/42")

2. Query Parameters and JSON Bodies

requests does the encoding for you. Don't build URLs by hand.

python
# Query params — auto-urlencoded
r = requests.get(
    "https://api.example.com/search",
    params={"q": "python tips", "page": 2, "limit": 50},
)
# Built URL: .../search?q=python+tips&page=2&limit=50

# JSON body — Python dict → JSON, sets Content-Type: application/json
r = requests.post(
    "https://api.example.com/users",
    json={"name": "Bob", "tags": ["py", "ml"]},
)

# Form body (old-school) — application/x-www-form-urlencoded
r = requests.post(
    "https://api.example.com/login",
    data={"username": "surya", "password": "..."},
)
+ setup added so this can run · defines requests
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

requests = _AutoMock('requests')

The json= and data= distinction matters. json= serialises to JSON and sets the header. data= urlencodes a dict (form post) or sends raw bytes if you pass a string. Mixing them up is a top-five API integration bug — APIs reject application/x-www-form-urlencoded when they expected application/json with a confusing 400.


3. Headers — Auth, Accept, Content-Type

python
HEADERS = {
    "Authorization": f"Bearer {token}",
    "Accept": "application/json",
    "User-Agent": "myapp/1.0 (+contact@myapp.com)",
}

r = requests.get("https://api.example.com/me", headers=HEADERS, timeout=10)
+ setup added so this can run · defines requests, token
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

requests = _AutoMock('requests')
token = _AutoMock('token')

You'll set headers in most real API calls — Authorization always, Accept to nudge the server toward JSON when it serves multiple formats, User-Agent to identify your client (the same etiquette as in scraping). For a session of related calls, attach headers to a Session once:

python
with requests.Session() as s:
    s.headers.update(HEADERS)
    me = s.get("https://api.example.com/me").json()
    prefs = s.get("https://api.example.com/me/prefs").json()
+ setup added so this can run · defines HEADERS, requests
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

HEADERS = _AutoMock('HEADERS')
requests = _AutoMock('requests')

Session persists headers, cookies, and reuses TCP connections. Faster and less code repetition.


4. Reading the Response

python
r = requests.get(url, timeout=10)

r.status_code          # int — 200, 404, 503, etc.
r.headers              # dict-like — case-insensitive lookup
r.text                 # str — decoded body
r.content              # bytes — raw body
r.json()               # parsed JSON, or raises JSONDecodeError
r.url                  # final URL (after redirects)
r.elapsed              # timedelta — how long the round-trip took
+ setup added so this can run · defines url, requests
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

url = _AutoMock('url')
requests = _AutoMock('requests')

The combination you'll write a hundred times:

python
r = requests.get(url, timeout=10)
r.raise_for_status()                            # 4xx/5xx → HTTPError
data = r.json()                                 # may raise JSONDecodeError on HTML/empty
+ setup added so this can run · defines url, requests
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

url = _AutoMock('url')
requests = _AutoMock('requests')

r.json() is the trapdoor. If the server returns 200 OK with Content-Type: text/html (a CDN intercept, a captive portal, an error page that forgot its status code), r.json() raises requests.exceptions.JSONDecodeError. Wrap it:

python
try:
    data = r.json()
except requests.exceptions.JSONDecodeError:
    raise RuntimeError(f"expected JSON, got {r.headers.get('Content-Type')}: {r.text[:200]}")
+ setup added so this can run · defines r, requests
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

r = _AutoMock('r')
requests = _AutoMock('requests')

5. Auth Patterns

API key in a header — the most common pattern. Keep the key in an environment variable, never in code. See envconfig.

python
import os, requests

key = os.environ["API_KEY"]
r = requests.get(
    "https://api.example.com/v1/things",
    headers={"X-API-Key": key},
    timeout=10,
)
+ setup added so this can run · defines
import os  # noqa: F401
os.environ.setdefault("API_KEY", "example-api-key")

Bearer token — OAuth-style, JWT-style, almost everything modern:

python
r = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=10)
+ setup added so this can run · defines url, requests, token
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

url = _AutoMock('url')
requests = _AutoMock('requests')
token = _AutoMock('token')

HTTP Basic — username/password in a header. Used by old APIs and many internal tools:

python
r = requests.get("https://api.example.com/admin", auth=("user", "pass"), timeout=10)
+ setup added so this can run · defines requests
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

requests = _AutoMock('requests')

requests base64-encodes the credentials and adds the Authorization: Basic ... header. Always use HTTPS — basic auth over HTTP is plaintext.

OAuth 2.0 — when the API requires user consent (logging in on behalf of a user). Don't roll it yourself. Use authlib or requests-oauthlib:

bash
pip install authlib

OAuth involves multiple round-trips, refresh tokens, callback URLs, PKCE — a library is genuinely worth it. The authorisation-code flow is the standard for server-side apps; PKCE is the standard for desktop/mobile/SPAs.


6. Pagination — Three Flavours

Almost every API paginates. The shape varies:

Page + limit — most common:

python
def fetch_all_paged(url, params=None, page_param="page"):
    params = dict(params or {})
    page = 1
    while True:
        params[page_param] = page
        r = requests.get(url, params=params, timeout=10)
        r.raise_for_status()
        chunk = r.json()
        if not chunk:
            return
        yield from chunk
        page += 1
+ setup added so this can run · defines requests
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

requests = _AutoMock('requests')

Cursor-based — the server hands you a token for the next page:

python
def fetch_all_cursor(url, params=None):
    params = dict(params or {})
    while True:
        r = requests.get(url, params=params, timeout=10)
        r.raise_for_status()
        body = r.json()
        yield from body["items"]
        cursor = body.get("next_cursor")
        if not cursor:
            return
        params["cursor"] = cursor
+ setup added so this can run · defines requests
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

requests = _AutoMock('requests')

Link header — GitHub, many REST APIs. The next page's URL is in the Link response header:

python
def fetch_all_link_header(url):
    while url:
        r = requests.get(url, timeout=10)
        r.raise_for_status()
        yield from r.json()
        url = r.links.get("next", {}).get("url")     # requests parses Link for you
+ setup added so this can run · defines requests
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

requests = _AutoMock('requests')

Use a generator (the yield from pattern above) so the caller streams records one at a time instead of holding everything in memory. See generators.


7. Rate Limits — Read the Headers

Well-behaved APIs tell you exactly how much budget you have left. The headers vary by provider but the pattern is standard:

python
X-RateLimit-Limit:     1000
X-RateLimit-Remaining: 4
X-RateLimit-Reset:     1715688000     (unix timestamp)
Retry-After:           42             (seconds, on a 429)

Read them. Back off before you get banned:

python
def get_with_rate_limit(url, **kwargs):
    r = requests.get(url, timeout=10, **kwargs)
    if r.status_code == 429:
        wait = int(r.headers.get("Retry-After", "60"))
        time.sleep(wait)
        return get_with_rate_limit(url, **kwargs)    # retry once
    remaining = int(r.headers.get("X-RateLimit-Remaining", "1"))
    if remaining < 5:
        time.sleep(1)                                # slow down voluntarily
    return r
+ setup added so this can run · defines requests, kwargs, time
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

requests = _AutoMock('requests')
kwargs = _AutoMock('kwargs')
time = _AutoMock('time')

Retry-After is the only header you must obey. Ignore it and you'll see 429s turn into 403s, then IP bans.


8. Timeouts — Always

python
# WRONG — can hang forever if the server stops sending
r = requests.get(url)

# RIGHT — fail fast
r = requests.get(url, timeout=10)

# BETTER — separate connect and read timeouts
r = requests.get(url, timeout=(3, 10))             # 3s connect, 10s read
+ setup added so this can run · defines url, requests
# Lightweight mock for objects whose attributes/methods aren't critical
class _AutoMock:
    def __init__(self, name='mock'): self._name = name
    def __getattr__(self, k): return _AutoMock(self._name + '.' + k)
    def __call__(self, *a, **kw):
        print('-> ' + self._name + '() called')
        return _AutoMock(self._name + '()')
    def __repr__(self): return '<mock ' + self._name + '>'
    def __str__(self): return '<mock ' + self._name + '>'
    def __bool__(self): return True
    def __iter__(self): return iter([])
    def __len__(self): return 0
    def __getitem__(self, k): return _AutoMock(self._name + '[...]')
    def __setitem__(self, k, v): pass
    def __enter__(self): return self
    def __exit__(self, *a): return False
    async def __aenter__(self): return self
    async def __aexit__(self, *a): return False
    def __add__(self, o): return self
    def __radd__(self, o): return self
    def __sub__(self, o): return self
    def __mul__(self, o): return self
    def __rmul__(self, o): return self
    def __truediv__(self, o): return self
    def __eq__(self, o): return isinstance(o, _AutoMock)
    def __hash__(self): return hash(self._name)
    def __lt__(self, o): return True
    def __le__(self, o): return True
    def __gt__(self, o): return False
    def __ge__(self, o): return False
    def __mro_entries__(self, bases): return (object,)

url = _AutoMock('url')
requests = _AutoMock('requests')

requests with no timeout will wait forever for a response. A flaky upstream API will silently freeze your entire script. Always pass timeout=. For most APIs, timeout=10 is fine; for endpoints that legitimately take a while (large reports, bulk operations), bump the read timeout.

The tuple form lets you cap the connect phase tightly (DNS + TCP handshake should be sub-second) while allowing a longer read for slow endpoints.


9. Retries — The Production Pattern

Stack urllib3.util.Retry on a Session adapter and let it handle the messy bits:

python
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def build_session():
    retry = Retry(
        total=5,
        backoff_factor=1,                            # 0, 2, 4, 8, 16 seconds
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["GET", "PUT", "DELETE", "HEAD", "OPTIONS"],
        respect_retry_after_header=True,
    )
    s = requests.Session()
    adapter = HTTPAdapter(max_retries=retry)
    s.mount("https://", adapter)
    s.mount("http://", adapter)
    return s


with build_session() as s:
    r = s.get("https://api.example.com/v1/users", timeout=10)

Key choices:

  • Retry 5xx and 429. They're transient or rate-limit issues, retryable by definition.
  • Don't retry 4xx (except 429). A 400 means your request is wrong; retrying it won't help.
  • backoff_factor=1 gives the canonical exponential delay. 0.5 for chattier APIs, 2 for very rate-sensitive ones.
  • allowed_methods — by default, urllib3 won't retry POST (non-idempotent). Add it only if your API uses idempotency keys.
  • Cap total. 5 attempts over ~30 seconds is a sane upper bound. If it fails 5 times, escalate to the caller — don't retry forever.

10. Error Handling — 4xx vs 5xx

Different status families need different responses:

python
try:
    r = session.get(url, timeout=10)
    r.raise_for_status()
except requests.HTTPError as e:
    code = e.response.status_code
    if code == 401:
        raise AuthError("token expired, refresh and retry")
    if code == 404:
        return None                                  # missing, not an error
    if 400 <= code < 500:
        raise ClientError(f"request invalid: {e.response.text}")
    if 500 <= code < 600:
        raise ServerError("upstream failed after retries")
    raise
except requests.RequestException as e:
    raise NetworkError("connection problem") from e

The hierarchy:

  • requests.RequestException — base for all requests errors.
  • requests.ConnectionError — DNS, TCP, connection refused.
  • requests.Timeout — your timeout= fired.
  • requests.HTTPError — non-2xx (only from raise_for_status()).
  • requests.exceptions.JSONDecodeError — r.json() on non-JSON.

Catch specifically. A blanket except Exception: will swallow KeyboardInterrupt (in older Pythons) and hide real bugs. See exceptions.


11. httpx — The Modern Alternative

httpx is the same shape as requests, plus async support, HTTP/2, and stricter defaults. It's what new code should reach for, especially anything that might go async later.

bash
pip install httpx

Synchronous — drop-in requests replacement:

python
import httpx

r = httpx.get("https://api.example.com/me", timeout=10)
r.raise_for_status()
print(r.json())

Async — the real reason to switch. Fan out concurrent requests with one event loop:

python
import asyncio
import httpx

async def fetch_user(client, user_id):
    r = await client.get(f"https://api.example.com/users/{user_id}", timeout=10)
    r.raise_for_status()
    return r.json()

async def main():
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[fetch_user(client, i) for i in range(1, 101)])
    return results

users = asyncio.run(main())

100 sequential requests.get calls at 200ms each = 20 seconds. 100 concurrent httpx calls = ~250ms (network-bound on the slowest single response). The win is huge when you have an API that's I/O-bound and rate-limit-tolerant. See async for the full picture, and never call blocking requests.get inside an async def — it freezes the entire event loop.


12. Building APIs — A Forward Pointer

This lesson is about consuming APIs. When you serve them, the toolkits are:

  • FastAPI — modern, async-first, automatic OpenAPI docs, Pydantic validation.
  • Flask — simpler, synchronous, still ubiquitous.
  • Django REST Framework — when you already have Django.

Pick FastAPI for new work unless you have a strong reason not to. A future lesson covers building your own.


13. Common Mistakes

1. No timeout
Default requests.get(url) waits forever. Always timeout=. The cost is one keyword argument; the benefit is your script not silently freezing in production.

2. r.json() without try
Non-JSON responses (HTML error pages, captive portals, gzipped bodies that failed to decode) raise JSONDecodeError. Wrap it or check Content-Type first.

3. Secrets in code
API keys, tokens, basic-auth passwords belong in environment variables. See envconfig. A committed token in git history is leaked forever — even if you git rm it five minutes later.

4. Ignoring rate-limit headers
Hammering past X-RateLimit-Remaining: 0 earns you a 429, then a longer ban. Read the headers; back off when they say to.

5. Not paginating
Many APIs default to a limit=20. Your script grabs the first page, you assume that's the full data, and three weeks later your "complete user list" is missing 90% of users. Paginate every list endpoint.

6. requests inside async def
requests.get is blocking. Called inside an async function, it freezes the entire event loop — every other task halts. Use httpx.AsyncClient for async, or run blocking calls in a thread (asyncio.to_thread). See async.

7. Retrying 5xx forever
Cap attempts. A genuinely-down service shouldn't keep your client tied up indefinitely; surface the error after N retries and let the caller decide.

8. Mixing data= and json=
data={"x": 1} sends x=1 urlencoded; json={"x": 1} sends {"x":1} with the right Content-Type. APIs that want one and got the other return baffling 400s.


🎯 Your Turn — Paginated Generator with Retry-After

Write fetch_all_pages(url, params=None, page_param="page") that fetches all pages from a page-number-paginated API and yields records one at a time (use a generator — don't accumulate in memory). It must:

  • Set timeout=10 on every request.
  • Stop when a page yields zero records.
  • Honour Retry-After on 429: sleep and retry that page.
  • raise_for_status() on any other non-2xx.

Assume the API returns a JSON list of records per page.

python
# Example use against a paginated API:
#   for record in fetch_all_pages("https://api.example.com/v1/users", {"limit": 100}):
#       process(record)

Skeleton:

python
import time
import requests

def fetch_all_pages(url, params=None, page_param="page"):
    params = dict(params or {})
    page = 1
    while True:
        params[page_param] = page
        # TODO 1: GET with timeout=10
        # TODO 2: if 429, sleep Retry-After and retry the same page (don't increment)
        # TODO 3: raise_for_status() on other non-2xx
        # TODO 4: parse JSON; if empty, return
        # TODO 5: yield records one at a time
        # TODO 6: increment page
        ...
Hint 1 — Retry without incrementing Wrap the GET in a while True: inside the page loop. On 429, sleep and continue the inner loop. On success, break out and process. This keeps page on the current value until we get a real response.
Hint 2 — Streaming via generator yield from records yields each item individually. The caller iterates one-by-one. Memory stays O(page size), not O(total dataset).
Show full solution
python
import time
import requests

def fetch_all_pages(url, params=None, page_param="page"):
    """Stream all records from a page-numbered API. Honours Retry-After on 429."""
    params = dict(params or {})
    page = 1
    while True:
        params[page_param] = page
        # Retry-After loop for the current page
        while True:
            r = requests.get(url, params=params, timeout=10)
            if r.status_code == 429:
                wait = int(r.headers.get("Retry-After", "10"))
                time.sleep(wait)
                continue                                 # retry the same page
            r.raise_for_status()
            break
        records = r.json()
        if not records:
            return                                       # past the last page
        yield from records
        page += 1


# Demo — using a public sandbox
# (https://jsonplaceholder.typicode.com/posts supports _page + _limit)
count = 0
for post in fetch_all_pages(
    "https://jsonplaceholder.typicode.com/posts",
    {"_limit": 25},
    page_param="_page",
):
    count += 1
    if count <= 3:
        print(f"  {post['id']}: {post['title'][:50]}...")
print(f"streamed {count} records total")

What this gets right:

  • Generator-based — yield from streams records one at a time. The caller can break early without paying for pages it never wanted.
  • timeout=10 — every single request. No silent hangs.
  • Retry-After respected — on 429, sleep exactly as long as the server asked. Don't second-guess the header.
  • raise_for_status() on everything else — a 500 or 401 fails loudly, not silently.
  • Empty-page termination — stops naturally when the API runs out of data, without needing a separate "total pages" call.

What's missing for production:

  • Wrap in Session with a Retry adapter so 5xx transients retry automatically (Section 9).
  • Cap iterations — a sanity max_pages guard so a buggy API can't loop forever.
  • Cursor-based pagination support — easy variant: yield/loop on the next_cursor field instead of incrementing a number.

But the core shape — generator + per-page fetch + Retry-After + clean termination — is the production pattern. Add the rest as the API demands.


What You Learned

  • Verbs imply intent. GET/PUT/DELETE are idempotent — safe to retry. POST usually isn't.
  • params= for query strings, json= for JSON bodies, data= for forms. Don't mix them up.
  • Always timeout=. Always check status (raise_for_status() or branch explicitly). Always wrap r.json() if non-JSON is possible.
  • Auth patterns: API key in header, bearer token, HTTP Basic, OAuth (use authlib). Keys live in env, never in code.
  • Pagination comes in three flavours: page+limit, cursor, Link-header. Use generators to stream, not slurp.
  • Read rate-limit headers — X-RateLimit-Remaining, Retry-After. Back off voluntarily; obey Retry-After absolutely.
  • Retries: mount urllib3.util.Retry on a Session. Retry 5xx and 429, never other 4xx. Cap total.
  • Errors: RequestException → network. HTTPError → status. JSONDecodeError → content. Catch specifically.
  • httpx — same API as requests, plus AsyncClient for concurrent calls. Never call blocking requests in async def.

Next: CSV & JSON — once the data's in your script, save it somewhere structured.