HTTPX and HTTPX2 Integration
Hishel adds HTTP caching to your HTTPX applications with minimal code changes.
This page also applies to HTTPX2, a fork of HTTPX maintained by Pydantic. Install the httpx2 extra and import from hishel.httpx2 instead:
from hishel.httpx2 import AsyncCacheClient, AsyncCacheTransport, CacheClient, CacheTransportInstallation
pip install "hishel[httpx]"Quick Start
The easiest way to add caching to your HTTPX application is using the cache-enabled client classes:
from hishel.httpx import CacheClient
with CacheClient() as client:
# First request - fetches from origin
client.get("https://api.example.com/data")
# Second request - served from cache
client.get("https://api.example.com/data")from hishel.httpx import AsyncCacheClient
async with AsyncCacheClient() as client:
# First request - fetches from origin
await client.get("https://api.example.com/data")
# Second request - served from cache
await client.get("https://api.example.com/data")Caching Modes
Default
Without any configuration, Hishel:
- caches successful
GETandHEADresponses, except206 Partial Content - uses the method and URL as the cache key
- ignores HTTP caching headers, including
no-store,privateandVary - serves an entry until the storage's TTL or TTI expires it
The cache is global. Everyone using it gets the same entries, whatever their Authorization or Cookie headers are, which suits a private cache inside a backend. If responses differ per user, use RFC mode or a separate storage for each user.
RFC 9111
With rfc=True, Hishel follows the HTTP caching specification instead, so caching headers decide what is stored, how long it stays fresh, and when it needs revalidating with the server:
from hishel.httpx import CacheClient
client = CacheClient(rfc=True)from hishel.httpx import AsyncCacheClient
client = AsyncCacheClient(rfc=True)To configure it, pass CacheOptions instead of True:
from hishel import CacheOptions
from hishel.httpx import CacheClient
client = CacheClient(rfc=CacheOptions(shared=False))CacheOptions accepts these options:
shared(defaultTrue):Trueacts as a shared cache, serving many users. It respectsprivateand doesn't store responses to requests withAuthorization, unless the response allows it (RFC 9111 Section 3.5).Falseacts as a private cache for a single user, which can store those responses.allow_stale(defaultFalse): Whether stale responses can be served without revalidating them, when caching directives allow it (RFC 9111 Section 4.2.4).supported_methods(default["GET", "HEAD"]): The methods whose responses can be stored. Only safe methods can be served from the cache, so listingPOSTstores responses that are never used.
Choosing a Storage
Clients and transports use a SQLite storage unless you pass another one:
from hishel import SqliteStorage
from hishel.httpx import CacheClient
client = CacheClient(storage=SqliteStorage(default_ttl=3600))from hishel import AsyncSqliteStorage
from hishel.httpx import AsyncCacheClient
client = AsyncCacheClient(storage=AsyncSqliteStorage(default_ttl=3600))Closing a client or transport also closes its storage, including one you passed in, so don't close it while another client is using the same storage.
Per-Request Configuration
Use the caching extension to override the mode, the storage, the expiry or the cache key for a single request:
from hishel import CacheOptions, SqliteStorage
from hishel.httpx import CacheClient
client = CacheClient()
# Follow RFC 9111 for this request only
client.get("https://api.example.com/data", extensions={"caching": {"rfc": True}})
# Or with options
client.get("https://api.example.com/data", extensions={"caching": {"rfc": CacheOptions(shared=False)}})
# Use a different storage for this request only
other_storage = SqliteStorage(database_path="other.db")
client.get("https://api.example.com/data", extensions={"caching": {"storage": other_storage}})
# Expire this response after 1 hour, or after 5 minutes without being used
client.get("https://api.example.com/data", extensions={"caching": {"ttl": 3600, "tti": 300}})
# Use your own cache key instead of the method and URL
client.get("https://api.example.com/data", extensions={"caching": {"key": "user-42:data"}})The extension is typed as CacheConfig for CacheClient and CacheTransport, and as AsyncCacheConfig for their async counterparts, both importable from hishel.httpx.
ttl and tti are in seconds and override the storage's default_ttl and default_tti. They apply when a response is stored, so a request answered by an existing entry doesn't change when that entry expires. They can't remove a limit set on the storage: None means the same as leaving them out.
key replaces the usual cache key, which is a hash of the method and URL in the default mode, and of the URL in RFC mode. For example, this keeps one user's entries apart from another's in the default mode.
- Lookup and store: the key is used as given, for finding entries and for storing a new one. A response stored under it isn't found by requests without it.
- Shared keys: in the default mode, requests with the same key and method get the same entry, whatever their URL. In RFC mode, an entry is only served to its own URL, so the other URLs store entries of their own under that key.
- Methods:
GETandHEADcan share a key, but don't answer each other. A key doesn't makePOSTresponses cacheable. - Invalidation: in RFC mode, a request with an unsafe method, such as
POST, removes the entries of its URL unless the response is an error. It only finds the entries under its own key, so it has to use the same key as the requests it should invalidate. - Redirects: the key also applies to the requests that follow a redirect. In the default mode, the final response is stored under it, so the next request gets that response directly, without the redirect in its history.
Cache Transports
For more control or to integrate with existing HTTPX clients, use cache transports directly:
import httpx
from hishel import SqliteStorage
from hishel.httpx import CacheTransport
# Create transport with caching
transport = CacheTransport(
next_transport=httpx.HTTPTransport(),
storage=SqliteStorage(),
rfc=True,
)
# Use with standard HTTPX client
client = httpx.Client(transport=transport)
response = client.get("https://api.example.com/data")import httpx
from hishel import AsyncSqliteStorage
from hishel.httpx import AsyncCacheTransport
# Create transport with caching
transport = AsyncCacheTransport(
next_transport=httpx.AsyncHTTPTransport(),
storage=AsyncSqliteStorage(),
rfc=True,
)
# Use with standard HTTPX client
client = httpx.AsyncClient(transport=transport)
response = await client.get("https://api.example.com/data")WARNING
CacheClient(transport=...) and AsyncCacheClient(transport=...) use the transport you pass as it is, without any caching. To cache on top of your own transport, wrap it in a cache transport as shown above.
Hits and Misses
Every response has a caching extension that says what the cache did:
from hishel.httpx import CacheClient
client = CacheClient()
response = client.get("https://api.example.com/data")
print(response.extensions["caching"]) # {'from_cache': False, 'revalidated': False, 'stored': True}
response = client.get("https://api.example.com/data")
print(response.extensions["caching"]) # {'from_cache': True, 'revalidated': False, 'stored': False}Its value is a hishel.CacheStatus:
from_cache: the body comes from the cache.revalidated: the response answers a conditional request, which the cache sent to check whether a stored response was still valid. With a 304,from_cacheis also true. Only in RFC mode.stored: the response is being stored. It is saved once its body has been read in full, so a stream you close early isn't stored after all.