0
# Utilities
1
2
Helper functions for header generation, multipart encoding, URL parsing, SSL configuration, and other common HTTP client tasks. These utilities simplify complex operations and provide building blocks for custom HTTP implementations.
3
4
## Capabilities
5
6
### Header Generation
7
8
Creates common HTTP headers with proper formatting and encoding.
9
10
```python { .api }
11
def make_headers(basic_auth=None, proxy_basic_auth=None, user_agent=None,
12
keep_alive=None, accept_encoding=None, disable_cache=None) -> dict:
13
"""
14
Generate common HTTP headers.
15
16
Parameters:
17
- basic_auth: Username and password for HTTP Basic Auth (tuple or string)
18
- proxy_basic_auth: Username and password for proxy Basic Auth (tuple or string)
19
- user_agent: User-Agent header value
20
- keep_alive: Whether to include Keep-Alive header (bool)
21
- accept_encoding: Accept-Encoding header value (bool or list)
22
- disable_cache: Whether to include cache-disabling headers (bool)
23
24
Returns:
25
dict: Dictionary of HTTP headers
26
"""
27
```
28
29
### Multipart Form Encoding
30
31
Encodes form data and files for multipart/form-data POST requests.
32
33
```python { .api }
34
def encode_multipart_formdata(fields, boundary=None) -> tuple[bytes, str]:
35
"""
36
Encode multipart form data for HTTP requests.
37
38
Parameters:
39
- fields: Sequence of (name, value) tuples or RequestField objects
40
- boundary: Multipart boundary string (auto-generated if None)
41
42
Returns:
43
tuple[bytes, str]: Encoded form data and content-type header value
44
"""
45
```
46
47
### Connection Pool Creation
48
49
Utilities for creating connection pools and managers from URLs.
50
51
```python { .api }
52
def connection_from_url(url: str, **kw) -> HTTPConnectionPool:
53
"""
54
Create connection pool from URL.
55
56
Parameters:
57
- url: Complete URL (scheme determines HTTP vs HTTPS)
58
- **kw: Additional arguments for connection pool
59
60
Returns:
61
HTTPConnectionPool: Configured connection pool for the URL's host
62
"""
63
64
def proxy_from_url(url: str, **kw) -> ProxyManager:
65
"""
66
Create ProxyManager from proxy URL.
67
68
Parameters:
69
- url: Proxy URL (e.g., 'http://proxy.example.com:8080')
70
- **kw: Additional arguments for ProxyManager
71
72
Returns:
73
ProxyManager: Configured proxy manager
74
"""
75
```
76
77
### URL Parsing and Validation
78
79
```python { .api }
80
def parse_url(url: str) -> Url:
81
"""
82
Parse URL string into components.
83
84
Parameters:
85
- url: URL string to parse
86
87
Returns:
88
Url: Named tuple with scheme, auth, host, port, path, query, fragment
89
"""
90
91
class Url:
92
"""URL components as named tuple"""
93
scheme: str # http, https, etc.
94
auth: str # username:password
95
host: str # hostname or IP
96
port: int # port number
97
path: str # path component
98
query: str # query string
99
fragment: str # fragment identifier
100
```
101
102
### SSL/TLS Utilities
103
104
```python { .api }
105
def create_urllib3_context(ssl_version=None, cert_reqs=None, options=None,
106
ciphers=None) -> ssl.SSLContext:
107
"""
108
Create SSL context with urllib3 defaults.
109
110
Parameters:
111
- ssl_version: SSL/TLS version to use
112
- cert_reqs: Certificate requirements (ssl.CERT_NONE, ssl.CERT_REQUIRED)
113
- options: SSL context options
114
- ciphers: Cipher suite string
115
116
Returns:
117
ssl.SSLContext: Configured SSL context
118
"""
119
120
def ssl_wrap_socket(sock, keyfile=None, certfile=None, cert_reqs=None,
121
ca_certs=None, server_hostname=None, ssl_version=None,
122
ciphers=None, ssl_context=None) -> ssl.SSLSocket:
123
"""
124
Wrap socket with SSL/TLS.
125
126
Returns:
127
ssl.SSLSocket: SSL-wrapped socket
128
"""
129
130
def assert_fingerprint(cert, fingerprint):
131
"""
132
Verify SSL certificate fingerprint.
133
134
Parameters:
135
- cert: Certificate in DER format
136
- fingerprint: Expected fingerprint (hex string)
137
138
Raises:
139
SSLError: If fingerprint doesn't match
140
"""
141
142
def resolve_cert_reqs(candidate) -> int:
143
"""
144
Resolve certificate requirements to ssl module constant.
145
146
Parameters:
147
- candidate: String ('CERT_NONE', 'CERT_REQUIRED') or ssl constant
148
149
Returns:
150
int: ssl.CERT_NONE or ssl.CERT_REQUIRED
151
"""
152
153
def resolve_ssl_version(candidate) -> int:
154
"""
155
Resolve SSL version to ssl module constant.
156
157
Parameters:
158
- candidate: String ('TLSv1', 'TLSv1_2') or ssl constant
159
160
Returns:
161
int: SSL version constant
162
"""
163
```
164
165
### Connection Utilities
166
167
```python { .api }
168
def is_connection_dropped(conn) -> bool:
169
"""
170
Check if connection has been dropped by the server.
171
172
Parameters:
173
- conn: Connection object to check
174
175
Returns:
176
bool: True if connection appears to be dropped
177
"""
178
179
def wait_for_read(sock, timeout=None) -> bool:
180
"""
181
Wait for socket to be ready for reading.
182
183
Parameters:
184
- sock: Socket object
185
- timeout: Timeout in seconds (None for no timeout)
186
187
Returns:
188
bool: True if socket is ready for reading
189
"""
190
191
def wait_for_write(sock, timeout=None) -> bool:
192
"""
193
Wait for socket to be ready for writing.
194
195
Parameters:
196
- sock: Socket object
197
- timeout: Timeout in seconds (None for no timeout)
198
199
Returns:
200
bool: True if socket is ready for writing
201
"""
202
```
203
204
### Form Field Utilities
205
206
```python { .api }
207
class RequestField:
208
def __init__(self, name, data, filename=None, headers=None):
209
"""
210
Represents a single multipart form field.
211
212
Parameters:
213
- name: Field name
214
- data: Field data (string, bytes, or file-like object)
215
- filename: Filename for file fields
216
- headers: Additional headers for this field
217
"""
218
219
def make_multipart(self, content_disposition=None, content_type=None,
220
content_location=None) -> RequestField:
221
"""Configure field for multipart encoding"""
222
223
def guess_content_type(filename, default='application/octet-stream') -> str:
224
"""
225
Guess MIME content type from filename.
226
227
Parameters:
228
- filename: Filename to analyze
229
- default: Default content type if guess fails
230
231
Returns:
232
str: MIME content type
233
"""
234
```
235
236
## Usage Examples
237
238
### Basic Header Generation
239
240
```python
241
import urllib3
242
243
# Basic authentication headers
244
headers = urllib3.make_headers(basic_auth='username:password')
245
print(headers) # {'authorization': 'Basic dXNlcm5hbWU6cGFzc3dvcmQ='}
246
247
# User agent header
248
headers = urllib3.make_headers(user_agent='MyApp/1.0')
249
print(headers) # {'user-agent': 'MyApp/1.0'}
250
251
# Multiple headers
252
headers = urllib3.make_headers(
253
basic_auth=('user', 'pass'),
254
user_agent='MyApp/1.0',
255
keep_alive=True,
256
accept_encoding=True
257
)
258
259
http = urllib3.PoolManager()
260
resp = http.request('GET', 'https://httpbin.org/headers', headers=headers)
261
```
262
263
### Multipart Form Data
264
265
```python
266
import urllib3
267
268
# Simple form fields
269
fields = [
270
('username', 'john'),
271
('password', 'secret'),
272
('remember', 'on')
273
]
274
275
body, content_type = urllib3.encode_multipart_formdata(fields)
276
277
http = urllib3.PoolManager()
278
resp = http.request('POST', 'https://httpbin.org/post',
279
body=body,
280
headers={'Content-Type': content_type})
281
```
282
283
### File Upload with Multipart
284
285
```python
286
import urllib3
287
288
# Upload file with form data
289
with open('/path/to/file.txt', 'rb') as f:
290
fields = [
291
('file', (f.name, f.read(), 'text/plain')),
292
('description', 'My uploaded file'),
293
('category', 'documents')
294
]
295
296
body, content_type = urllib3.encode_multipart_formdata(fields)
297
298
http = urllib3.PoolManager()
299
resp = http.request('POST', 'https://httpbin.org/post',
300
body=body,
301
headers={'Content-Type': content_type})
302
```
303
304
### Advanced File Upload with RequestField
305
306
```python
307
import urllib3
308
309
# Create custom request field
310
field = urllib3.fields.RequestField(
311
name='document',
312
data=open('/path/to/document.pdf', 'rb').read(),
313
filename='document.pdf'
314
)
315
field.make_multipart(content_type='application/pdf')
316
317
# Combine with other fields
318
fields = [
319
field,
320
('title', 'Important Document'),
321
('tags', 'important,pdf')
322
]
323
324
body, content_type = urllib3.encode_multipart_formdata(fields)
325
326
http = urllib3.PoolManager()
327
resp = http.request('POST', 'https://api.example.com/upload',
328
body=body,
329
headers={'Content-Type': content_type})
330
```
331
332
### URL Parsing
333
334
```python
335
import urllib3
336
337
# Parse complex URL
338
url = urllib3.util.parse_url('https://user:pass@api.example.com:8443/v1/data?format=json#results')
339
340
print(f"Scheme: {url.scheme}") # https
341
print(f"Auth: {url.auth}") # user:pass
342
print(f"Host: {url.host}") # api.example.com
343
print(f"Port: {url.port}") # 8443
344
print(f"Path: {url.path}") # /v1/data
345
print(f"Query: {url.query}") # format=json
346
print(f"Fragment: {url.fragment}") # results
347
348
# Create connection pool from parsed URL
349
pool = urllib3.connection_from_url(str(url))
350
```
351
352
### SSL Context Configuration
353
354
```python
355
import urllib3
356
import ssl
357
358
# Create custom SSL context
359
ssl_context = urllib3.util.create_urllib3_context(
360
ssl_version=ssl.PROTOCOL_TLS,
361
cert_reqs=ssl.CERT_REQUIRED,
362
ciphers='HIGH:!aNULL:!eNULL:!EXPORT:!DES:!RC4:!MD5:!PSK:!SRP:!CAMELLIA'
363
)
364
365
# Use with connection pool
366
pool = urllib3.HTTPSConnectionPool(
367
'secure.example.com',
368
ssl_context=ssl_context
369
)
370
371
resp = pool.urlopen('GET', '/secure-endpoint')
372
```
373
374
### Certificate Fingerprint Verification
375
376
```python
377
import urllib3
378
379
def verify_fingerprint_callback(conn, cert, errno, depth, ok):
380
"""Custom certificate verification with fingerprint check"""
381
if depth == 0: # Only check end-entity certificate
382
try:
383
urllib3.util.assert_fingerprint(cert, 'AA:BB:CC:DD:...')
384
return True
385
except urllib3.exceptions.SSLError:
386
return False
387
return ok
388
389
# This would be used in a custom SSL context setup
390
```
391
392
### Connection Testing
393
394
```python
395
import urllib3
396
import socket
397
398
# Test if connection is still alive
399
http = urllib3.PoolManager()
400
resp = http.request('GET', 'https://httpbin.org/get')
401
402
# Get underlying connection from response
403
conn = resp.connection
404
405
# Check if connection was dropped
406
if urllib3.util.is_connection_dropped(conn):
407
print("Connection was dropped by server")
408
else:
409
print("Connection is still alive")
410
411
resp.close()
412
```
413
414
### Response Utilities
415
416
```python { .api }
417
def is_fp_closed(obj) -> bool:
418
"""
419
Check if file-like object is closed.
420
421
Parameters:
422
- obj: File-like object to check
423
424
Returns:
425
bool: True if object is closed or unavailable
426
"""
427
```
428
429
### Proxy Configuration
430
431
```python
432
import urllib3
433
434
# Create proxy manager from URL
435
proxy = urllib3.proxy_from_url('http://proxy.example.com:8080')
436
437
# Or with authentication
438
proxy = urllib3.proxy_from_url('http://user:pass@proxy.example.com:8080')
439
440
# Use proxy for requests
441
resp = proxy.request('GET', 'https://httpbin.org/ip')
442
print(f"IP through proxy: {resp.json()}")
443
```
444
445
### Advanced Header Manipulation
446
447
```python
448
import urllib3
449
450
# Generate headers for different scenarios
451
auth_headers = urllib3.make_headers(basic_auth='admin:secret')
452
proxy_headers = urllib3.make_headers(proxy_basic_auth='proxy_user:proxy_pass')
453
ua_headers = urllib3.make_headers(user_agent='Mozilla/5.0 (Custom Bot)')
454
455
# Combine multiple header sets
456
combined_headers = {}
457
combined_headers.update(auth_headers)
458
combined_headers.update(ua_headers)
459
combined_headers.update({'Custom-Header': 'CustomValue'})
460
461
http = urllib3.PoolManager()
462
resp = http.request('GET', 'https://httpbin.org/headers',
463
headers=combined_headers)
464
```
465
466
### Content Type Guessing
467
468
```python
469
import urllib3
470
471
# Guess content type from filename
472
content_type = urllib3.fields.guess_content_type('document.pdf')
473
print(content_type) # application/pdf
474
475
content_type = urllib3.fields.guess_content_type('image.jpg')
476
print(content_type) # image/jpeg
477
478
content_type = urllib3.fields.guess_content_type('unknown.xyz')
479
print(content_type) # application/octet-stream
480
```
481
482
### Logging and Warning Utilities
483
484
```python { .api }
485
def add_stderr_logger(level: int = logging.DEBUG) -> logging.StreamHandler:
486
"""
487
Add a StreamHandler to the logger for debugging purposes.
488
489
Parameters:
490
- level: Logging level (default: logging.DEBUG)
491
492
Returns:
493
logging.StreamHandler: The added handler
494
"""
495
496
def disable_warnings(category: type[Warning] = exceptions.HTTPWarning) -> None:
497
"""
498
Helper for quickly disabling urllib3 warnings.
499
500
Parameters:
501
- category: Warning category to disable (default: HTTPWarning)
502
"""
503
```
504
505
#### Logging and Warning Examples
506
507
```python
508
import urllib3
509
import logging
510
511
# Enable debug logging
512
handler = urllib3.add_stderr_logger(level=logging.DEBUG)
513
514
# Make a request with debug output
515
http = urllib3.PoolManager()
516
resp = http.request('GET', 'https://httpbin.org/get')
517
518
# Disable specific warnings
519
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
520
521
# Disable all urllib3 warnings
522
urllib3.disable_warnings()
523
524
# Make request that would normally generate warnings
525
resp = http.request('GET', 'https://self-signed.badssl.com/', verify=False)
526
```
527
528
### Advanced RequestField Operations
529
530
```python { .api }
531
def format_multipart_header_param(name: str, value: str | bytes) -> str:
532
"""
533
Format and quote a single multipart header parameter.
534
535
Parameters:
536
- name: Parameter name (ASCII-only string)
537
- value: Parameter value (string or UTF-8 encoded bytes)
538
539
Returns:
540
str: Formatted parameter string 'name="value"' with proper escaping
541
"""
542
543
class RequestField:
544
@classmethod
545
def from_tuples(cls, fieldname: str, value, header_formatter=None) -> RequestField:
546
"""
547
Create RequestField from old-style tuple parameters.
548
549
Parameters:
550
- fieldname: Field name
551
- value: Field value (string, bytes, or tuple of (filename, data, mime_type))
552
- header_formatter: Custom header formatter function
553
554
Returns:
555
RequestField: Configured request field
556
"""
557
```
558
559
#### Advanced RequestField Examples
560
561
```python
562
import urllib3
563
564
# Create RequestField from tuple (filename, data, mime_type)
565
field = urllib3.fields.RequestField.from_tuples(
566
'document',
567
('report.pdf', open('report.pdf', 'rb').read(), 'application/pdf')
568
)
569
570
# Create RequestField from simple value
571
field = urllib3.fields.RequestField.from_tuples('username', 'john_doe')
572
573
# Custom multipart parameter formatting
574
formatted = urllib3.fields.format_multipart_header_param(
575
'filename', 'my document.txt'
576
)
577
print(formatted) # filename="my%20document.txt"
578
579
# Use in multipart form data
580
fields = [
581
urllib3.fields.RequestField.from_tuples('file', ('data.csv', csv_data, 'text/csv')),
582
urllib3.fields.RequestField.from_tuples('description', 'Monthly report')
583
]
584
585
body, content_type = urllib3.encode_multipart_formdata(fields)
586
```
587
588
## Utility Constants
589
590
```python
591
# Skip header generation (sentinel value)
592
SKIP_HEADER = object()
593
594
# Headers that can be skipped in make_headers
595
SKIPPABLE_HEADERS = frozenset([
596
'accept-encoding', 'host', 'user-agent'
597
])
598
599
# SSL/TLS availability
600
IS_PYOPENSSL = bool(...) # True if PyOpenSSL is available
601
HAS_SNI = bool(...) # True if SNI support is available
602
603
# Default SSL protocols
604
ALPN_PROTOCOLS = ['h2', 'http/1.1'] # Supported ALPN protocols
605
```