0
# Advanced Features
1
2
Configuration classes, status code utilities, and advanced networking options for fine-tuning HTTP behavior and handling complex scenarios. These features provide granular control over request behavior, timeout handling, retry logic, and status code management.
3
4
## Capabilities
5
6
### Configuration Classes
7
8
Advanced configuration classes for fine-tuning request behavior.
9
10
```python { .api }
11
class TimeoutConfiguration:
12
"""
13
Configuration for request timeouts.
14
15
Provides detailed control over connection and read timeouts,
16
allowing separate configuration for different phases of the request.
17
"""
18
19
def __init__(
20
self,
21
connect: float | None = None,
22
read: float | None = None,
23
total: float | None = None,
24
pool: float | None = None
25
):
26
"""
27
Initialize timeout configuration.
28
29
Args:
30
connect: Timeout for establishing connection (seconds)
31
read: Timeout for reading response data (seconds)
32
total: Total timeout for entire request (seconds)
33
pool: Timeout for getting connection from pool (seconds)
34
"""
35
36
@property
37
def connect_timeout(self) -> float | None:
38
"""Connection establishment timeout."""
39
40
@property
41
def read_timeout(self) -> float | None:
42
"""Response reading timeout."""
43
44
class RetryConfiguration:
45
"""
46
Configuration for request retry behavior.
47
48
Provides sophisticated retry logic with backoff strategies,
49
status code filtering, and method-specific retry policies.
50
"""
51
52
def __init__(
53
self,
54
total: int = 3,
55
connect: int | None = None,
56
read: int | None = None,
57
redirect: int | None = None,
58
status: int | None = None,
59
other: int | None = None,
60
allowed_methods: frozenset[str] | None = None,
61
status_forcelist: frozenset[int] | None = None,
62
backoff_factor: float = 0,
63
backoff_max: float = 120,
64
raise_on_redirect: bool = True,
65
raise_on_status: bool = True,
66
history: tuple | None = None,
67
respect_retry_after_header: bool = True,
68
remove_headers_on_redirect: frozenset[str] | None = None,
69
):
70
"""
71
Initialize retry configuration.
72
73
Args:
74
total: Total number of retries
75
connect: Retries for connection errors
76
read: Retries for read errors
77
redirect: Retries for redirect responses
78
status: Retries for specific status codes
79
other: Retries for other errors
80
allowed_methods: HTTP methods that can be retried
81
status_forcelist: HTTP status codes to force retry
82
backoff_factor: Backoff multiplication factor
83
backoff_max: Maximum backoff time
84
raise_on_redirect: Raise exception on redirect failures
85
raise_on_status: Raise exception on status failures
86
history: Retry history tracking
87
respect_retry_after_header: Honor Retry-After header
88
remove_headers_on_redirect: Headers to remove on redirect
89
"""
90
91
def new(self, **kwargs) -> RetryConfiguration:
92
"""
93
Create new RetryConfiguration with updated parameters.
94
95
Args:
96
**kwargs: Parameters to update
97
98
Returns:
99
New RetryConfiguration instance
100
"""
101
102
def get_backoff_time(self) -> float:
103
"""
104
Calculate backoff time for next retry.
105
106
Returns:
107
Backoff time in seconds
108
"""
109
```
110
111
### Status Code Utilities
112
113
Comprehensive status code lookup and management utilities.
114
115
```python { .api }
116
codes: LookupDict
117
"""
118
Dictionary-like object mapping HTTP status names to status codes.
119
120
Provides multiple access patterns for HTTP status codes:
121
- Attribute access: codes.ok, codes.not_found
122
- Dictionary access: codes['ok'], codes['not_found']
123
- Case-insensitive access: codes.OK, codes.Not_Found
124
- Special aliases: codes['\o/'] for 200
125
126
Common status codes:
127
- codes.ok = 200
128
- codes.created = 201
129
- codes.accepted = 202
130
- codes.no_content = 204
131
- codes.moved_permanently = 301
132
- codes.found = 302
133
- codes.not_modified = 304
134
- codes.temporary_redirect = 307
135
- codes.permanent_redirect = 308
136
- codes.bad_request = 400
137
- codes.unauthorized = 401
138
- codes.forbidden = 403
139
- codes.not_found = 404
140
- codes.method_not_allowed = 405
141
- codes.conflict = 409
142
- codes.gone = 410
143
- codes.unprocessable_entity = 422
144
- codes.too_many_requests = 429
145
- codes.internal_server_error = 500
146
- codes.bad_gateway = 502
147
- codes.service_unavailable = 503
148
- codes.gateway_timeout = 504
149
"""
150
151
class LookupDict(dict):
152
"""
153
Dictionary-like object for status code lookups.
154
155
Supports multiple access patterns and case-insensitive lookups.
156
"""
157
158
def __getitem__(self, key: str | int) -> int:
159
"""Get status code by name or return key if it's already a code."""
160
161
def get(self, key: str | int, default: int | None = None) -> int | None:
162
"""Get status code with default fallback."""
163
```
164
165
### Utility Functions
166
167
The utils module provides various utility functions for URL handling, encoding, authentication, and more.
168
169
```python { .api }
170
# URL and encoding utilities
171
def requote_uri(uri: str) -> str:
172
"""Re-quote the given URI with safe characters."""
173
174
def quote(s: str, safe: str = '') -> str:
175
"""Quote special characters in string."""
176
177
def unquote(s: str) -> str:
178
"""Unquote percent-encoded characters in string."""
179
180
def urldefrag(url: str) -> tuple[str, str]:
181
"""Remove fragment from URL."""
182
183
# Authentication utilities
184
def get_auth_from_url(url: str) -> tuple[str, str] | None:
185
"""Extract authentication credentials from URL."""
186
187
def get_netrc_auth(url: str, raise_errors: bool = False) -> tuple[str, str] | None:
188
"""Get authentication from .netrc file."""
189
190
# Proxy utilities
191
def get_environ_proxies(url: str, no_proxy: str | None = None) -> dict[str, str]:
192
"""Get proxy configuration from environment variables."""
193
194
def should_bypass_proxies(url: str, no_proxy: str | None = None) -> bool:
195
"""Check if URL should bypass proxy configuration."""
196
197
def resolve_proxies(request: PreparedRequest, proxies: dict, trust_env: bool = True) -> dict:
198
"""Resolve proxy configuration for a request."""
199
200
# Header utilities
201
def default_headers() -> CaseInsensitiveDict:
202
"""Get default headers for requests."""
203
204
def default_user_agent(name: str = "niquests") -> str:
205
"""Get default User-Agent string."""
206
207
# Content utilities
208
def stream_decode_response_unicode(iterator: Iterator[bytes], encoding: str) -> Iterator[str]:
209
"""Decode response content as unicode stream."""
210
211
def get_encoding_from_headers(headers: HeadersType) -> str | None:
212
"""Extract encoding from Content-Type header."""
213
214
# Certificate and security utilities
215
def is_ocsp_capable() -> bool:
216
"""Check if OCSP certificate verification is available."""
217
218
def is_crl_capable() -> bool:
219
"""Check if CRL certificate verification is available."""
220
221
# Data structure utilities
222
def to_key_val_list(value: dict | list | None) -> list[tuple[str, str]]:
223
"""Convert various data types to key-value pair list."""
224
```
225
226
### Legacy Compatibility
227
228
Compatibility flags and utilities for working with different urllib3 versions.
229
230
```python { .api }
231
HAS_LEGACY_URLLIB3: bool
232
"""
233
Boolean flag indicating if legacy urllib3 is being used.
234
235
This flag helps determine which features and behaviors are available
236
based on the urllib3 version. Some advanced features may not be
237
available with older urllib3 versions.
238
239
Usage:
240
if HAS_LEGACY_URLLIB3:
241
# Use legacy behavior
242
pass
243
else:
244
# Use modern features
245
pass
246
"""
247
```
248
249
### Package Metadata
250
251
Version and build information for the niquests package.
252
253
```python { .api }
254
__version__: str # Package version (e.g., "3.15.2")
255
__title__: str # Package title ("niquests")
256
__description__: str # Package description
257
__url__: str # Package homepage URL
258
__author__: str # Author name
259
__author_email__: str # Author email address
260
__license__: str # License type ("Apache 2.0")
261
__copyright__: str # Copyright notice
262
__build__: str # Build information
263
__cake__: str # Special build marker
264
```
265
266
## Usage Examples
267
268
### Advanced Timeout Configuration
269
270
```python
271
import niquests
272
from niquests import TimeoutConfiguration
273
274
# Simple timeout (applies to both connect and read)
275
response = niquests.get('https://api.example.com/data', timeout=10.0)
276
277
# Separate connect and read timeouts
278
timeout_config = TimeoutConfiguration(connect=5.0, read=30.0)
279
response = niquests.get('https://api.example.com/data', timeout=timeout_config)
280
281
# Total timeout with separate phases
282
timeout_config = TimeoutConfiguration(
283
connect=5.0, # 5 seconds to establish connection
284
read=30.0, # 30 seconds to read response
285
total=60.0 # Total request must complete within 60 seconds
286
)
287
288
# Use with session for persistent configuration
289
with niquests.Session() as session:
290
session.timeout = timeout_config
291
response1 = session.get('https://api.example.com/endpoint1')
292
response2 = session.get('https://api.example.com/endpoint2')
293
```
294
295
### Advanced Retry Configuration
296
297
```python
298
import niquests
299
from niquests import RetryConfiguration
300
301
# Simple retry configuration
302
retry_config = RetryConfiguration(total=5)
303
response = niquests.get('https://api.example.com/data', retries=retry_config)
304
305
# Advanced retry with backoff
306
retry_config = RetryConfiguration(
307
total=3, # Maximum 3 retries
308
backoff_factor=0.5, # Wait 0.5, 1.0, 2.0 seconds between retries
309
backoff_max=10.0, # Maximum 10 seconds backoff
310
status_forcelist=[500, 502, 503, 504], # Retry on server errors
311
allowed_methods=['GET', 'POST'], # Only retry safe methods
312
respect_retry_after_header=True # Honor server's Retry-After header
313
)
314
315
# Use with session
316
with niquests.Session() as session:
317
session.retries = retry_config
318
319
try:
320
response = session.get('https://unreliable-api.example.com/data')
321
print("Success after retries:", response.json())
322
except niquests.RequestException as e:
323
print("Failed after all retries:", e)
324
325
# Method-specific retry configuration
326
retry_config = RetryConfiguration(
327
total=5,
328
connect=2, # Only 2 retries for connection errors
329
read=3, # 3 retries for read timeouts
330
status=4 # 4 retries for HTTP status errors
331
)
332
```
333
334
### Status Code Handling
335
336
```python
337
import niquests
338
339
response = niquests.get('https://api.example.com/data')
340
341
# Using status code constants
342
if response.status_code == niquests.codes.ok:
343
print("Request successful")
344
elif response.status_code == niquests.codes.not_found:
345
print("Resource not found")
346
elif response.status_code == niquests.codes.unauthorized:
347
print("Authentication required")
348
elif response.status_code >= niquests.codes.internal_server_error:
349
print("Server error")
350
351
# Alternative access patterns
352
if response.status_code == niquests.codes['ok']:
353
print("Success")
354
355
if response.status_code == niquests.codes.OK: # Case insensitive
356
print("Success")
357
358
# Status code ranges
359
def categorize_status(status_code):
360
if 200 <= status_code < 300:
361
return "success"
362
elif 300 <= status_code < 400:
363
return "redirect"
364
elif 400 <= status_code < 500:
365
return "client_error"
366
elif 500 <= status_code < 600:
367
return "server_error"
368
else:
369
return "unknown"
370
371
category = categorize_status(response.status_code)
372
print(f"Response category: {category}")
373
374
# Common status code checks
375
success_codes = [
376
niquests.codes.ok,
377
niquests.codes.created,
378
niquests.codes.accepted,
379
niquests.codes.no_content
380
]
381
382
if response.status_code in success_codes:
383
print("Operation successful")
384
```
385
386
### Utility Functions Usage
387
388
```python
389
import niquests
390
from niquests import utils
391
392
# URL manipulation
393
original_url = "https://example.com/path with spaces"
394
safe_url = utils.requote_uri(original_url)
395
print(f"Safe URL: {safe_url}")
396
397
# Extract auth from URL
398
auth_url = "https://user:pass@api.example.com/data"
399
auth = utils.get_auth_from_url(auth_url)
400
if auth:
401
username, password = auth
402
print(f"Extracted auth: {username}")
403
404
# Environment proxy detection
405
url = "https://api.example.com/data"
406
proxies = utils.get_environ_proxies(url)
407
if proxies:
408
print(f"Using proxies: {proxies}")
409
410
# Check if should bypass proxies
411
if utils.should_bypass_proxies(url):
412
print("Bypassing proxy for this URL")
413
414
# Default headers
415
default_headers = utils.default_headers()
416
print(f"Default headers: {dict(default_headers)}")
417
418
# Custom User-Agent
419
user_agent = utils.default_user_agent("MyApp")
420
print(f"User-Agent: {user_agent}")
421
```
422
423
### Feature Detection
424
425
```python
426
import niquests
427
428
# Check advanced security features
429
if niquests.is_ocsp_capable():
430
print("OCSP certificate verification available")
431
# Can use OCSP-based certificate verification
432
else:
433
print("OCSP not available, using standard verification")
434
435
if niquests.is_crl_capable():
436
print("CRL certificate verification available")
437
# Can use Certificate Revocation Lists
438
else:
439
print("CRL not available")
440
441
# Check urllib3 compatibility
442
if niquests.HAS_LEGACY_URLLIB3:
443
print("Using legacy urllib3 - some features may be limited")
444
# Adjust behavior for legacy compatibility
445
else:
446
print("Using modern urllib3 - all features available")
447
# Can use all advanced features
448
449
# Package information
450
print(f"Niquests version: {niquests.__version__}")
451
print(f"Build info: {niquests.__build__}")
452
```
453
454
### Advanced Session Configuration
455
456
```python
457
import niquests
458
from niquests import TimeoutConfiguration, RetryConfiguration
459
460
# Comprehensive session configuration
461
class APISession(niquests.Session):
462
"""Custom session with advanced configuration."""
463
464
def __init__(self, base_url, api_key=None):
465
super().__init__()
466
467
self.base_url = base_url
468
469
# Configure timeouts
470
self.timeout = TimeoutConfiguration(
471
connect=5.0,
472
read=30.0,
473
total=60.0
474
)
475
476
# Configure retries
477
self.retries = RetryConfiguration(
478
total=3,
479
backoff_factor=0.5,
480
status_forcelist=[500, 502, 503, 504],
481
allowed_methods=['GET', 'POST', 'PUT', 'PATCH', 'DELETE']
482
)
483
484
# Set default headers
485
self.headers.update({
486
'User-Agent': niquests.utils.default_user_agent('MyAPIClient/1.0'),
487
'Accept': 'application/json',
488
'Content-Type': 'application/json'
489
})
490
491
# Set API key if provided
492
if api_key:
493
self.headers['Authorization'] = f'Bearer {api_key}'
494
495
def request(self, method, endpoint, **kwargs):
496
"""Make request with automatic URL building."""
497
url = f"{self.base_url.rstrip('/')}/{endpoint.lstrip('/')}"
498
return super().request(method, url, **kwargs)
499
500
# Usage
501
with APISession('https://api.example.com', api_key='secret') as session:
502
# All requests will use the configured timeouts, retries, and headers
503
users = session.get('/users').json()
504
505
new_user = session.post('/users', json={
506
'name': 'John Doe',
507
'email': 'john@example.com'
508
}).json()
509
510
updated_user = session.put(f'/users/{new_user["id"]}', json={
511
'name': 'John Smith'
512
}).json()
513
```
514
515
### Custom Retry Logic
516
517
```python
518
import time
519
import niquests
520
from niquests import RetryConfiguration
521
522
def custom_retry_request(url, max_retries=3, backoff_base=2):
523
"""Custom retry logic with exponential backoff."""
524
525
last_exception = None
526
527
for attempt in range(max_retries + 1):
528
try:
529
response = niquests.get(url, timeout=10.0)
530
531
# Check for retryable status codes
532
if response.status_code in [500, 502, 503, 504]:
533
if attempt < max_retries:
534
wait_time = backoff_base ** attempt
535
print(f"Server error {response.status_code}, retrying in {wait_time}s...")
536
time.sleep(wait_time)
537
continue
538
else:
539
response.raise_for_status()
540
541
# Success or non-retryable error
542
return response
543
544
except (niquests.ConnectionError, niquests.Timeout) as e:
545
last_exception = e
546
547
if attempt < max_retries:
548
wait_time = backoff_base ** attempt
549
print(f"Network error, retrying in {wait_time}s...")
550
time.sleep(wait_time)
551
else:
552
raise
553
554
# If we get here, all retries were exhausted
555
raise last_exception
556
557
# Usage
558
try:
559
response = custom_retry_request('https://unreliable-api.example.com/data')
560
print("Success:", response.json())
561
except niquests.RequestException as e:
562
print(f"Failed after retries: {e}")
563
```
564
565
## Best Practices
566
567
### Configuration Management
568
569
1. **Use configuration objects** for complex timeout and retry scenarios
570
2. **Set reasonable defaults** that work for your use case
571
3. **Configure at the session level** for consistency across requests
572
4. **Monitor and tune** timeout and retry settings based on actual performance
573
574
### Error Handling with Advanced Features
575
576
1. **Use status code constants** instead of magic numbers
577
2. **Implement progressive backoff** for retries
578
3. **Differentiate between retryable and non-retryable errors**
579
4. **Log retry attempts** for debugging and monitoring
580
581
### Performance Optimization
582
583
1. **Use appropriate timeout values** to prevent hanging requests
584
2. **Configure connection pooling** through sessions
585
3. **Implement request/response caching** where appropriate
586
4. **Monitor request patterns** and adjust configuration accordingly