Consuming REST APIs
1 · The lesson
readIf a site has an API, use it. Scraping HTML for data that is already available as JSON takes more work and breaks more often. A REST API gives you a contract: documented endpoints, stable schemas, real status codes, and rate-limit headers you can rely on.
This lesson covers what a good API client needs: auth, pagination, retries and timeouts. We'll use requests for the synchronous examples, and introduce httpx for when you move to async code.
1. HTTP Verbs: What Each One Means
The verb you pick tells the API what you want to do, and APIs treat each verb differently.
| 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. | Not guaranteed | Yes |
DELETE | Remove a resource. | Yes | No |
"Idempotent" means calling it twice has the same effect as calling it once. This matters 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 parameters are URL-encoded for you 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: the dict is turned into JSON and Content-Type is set to JSON r = requests.post( "https://api.example.com/users", json={"name": "Bob", "tags": ["py", "ml"]}, ) # Form body (the older style), sent as 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 common bug. An API that expects application/json and receives application/x-www-form-urlencoded replies with a 400 error that is often hard to understand.
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 ask for JSON when the server offers several formats, and User-Agent to identify your client (the same good manners 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')
A Session keeps headers and cookies between requests, and reuses TCP connections. Your code is faster and repeats itself less.
4. Reading the Response
r = requests.get(url, timeout=10) r.status_code # an int, for example 200, 404 or 503 r.headers # like a dict, but the keys ignore upper and lower case r.text # the body decoded into a str r.content # the raw body as bytes r.json() # parsed JSON, or raises JSONDecodeError r.url # final URL (after redirects) r.elapsed # a timedelta: how long the request and reply 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() # a 4xx or 5xx status raises HTTPError data = r.json() # raises JSONDecodeError if the body is HTML or 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 where things often go wrong. 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: used by OAuth, JWT and most modern APIs:
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 and 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. Base64 is not encryption, so basic auth over plain HTTP sends your password in readable form.
OAuth 2.0: for when the API needs a user's consent (your app acts on behalf of a user). Don't write it yourself. Use authlib or requests-oauthlib:
pip install authlib
OAuth involves several round-trips, refresh tokens, callback URLs and PKCE, so a library saves you a lot of work and mistakes. The authorisation-code flow is the standard for server-side apps, and PKCE is the standard for desktop/mobile/SPAs.
6. Pagination: Three Kinds
Almost every API splits long lists into pages. How it does that varies:
Page + limit: the 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 gives 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: used by GitHub and many other 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 many requests 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, and slow down 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) r = requests.get(url, timeout=10, **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 one header you must obey. If you ignore it, many APIs block you for longer, or ban your IP address.
8. Always Set a Timeout
# Wrong: this can wait forever if the server stops sending r = requests.get(url) # Right: give up quickly with an error r = requests.get(url, timeout=10) # Better: separate timeouts for connecting and for reading 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. If the API you call stops responding, your whole script freezes with no error. Always pass timeout=. For most APIs, timeout=10 is fine. For endpoints that are slow for a good reason (large reports, bulk operations), increase the read timeout.
The tuple form lets you keep the connect timeout short (DNS + TCP handshake should take less than a second) while allowing a longer read for slow endpoints.
9. Retries: The Production Pattern
Mount urllib3.util.Retry on a Session adapter and let it handle the retries:
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. These are temporary server problems or rate limits, so the same request can succeed later.
- Don't retry 4xx (except 429). A 400 means your request is wrong, so retrying it won't help.
backoff_factor=1gives the standard exponential delay. Use0.5for APIs you call very often, and2for APIs with strict rate limits.allowed_methods: by default,urllib3won't retryPOST(it is not idempotent). Add it only if your API uses idempotency keys.- Cap
total. 5 retries over about 30 seconds is a sensible upper limit. If they all fail, pass the error 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: the base class for allrequestserrors.requests.ConnectionError: DNS, TCP, connection refused.requests.Timeout: yourtimeout=ran out.requests.HTTPError: non-2xx status (only fromraise_for_status()).requests.exceptions.JSONDecodeError:r.json()on non-JSON.
Catch specific exceptions. A blanket except Exception: also catches your own bugs (a KeyError, a typo in a name) and hides them. See exceptions.
11. httpx: The Modern Alternative
httpx has almost the same API as requests, plus async support, HTTP/2 and stricter defaults. It is a good choice for new code, especially code that might become async later.
pip install httpx
Synchronous use looks almost the same as requests:
import httpx r = httpx.get("https://api.example.com/me", timeout=10) r.raise_for_status() print(r.json())
Async is the main reason to switch. You can send many requests at the same time from 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 take 20 seconds. 100 concurrent httpx calls take about 250ms, because you mostly wait for the slowest single response. This helps most when the work is I/O-bound and the API's rate limit allows many requests at once. See async for more. Never call the blocking requests.get inside an async def: it freezes the entire event loop.
12. Building APIs: What Comes Next
This lesson is about consuming APIs. When you want to build your own, the main frameworks are:
- FastAPI: modern, async-first, automatic OpenAPI docs, Pydantic validation.
- Flask: simpler, synchronous, still very widely used.
- Django REST Framework: for when you already use 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=. It costs one keyword argument, and it stops your script from 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 token committed to git stays in the history, even if you git rm it five minutes later. Treat it as leaked.
4. Ignoring rate-limit headers
Sending requests after X-RateLimit-Remaining: 0 gets you a 429, then a longer ban. Read the headers and slow down when they tell you 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 you find that your "complete user list" is missing most of the users. Paginate every list endpoint.
6. requests inside async defrequests.get is blocking. Called inside an async function, it freezes the entire event loop, and every other task stops. Use httpx.AsyncClient for async, or run blocking calls in a thread (asyncio.to_thread). See async.
7. Retrying 5xx forever
Cap attempts. A service that is really down shouldn't keep your client waiting forever. Raise the error after N retries and let the caller decide.
8. Mixing data= and json=data={"x": 1} sends x=1 urlencoded, and json={"x": 1} sends {"x":1} with the right Content-Type. An API that expects one and gets the other returns a 400 error that is hard to understand.
🎯 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, and don't collect them all 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 with a 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 test API # (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 fromgives records one at a time. The caller canbreakearly, and pages it doesn't need are never fetched. timeout=10on every request, so the script never hangs without an error.Retry-Afterrespected: on 429, it sleeps exactly as long as the server asked.raise_for_status()on everything else, so a 500 or 401 raises an error instead of being ignored.- Stops on an empty page, when the API has no more data. There is no need for a separate "total pages" call.
What's missing for production:
- Wrap in
Sessionwith aRetryadapter so 5xx transients retry automatically (Section 9). - Cap iterations with a
max_pageslimit, so a buggy API can't make it loop forever. - Cursor-based pagination support: loop on the
next_cursorfield instead of increasing a page number.
But the core structure (a generator, one fetch per page, Retry-After handling, a clean stop) is the same one production code uses. Add the rest when the API needs it.
What You Learned
- Verbs show intent. GET, PUT and DELETE are idempotent, so they are 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 kinds: page+limit, cursor, Link header. Use generators to process records one at a time instead of loading them all.
- Read rate-limit headers such as
X-RateLimit-RemainingandRetry-After. Slow down early, and always obeyRetry-After. - 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: almost the same API asrequests, plusAsyncClientfor concurrent calls. Never call blockingrequestsinasync def.
Next: CSV & JSON. Once the data is in your script, save it in a structured format.
Found this useful? Share it with a friend who’s learning Python.