Event hooks and custom transport adapters

Requests · Expert · 13 min

Hooks and adapters are how logging, metrics, caching and non-HTTP backends plug into Requests cleanly.

Requests has two extension points beyond auth.

Event hooks. The only hook is "response": a callable that receives the Response (and keyword arguments such as timeout and verify) after it arrives. Register it per request (hooks={"response": fn}) or on a session (s.hooks["response"].append(fn)). If the hook returns a value, that value replaces the response; returning None keeps it. Typical uses are logging, metrics, and calling raise_for_status automatically for every request of a client.

Transport adapters. Session.mount(prefix, adapter) routes every URL that starts with prefix to an adapter, choosing the longest matching prefix. So s.mount("https://api.example.com/", special) overrides the default https:// adapter for that one API only. An adapter subclasses requests.adapters.BaseAdapter (or HTTPAdapter) and implements send(request, **kwargs) returning a Response, plus close().

Custom adapters power:

  • per-host configuration (different retries or pool sizes),
  • caching (requests-cache and CacheControl work this way),
  • non-network backends: serving file:// URLs, routing to a WSGI app in tests, or Unix sockets (requests-unixsocket),
  • fault injection in tests.

When subclassing HTTPAdapter, override send and call super().send(...) to keep the normal behaviour around your additions. Keep hooks fast and side-effect free: they run synchronously in the request path.

Example

import requests
from requests.adapters import BaseAdapter

class StaticAdapter(BaseAdapter):
    """Answers every request locally; useful for tests or offline mode."""
    def send(self, request, **kwargs):
        resp = requests.Response()
        resp.status_code = 200
        resp._content = f'{{"echo": "{request.method} {request.path_url}"}}'.encode()
        resp.headers["Content-Type"] = "application/json"
        resp.url, resp.request = request.url, request
        return resp
    def close(self):
        pass

log = []
with requests.Session() as s:
    s.mount("https://api.example.com/", StaticAdapter())   # longest prefix wins
    s.hooks["response"].append(lambda r, *a, **kw: log.append(r.status_code))
    r = s.get("https://api.example.com/users?id=1", timeout=5)
    print(r.json())     # {'echo': 'GET /users?id=1'}
print(log)              # [200]

Remember

  • The 'response' hook runs on every response and may replace it
  • mount(prefix, adapter) routes URLs by longest matching prefix
  • Custom adapters enable caching, test doubles and non-HTTP transports

Watch out for

  • Doing slow work (like network calls) inside a response hook, which delays every request.

Check yourself

  1. A session has adapters mounted at 'https://' and 'https://api.example.com/'. Which handles https://api.example.com/v1/items?

Sign in to answer, see the explanations, and get this lesson back as a quick review later.

Official docs: requests.readthedocs.io/en/latest/user/advanced/#transport-adapters