0
# Cache Transports
1

2
HTTPX transport implementations that add a caching layer on top of existing transports. These transports handle cache lookups, storage, validation, and seamlessly integrate with HTTPX's transport architecture.
3

4
## Capabilities
5

6
### Synchronous Cache Transport
7

8
HTTP transport that wraps existing synchronous transports to add caching capabilities.
9

10
```python { .api }
11
class CacheTransport(httpx.BaseTransport):
12
    def __init__(self, *, transport: httpx.BaseTransport, storage=None, controller=None):
13
        """
14
        Synchronous HTTP transport with caching.
15
        
16
        Parameters:
17
        - transport: Underlying HTTPX transport to wrap
18
        - storage: Storage backend for cached responses (defaults to FileStorage)
19
        - controller: Cache controller for caching logic (defaults to Controller())
20
        """
21
    
22
    def handle_request(self, request: httpx.Request) -> httpx.Response:
23
        """Handle HTTP request with caching logic"""
24
    
25
    def close(self) -> None:
26
        """Close transport and storage resources"""
27
    
28
    def __enter__(self) -> "CacheTransport": ...
29
    def __exit__(self, *args) -> None: ...
30
```
31

32
**Usage Examples:**
33

34
```python
35
import httpx
36
import hishel
37

38
# Wrap existing transport with caching
39
base_transport = httpx.HTTPTransport()
40
cache_transport = hishel.CacheTransport(
41
    transport=base_transport,
42
    storage=hishel.FileStorage(),
43
    controller=hishel.Controller()
44
)
45

46
# Use directly with httpx
47
with httpx.Client(transport=cache_transport) as client:
48
    response = client.get("https://api.example.com/data")
49

50
# Custom transport configuration
51
import ssl
52
context = ssl.create_default_context()
53
base_transport = httpx.HTTPTransport(
54
    verify=context,
55
    cert=('client.crt', 'client.key'),
56
    timeout=30.0
57
)
58
cache_transport = hishel.CacheTransport(transport=base_transport)
59

60
with httpx.Client(transport=cache_transport) as client:
61
    response = client.get("https://api.example.com/data")
62
```
63

64
### Asynchronous Cache Transport
65

66
HTTP transport that wraps existing asynchronous transports to add caching capabilities.
67

68
```python { .api }
69
class AsyncCacheTransport(httpx.AsyncBaseTransport):
70
    def __init__(self, *, transport: httpx.AsyncBaseTransport, storage=None, controller=None):
71
        """
72
        Asynchronous HTTP transport with caching.
73
        
74
        Parameters:
75
        - transport: Underlying async HTTPX transport to wrap
76
        - storage: Async storage backend for cached responses (defaults to AsyncFileStorage)
77
        - controller: Cache controller for caching logic (defaults to Controller())
78
        """
79
    
80
    async def handle_async_request(self, request: httpx.Request) -> httpx.Response:
81
        """Handle HTTP request with caching logic asynchronously"""
82
    
83
    async def aclose(self) -> None:
84
        """Close transport and storage resources asynchronously"""
85
    
86
    async def __aenter__(self) -> "AsyncCacheTransport": ...
87
    async def __aexit__(self, *args) -> None: ...
88
```
89

90
**Usage Examples:**
91

92
```python
93
import httpx
94
import hishel
95
import asyncio
96

97
async def main():
98
    # Wrap existing async transport with caching
99
    base_transport = httpx.AsyncHTTPTransport()
100
    cache_transport = hishel.AsyncCacheTransport(
101
        transport=base_transport,
102
        storage=hishel.AsyncRedisStorage(),
103
        controller=hishel.Controller(allow_heuristics=True)
104
    )
105

106
    # Use directly with async httpx
107
    async with httpx.AsyncClient(transport=cache_transport) as client:
108
        response = await client.get("https://api.example.com/data")
109

110
    # Custom async transport configuration
111
    import ssl
112
    context = ssl.create_default_context()
113
    base_transport = httpx.AsyncHTTPTransport(
114
        verify=context,
115
        http2=True,
116
        limits=httpx.Limits(max_connections=100)
117
    )
118
    cache_transport = hishel.AsyncCacheTransport(transport=base_transport)
119

120
    async with httpx.AsyncClient(transport=cache_transport) as client:
121
        response = await client.get("https://api.example.com/data")
122

123
asyncio.run(main())
124
```
125

126
### Cache Streams
127

128
Stream implementations for serving cached response content efficiently.
129

130
```python { .api }
131
class CacheStream(httpx.SyncByteStream):
132
    def __init__(self, httpcore_stream: Iterable[bytes]):
133
        """
134
        Synchronous byte stream for cached responses.
135
        
136
        Parameters:
137
        - httpcore_stream: Iterable of bytes for response content
138
        """
139
    
140
    def __iter__(self) -> Iterator[bytes]:
141
        """Iterate over response content bytes"""
142
    
143
    def close(self) -> None:
144
        """Close the stream"""
145

146
class AsyncCacheStream(httpx.AsyncByteStream):
147
    def __init__(self, httpcore_stream: AsyncIterable[bytes]):
148
        """
149
        Asynchronous byte stream for cached responses.
150
        
151
        Parameters:
152
        - httpcore_stream: Async iterable of bytes for response content
153
        """
154
    
155
    async def __aiter__(self) -> AsyncIterator[bytes]:
156
        """Async iterate over response content bytes"""
157
    
158
    async def aclose(self) -> None:
159
        """Close the stream asynchronously"""
160
```
161

162
## Transport Integration
163

164
Cache transports seamlessly integrate with HTTPX's transport system:
165

166
### With Connection Pooling
167

168
```python
169
import httpx
170
import hishel
171

172
# HTTP/1.1 with connection pooling
173
base_transport = httpx.HTTPTransport(
174
    limits=httpx.Limits(
175
        max_keepalive_connections=20,
176
        max_connections=100,
177
        keepalive_expiry=30.0
178
    )
179
)
180
cache_transport = hishel.CacheTransport(transport=base_transport)
181

182
with httpx.Client(transport=cache_transport) as client:
183
    response = client.get("https://api.example.com/data")
184
```
185

186
### With HTTP/2 Support
187

188
```python
189
import httpx
190
import hishel
191

192
# HTTP/2 with caching
193
base_transport = httpx.AsyncHTTPTransport(http2=True)
194
cache_transport = hishel.AsyncCacheTransport(transport=base_transport)
195

196
async with httpx.AsyncClient(transport=cache_transport) as client:
197
    response = await client.get("https://api.example.com/data")
198
```
199

200
### With Custom Authentication
201

202
```python
203
import httpx
204
import hishel
205

206
# Custom auth with caching
207
auth = httpx.DigestAuth("username", "password")
208
base_transport = httpx.HTTPTransport()
209
cache_transport = hishel.CacheTransport(transport=base_transport)
210

211
with httpx.Client(transport=cache_transport, auth=auth) as client:
212
    response = client.get("https://api.example.com/protected")
213
```
214

215
### With Proxy Support
216

217
```python
218
import httpx
219
import hishel
220

221
# Proxy transport with caching
222
proxy_transport = httpx.HTTPTransport(proxy="http://proxy.example.com:8080")
223
cache_transport = hishel.CacheTransport(transport=proxy_transport)
224

225
with httpx.Client(transport=cache_transport) as client:
226
    response = client.get("https://api.example.com/data")
227
```
228

229
## Cache Transport Behavior
230

231
### Request Flow
232

233
1. **Cache Lookup**: Check if request has a valid cached response
234
2. **Freshness Check**: Validate cached response freshness using Controller
235
3. **Conditional Request**: Send validation request if needed (ETag, Last-Modified)
236
4. **Response Handling**: Store new responses or update cached responses
237
5. **Response Serving**: Return cached or fresh response to client
238

239
### Error Handling
240

241
Cache transports gracefully handle errors:
242

243
- **Storage Errors**: Fall back to network requests if cache storage fails
244
- **Network Errors**: Serve stale responses when allowed and available
245
- **Serialization Errors**: Skip caching and serve fresh responses
246
- **Validation Errors**: Treat as cache miss and fetch fresh response
247

248
### 504 Gateway Timeout
249

250
When network fails and no cached response is available:
251

252
```python
253
# Returns 504 status code when network is unavailable
254
# and no cached response exists
255
response = client.get("https://unreachable.example.com/data")
256
assert response.status_code == 504
257
```
258

259
## Advanced Usage
260

261
### Custom Transport Chains
262

263
```python
264
import httpx
265
import hishel
266

267
# Chain multiple transports
268
retry_transport = httpx.HTTPTransport(retries=3)
269
cache_transport = hishel.CacheTransport(transport=retry_transport)
270

271
# Custom middleware transport
272
class LoggingTransport(httpx.BaseTransport):
273
    def __init__(self, transport):
274
        self.transport = transport
275
    
276
    def handle_request(self, request):
277
        print(f"Requesting: {request.url}")
278
        response = self.transport.handle_request(request)
279
        print(f"Response: {response.status_code}")
280
        return response
281

282
logging_transport = LoggingTransport(cache_transport)
283

284
with httpx.Client(transport=logging_transport) as client:
285
    response = client.get("https://api.example.com/data")
286
```
287

288
### Manual Transport Usage
289

290
```python
291
import httpx
292
import hishel
293

294
# Use transport directly without client
295
transport = hishel.CacheTransport(
296
    transport=httpx.HTTPTransport(),
297
    storage=hishel.InMemoryStorage(),
298
    controller=hishel.Controller()
299
)
300

301
request = httpx.Request("GET", "https://api.example.com/data")
302
response = transport.handle_request(request)
303

304
transport.close()
305
```