Consuming REST APIs
1 · The lesson
readIf 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.
| Verb | Means | Idempotent? | Has a body? |
|---|---|---|---|
GET | Fetch a resource. Pure read. | Yes | No |
POST | Create a new resource (or run an action). | No | Yes |
PUT | Replace a resource entirely. | Yes | Yes |
PATCH | Partially update a resource. | Usually | Yes |
DELETE | Remove a resource. | Yes | No |
"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.
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.
# 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
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:
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
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:
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:
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.
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:
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:
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:
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:
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:
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:
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:
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:
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
# 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:
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=1gives the canonical exponential delay.0.5for chattier APIs,2for very rate-sensitive ones.allowed_methods— by default,urllib3won't retryPOST(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:
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 allrequestserrors.requests.ConnectionError— DNS, TCP, connection refused.requests.Timeout— yourtimeout=fired.requests.HTTPError— non-2xx (only fromraise_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.
pip install httpx
Synchronous — drop-in requests replacement:
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:
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 defrequests.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=10on every request. - Stop when a page yields zero records.
- Honour
Retry-Afteron 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.
# Example use against a paginated API: # for record in fetch_all_pages("https://api.example.com/v1/users", {"limit": 100}): # process(record)
Skeleton:
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 awhile 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
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 fromstreams records one at a time. The caller canbreakearly without paying for pages it never wanted. timeout=10— every single request. No silent hangs.Retry-Afterrespected — 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
Sessionwith aRetryadapter so 5xx transients retry automatically (Section 9). - Cap iterations — a sanity
max_pagesguard so a buggy API can't loop forever. - Cursor-based pagination support — easy variant: yield/loop on the
next_cursorfield 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 wrapr.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; obeyRetry-Afterabsolutely. - Retries: mount
urllib3.util.Retryon aSession. Retry 5xx and 429, never other 4xx. Cap total. - Errors:
RequestException→ network.HTTPError→ status.JSONDecodeError→ content. Catch specifically. httpx— same API asrequests, plusAsyncClientfor concurrent calls. Never call blockingrequestsinasync def.
Next: CSV & JSON — once the data's in your script, save it somewhere structured.