0
# Transport Adapters
1

2
Transport adapters handle the actual HTTP communication, providing the interface between requests and underlying HTTP libraries. Adapters enable connection pooling, SSL/TLS handling, and protocol-specific optimizations.
3

4
## Capabilities
5

6
### BaseAdapter Class
7

8
Abstract base class that all transport adapters inherit from.
9

10
```python { .api }
11
class BaseAdapter:
12
    """
13
    The base adapter class for transports.
14
    
15
    All transport adapters should inherit from this class.
16
    """
17

18
    def __init__(self):
19
        """Initialize the adapter."""
20

21
    def send(self, request, stream=False, timeout=None, verify=True, 
22
             cert=None, proxies=None) -> 'Response':
23
        """
24
        Send a PreparedRequest and return a Response.
25
        
26
        Parameters:
27
        - request: PreparedRequest object to send
28
        - stream: Whether to stream the response content
29
        - timeout: Timeout value in seconds
30
        - verify: SSL verification setting
31
        - cert: Client certificate
32
        - proxies: Proxy configuration
33
        
34
        Returns:
35
        Response object
36
        
37
        Raises:
38
        NotImplementedError: Must be implemented by subclasses
39
        """
40

41
    def close(self):
42
        """
43
        Clean up adapter resources.
44
        
45
        Called when session is closed.
46
        """
47
```
48

49
### HTTPAdapter Class
50

51
Built-in HTTP/HTTPS adapter using urllib3 for connection pooling and transport.
52

53
```python { .api }
54
class HTTPAdapter(BaseAdapter):
55
    """
56
    Built-in HTTP adapter using urllib3.
57
    
58
    Provides connection pooling, SSL/TLS handling, and HTTP protocol support.
59
    """
60

61
    def __init__(self, pool_connections=10, pool_maxsize=10, max_retries=0, 
62
                 pool_block=False):
63
        """
64
        Initialize HTTPAdapter.
65
        
66
        Parameters:
67
        - pool_connections: Number of urllib3 connection pools to cache
68
        - pool_maxsize: Maximum connections per pool
69
        - max_retries: Maximum number of retries per request
70
        - pool_block: Whether to block when pool is full
71
        """
72

73
    def send(self, request, stream=False, timeout=None, verify=True, 
74
             cert=None, proxies=None) -> 'Response':
75
        """
76
        Send a request using urllib3.
77
        
78
        Parameters:
79
        - request: PreparedRequest to send
80
        - stream: Stream response content
81
        - timeout: Request timeout (connect, read) tuple or single value
82
        - verify: SSL certificate verification (bool or CA bundle path)
83
        - cert: Client certificate (path or (cert, key) tuple)
84
        - proxies: Proxy configuration dict
85
        
86
        Returns:
87
        Response object
88
        """
89

90
    def close(self):
91
        """Close all pooled connections."""
92

93
    # Connection management
94
    def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
95
        """
96
        Initialize urllib3 PoolManager.
97
        
98
        Parameters:
99
        - connections: Number of connection pools
100
        - maxsize: Maximum connections per pool
101
        - block: Whether to block when pool is full
102
        - **pool_kwargs: Additional pool arguments
103
        """
104

105
    def get_connection_with_tls_context(self, request, verify, proxies=None, cert=None):
106
        """
107
        Get connection pool with TLS context.
108
        
109
        Parameters:
110
        - request: Request object
111
        - verify: SSL verification setting
112
        - proxies: Proxy configuration
113
        - cert: Client certificate
114
        
115
        Returns:
116
        ConnectionPool instance
117
        """
118

119
    def get_connection(self, url, proxies=None):
120
        """
121
        DEPRECATED: Get connection for URL.
122
        
123
        Parameters:
124
        - url: URL to get connection for
125
        - proxies: Proxy configuration
126
        
127
        Returns:
128
        ConnectionPool instance
129
        """
130

131
    def proxy_manager_for(self, proxy, **proxy_kwargs):
132
        """
133
        Get ProxyManager for proxy URL.
134
        
135
        Parameters:
136
        - proxy: Proxy URL
137
        - **proxy_kwargs: Additional proxy arguments
138
        
139
        Returns:
140
        ProxyManager instance
141
        """
142

143
    # Request/Response processing
144
    def build_response(self, req, resp) -> 'Response':
145
        """
146
        Build Response object from urllib3 response.
147
        
148
        Parameters:
149
        - req: PreparedRequest object
150
        - resp: urllib3 HTTPResponse object
151
        
152
        Returns:
153
        Response object
154
        """
155

156
    def request_url(self, request, proxies) -> str:
157
        """
158
        Get the URL to use for the request.
159
        
160
        Parameters:
161
        - request: PreparedRequest object
162
        - proxies: Proxy configuration
163
        
164
        Returns:
165
        URL string to use
166
        """
167

168
    def add_headers(self, request, **kwargs):
169
        """
170
        Add headers to the request.
171
        
172
        Parameters:
173
        - request: PreparedRequest object
174
        - **kwargs: Additional arguments
175
        """
176

177
    def proxy_headers(self, proxy) -> dict:
178
        """
179
        Get headers to add for proxy requests.
180
        
181
        Parameters:
182
        - proxy: Proxy URL
183
        
184
        Returns:
185
        Dict of headers
186
        """
187

188
    # SSL/TLS handling
189
    def cert_verify(self, conn, url, verify, cert):
190
        """
191
        Verify SSL certificates and configure client certs.
192
        
193
        Parameters:
194
        - conn: Connection object
195
        - url: Request URL
196
        - verify: SSL verification setting
197
        - cert: Client certificate
198
        """
199

200
    def build_connection_pool_key_attributes(self, request, verify, cert=None) -> tuple:
201
        """
202
        Build key attributes for connection pooling.
203
        
204
        Parameters:
205
        - request: PreparedRequest object
206
        - verify: SSL verification setting
207
        - cert: Client certificate
208
        
209
        Returns:
210
        Tuple of (pool_kwargs, connection_pool_kwargs)
211
        """
212
```
213

214
### Adapter Constants
215

216
```python { .api }
217
DEFAULT_POOLBLOCK: bool     # False
218
DEFAULT_POOLSIZE: int       # 10
219
DEFAULT_RETRIES: int        # 0
220
DEFAULT_POOL_TIMEOUT: None  # None
221
```
222

223
## Usage Examples
224

225
### Basic Adapter Usage
226

227
```python
228
import requests
229
from requests.adapters import HTTPAdapter
230

231
# Adapters are used automatically
232
response = requests.get('https://httpbin.org/get')
233
print(f"Status: {response.status_code}")
234

235
# Access session adapters
236
session = requests.Session()
237
print("Mounted adapters:")
238
for prefix, adapter in session.adapters.items():
239
    print(f"  {prefix}: {adapter}")
240
```
241

242
### Custom Adapter Configuration
243

244
```python
245
import requests
246
from requests.adapters import HTTPAdapter
247

248
# Create adapter with custom settings
249
adapter = HTTPAdapter(
250
    pool_connections=20,    # More connection pools
251
    pool_maxsize=50,       # More connections per pool
252
    max_retries=3,         # Retry failed requests
253
    pool_block=True        # Wait when pool is full
254
)
255

256
# Mount adapter to session
257
session = requests.Session()
258
session.mount('https://', adapter)
259
session.mount('http://', adapter)
260

261
# Requests will use the custom adapter
262
response = session.get('https://httpbin.org/get')
263
```
264

265
### Protocol-Specific Adapters
266

267
```python
268
import requests
269
from requests.adapters import HTTPAdapter
270

271
# Different adapters for different hosts
272
class APIAdapter(HTTPAdapter):
273
    """Custom adapter for API endpoints."""
274
    
275
    def __init__(self, api_key, **kwargs):
276
        self.api_key = api_key
277
        super().__init__(**kwargs)
278
    
279
    def add_headers(self, request, **kwargs):
280
        request.headers['Authorization'] = f'Bearer {self.api_key}'
281
        super().add_headers(request, **kwargs)
282

283
# Mount custom adapter
284
session = requests.Session()
285
api_adapter = APIAdapter(api_key='your-api-key', max_retries=3)
286
session.mount('https://api.example.com/', api_adapter)
287

288
# Regular adapter for other URLs
289
session.mount('https://', HTTPAdapter(max_retries=1))
290

291
# Different adapters used based on URL
292
api_response = session.get('https://api.example.com/data')      # Uses APIAdapter
293
web_response = session.get('https://other-site.com/page')      # Uses HTTPAdapter
294
```
295

296
### Connection Pool Management
297

298
```python
299
import requests
300
from requests.adapters import HTTPAdapter
301

302
# Configure connection pooling
303
adapter = HTTPAdapter(
304
    pool_connections=10,    # 10 connection pools
305
    pool_maxsize=100,      # 100 connections per pool
306
    pool_block=False       # Don't block when pool full
307
)
308

309
session = requests.Session()
310
session.mount('https://', adapter)
311

312
# Make many requests - connections are pooled and reused
313
urls = [f'https://httpbin.org/delay/{i}' for i in range(5)]
314

315
for url in urls:
316
    response = session.get(url)
317
    print(f"Response from {url}: {response.status_code}")
318

319
# Close session to clean up connection pools
320
session.close()
321
```
322

323
### SSL/TLS Configuration
324

325
```python
326
import requests
327
from requests.adapters import HTTPAdapter
328

329
# Custom adapter with SSL settings
330
class SecureAdapter(HTTPAdapter):
331
    def cert_verify(self, conn, url, verify, cert):
332
        # Custom SSL verification logic
333
        super().cert_verify(conn, url, verify, cert)
334
        print(f"SSL verification for {url}: verify={verify}")
335

336
# Use secure adapter
337
session = requests.Session()
338
session.mount('https://', SecureAdapter())
339

340
# Configure SSL verification
341
response = session.get('https://httpbin.org/get', 
342
                      verify=True,  # Verify SSL certificates
343
                      cert=None)    # No client certificate
344
```
345

346
### Retry Configuration
347

348
```python
349
import requests
350
from requests.adapters import HTTPAdapter
351
from urllib3.util.retry import Retry
352

353
# Configure retry strategy
354
retry_strategy = Retry(
355
    total=3,                    # Total retries
356
    status_forcelist=[429, 500, 502, 503, 504],  # Status codes to retry
357
    method_whitelist=["HEAD", "GET", "OPTIONS"], # Methods to retry
358
    backoff_factor=1           # Backoff between retries
359
)
360

361
# Create adapter with retry strategy
362
adapter = HTTPAdapter(max_retries=retry_strategy)
363

364
session = requests.Session()
365
session.mount('http://', adapter)
366
session.mount('https://', adapter)
367

368
# Requests will automatically retry on failures
369
response = session.get('https://httpbin.org/status/500')
370
```
371

372
### Custom Transport Adapter
373

374
```python
375
import requests
376
from requests.adapters import BaseAdapter
377
from requests.models import Response
378

379
class MockAdapter(BaseAdapter):
380
    """Mock adapter for testing."""
381
    
382
    def __init__(self, responses=None):
383
        super().__init__()
384
        self.responses = responses or {}
385
    
386
    def send(self, request, **kwargs):
387
        """Return mock response."""
388
        response = Response()
389
        response.status_code = 200
390
        response.headers['Content-Type'] = 'application/json'
391
        
392
        # Mock response based on URL
393
        if request.url in self.responses:
394
            response._content = self.responses[request.url].encode('utf-8')
395
        else:
396
            response._content = b'{"mock": "response"}'
397
        
398
        response.url = request.url
399
        response.request = request
400
        return response
401
    
402
    def close(self):
403
        pass
404

405
# Use mock adapter for testing
406
mock_responses = {
407
    'https://api.example.com/data': '{"users": [{"name": "John"}]}'
408
}
409

410
session = requests.Session()
411
session.mount('https://api.example.com/', MockAdapter(mock_responses))
412

413
# This returns the mock response
414
response = session.get('https://api.example.com/data')
415
print(response.json())  # {'users': [{'name': 'John'}]}
416
```
417

418
### Proxy Adapter Configuration
419

420
```python
421
import requests
422
from requests.adapters import HTTPAdapter
423

424
class ProxyAdapter(HTTPAdapter):
425
    """Adapter with proxy configuration."""
426
    
427
    def __init__(self, proxy_url, **kwargs):
428
        self.proxy_url = proxy_url
429
        super().__init__(**kwargs)
430
    
431
    def proxy_headers(self, proxy):
432
        """Add custom proxy headers."""
433
        headers = super().proxy_headers(proxy)
434
        headers['Proxy-Authorization'] = 'Basic dXNlcjpwYXNz'  # base64 user:pass
435
        return headers
436

437
# Configure proxy adapter
438
proxy_adapter = ProxyAdapter('http://proxy.example.com:8080')
439

440
session = requests.Session()
441
session.mount('http://', proxy_adapter)
442
session.mount('https://', proxy_adapter)
443

444
# Requests go through the proxy
445
response = session.get('https://httpbin.org/ip')
446
print(response.json())  # Shows proxy IP
447
```
448

449
### Performance Monitoring Adapter
450

451
```python
452
import requests
453
import time
454
from requests.adapters import HTTPAdapter
455

456
class TimingAdapter(HTTPAdapter):
457
    """Adapter that measures request timing."""
458
    
459
    def send(self, request, **kwargs):
460
        start_time = time.time()
461
        response = super().send(request, **kwargs)
462
        end_time = time.time()
463
        
464
        # Add timing information to response
465
        response.elapsed_total = end_time - start_time
466
        print(f"Request to {request.url} took {response.elapsed_total:.3f}s")
467
        
468
        return response
469

470
# Use timing adapter
471
session = requests.Session()
472
session.mount('https://', TimingAdapter())
473

474
response = session.get('https://httpbin.org/delay/2')
475
print(f"Total time: {response.elapsed_total:.3f}s")
476
```
477

478
### Adapter Debugging
479

480
```python
481
import requests
482
from requests.adapters import HTTPAdapter
483

484
class DebugAdapter(HTTPAdapter):
485
    """Adapter with detailed logging."""
486
    
487
    def send(self, request, **kwargs):
488
        print(f"Sending {request.method} request to {request.url}")
489
        print(f"Headers: {dict(request.headers)}")
490
        
491
        if request.body:
492
            print(f"Body: {request.body[:100]}...")
493
        
494
        response = super().send(request, **kwargs)
495
        
496
        print(f"Received {response.status_code} response")
497
        print(f"Response headers: {dict(response.headers)}")
498
        
499
        return response
500

501
# Use debug adapter
502
session = requests.Session()
503
session.mount('https://', DebugAdapter())
504

505
response = session.post('https://httpbin.org/post', 
506
                       json={'key': 'value'})
507
```
508

509
## Adapter Best Practices
510

511
### Resource Management
512

513
```python
514
import requests
515
from requests.adapters import HTTPAdapter
516

517
# Always close sessions to clean up connection pools
518
session = requests.Session()
519
adapter = HTTPAdapter(pool_connections=10, pool_maxsize=50)
520
session.mount('https://', adapter)
521

522
try:
523
    # Use session
524
    response = session.get('https://api.example.com/data')
525
    # Process response...
526
finally:
527
    # Ensure cleanup
528
    session.close()
529

530
# Or use context manager
531
with requests.Session() as session:
532
    adapter = HTTPAdapter(pool_connections=10, pool_maxsize=50)
533
    session.mount('https://', adapter)
534
    response = session.get('https://api.example.com/data')
535
# Session automatically closed
536
```
537

538
### Adapter Selection
539

540
```python
541
import requests
542
from requests.adapters import HTTPAdapter
543

544
# Different configurations for different services
545
session = requests.Session()
546

547
# High-performance adapter for API calls
548
api_adapter = HTTPAdapter(
549
    pool_connections=20,
550
    pool_maxsize=100,
551
    max_retries=3
552
)
553

554
# Conservative adapter for file downloads
555
download_adapter = HTTPAdapter(
556
    pool_connections=5,
557
    pool_maxsize=10,
558
    max_retries=1
559
)
560

561
# Mount adapters with specific prefixes
562
session.mount('https://api.fastservice.com/', api_adapter)
563
session.mount('https://downloads.example.com/', download_adapter)
564
session.mount('https://', HTTPAdapter())  # Default adapter
565

566
# Requests automatically use appropriate adapter
567
api_response = session.get('https://api.fastservice.com/data')
568
file_response = session.get('https://downloads.example.com/file.zip')
569
other_response = session.get('https://other-site.com/page')
570
```