0
# Configuration & Utilities
1

2
Global configuration management, region/zone settings, and utility functions for encoding, data transformation, and SDK customization.
3

4
## Capabilities
5

6
### Global Configuration
7

8
SDK-wide configuration management for default settings and service endpoints.
9

10
```python { .api }
11
def set_default(**kwargs):
12
    """
13
    Set global configuration defaults.
14
    
15
    Args:
16
        default_zone: Default upload zone
17
        connection_retries: Number of connection retries (default: 3)
18
        connection_pool: Connection pool size
19
        connection_timeout: Connection timeout in seconds (default: 30)
20
        default_rs_host: Default resource management host
21
        default_uc_host: Default space info host
22
        default_rsf_host: Default list operations host
23
        default_api_host: Default data processing host
24
        default_upload_threshold: File size threshold for resumable upload (default: 4MB)
25
        default_chunk_size: Chunk size for resumable uploads (default: 4MB)
26
        default_upload_recorder_root_directory: Root directory for upload progress records
27
        connection_pool_size: HTTP connection pool size
28
        connection_pool_max_size: Maximum connection pool size
29
        preferred_scheme: Preferred URL scheme ('http' or 'https')
30
    """
31

32
def get_default(key: str):
33
    """
34
    Get configuration value.
35
    
36
    Args:
37
        key: Configuration key name
38
        
39
    Returns:
40
        Configuration value or None if not set
41
    """
42

43
def is_customized_default(key: str) -> bool:
44
    """
45
    Check if configuration value has been customized.
46
    
47
    Args:
48
        key: Configuration key name
49
        
50
    Returns:
51
        True if value was customized from default
52
    """
53
```
54

55
### Configuration Constants
56

57
```python { .api }
58
# Service Host Constants
59
RS_HOST = 'http://rs.qiniu.com'              # Resource management operations
60
RSF_HOST = 'http://rsf.qbox.me'              # List operations
61
API_HOST = 'http://api.qiniuapi.com'         # Data processing operations
62
QUERY_REGION_HOST = 'https://uc.qiniuapi.com'  # Region query
63
UC_HOST = QUERY_REGION_HOST                   # Space info operations
64

65
# Upload Configuration
66
_BLOCK_SIZE = 4 * 1024 * 1024                # Resume upload block size (4MB)
67
```
68

69
### Region and Zone Management
70

71
Upload region configuration and endpoint discovery for optimal performance.
72

73
```python { .api }
74
class Region:
75
    def __init__(self, up_host: str = None, up_host_backup: str = None, io_host: str = None, host_cache = None, home_dir: str = None, scheme: str = "http", rs_host: str = None, rsf_host: str = None, api_host: str = None, accelerate_uploading: bool = False):
76
        """
77
        Initialize region configuration.
78
        
79
        Args:
80
            up_host: Primary upload host
81
            up_host_backup: Backup upload host
82
            io_host: IO operations host
83
            host_cache: Host cache instance
84
            home_dir: Home directory for cache
85
            scheme: URL scheme ('http' or 'https')
86
            rs_host: Resource management host
87
            rsf_host: List operations host
88
            api_host: API operations host
89
            accelerate_uploading: Enable upload acceleration
90
        """
91

92
    def get_up_host_by_token(self, up_token: str, home_dir: str) -> str:
93
        """
94
        Get upload host from token.
95
        
96
        Args:
97
            up_token: Upload token
98
            home_dir: Home directory for cache
99
            
100
        Returns:
101
            Primary upload host URL
102
        """
103

104
    def get_up_host_backup_by_token(self, up_token: str, home_dir: str) -> str:
105
        """
106
        Get backup upload host from token.
107
        
108
        Args:
109
            up_token: Upload token
110
            home_dir: Home directory for cache
111
            
112
        Returns:
113
            Backup upload host URL
114
        """
115

116
    def get_io_host(self, ak: str, bucket: str, home_dir: str = None) -> str:
117
        """
118
        Get IO operations host for bucket.
119
        
120
        Args:
121
            ak: Access key
122
            bucket: Bucket name
123
            home_dir: Home directory for cache
124
            
125
        Returns:
126
            IO host URL
127
        """
128

129
    def get_rs_host(self, ak: str, bucket: str, home_dir: str = None) -> str:
130
        """
131
        Get resource management host for bucket.
132
        
133
        Args:
134
            ak: Access key
135
            bucket: Bucket name
136
            home_dir: Home directory for cache
137
            
138
        Returns:
139
            Resource management host URL
140
        """
141

142
    def get_rsf_host(self, ak: str, bucket: str, home_dir: str = None) -> str:
143
        """
144
        Get listing operations host for bucket.
145
        
146
        Args:
147
            ak: Access key
148
            bucket: Bucket name
149
            home_dir: Home directory for cache
150
            
151
        Returns:
152
            Listing host URL
153
        """
154

155
    def get_api_host(self, ak: str, bucket: str, home_dir: str = None) -> str:
156
        """
157
        Get API operations host for bucket.
158
        
159
        Args:
160
            ak: Access key
161
            bucket: Bucket name
162
            home_dir: Home directory for cache
163
            
164
        Returns:
165
            API host URL
166
        """
167

168
    def get_up_host(self, ak: str, bucket: str, home_dir: str) -> tuple:
169
        """
170
        Get upload hosts for bucket.
171
        
172
        Args:
173
            ak: Access key
174
            bucket: Bucket name
175
            home_dir: Home directory for cache
176
            
177
        Returns:
178
            (primary_host, backup_host): Upload host URLs
179
        """
180

181
    def unmarshal_up_token(self, up_token: str) -> dict:
182
        """
183
        Parse upload token to extract policy information.
184
        
185
        Args:
186
            up_token: Upload token to parse
187
            
188
        Returns:
189
            Dictionary containing token policy data
190
        """
191

192
    def get_bucket_hosts(self, ak: str, bucket: str, home_dir: str = None, force: bool = False) -> dict:
193
        """
194
        Get all service hosts for bucket.
195
        
196
        Args:
197
            ak: Access key
198
            bucket: Bucket name
199
            home_dir: Home directory for cache
200
            force: Force refresh from server
201
            
202
        Returns:
203
            Dictionary containing all service host URLs
204
        """
205

206
    def bucket_hosts(self, ak: str, bucket: str) -> dict:
207
        """
208
        Get bucket service hosts information.
209
        
210
        Args:
211
            ak: Access key
212
            bucket: Bucket name
213
            
214
        Returns:
215
            Dictionary containing bucket hosts info
216
        """
217

218
class Zone:
219
    """Alias for Region class for backward compatibility."""
220
    # Inherits all methods from Region
221
```
222

223
### Encoding and Data Transformation
224

225
Utility functions for data encoding, decoding, and format transformation.
226

227
```python { .api }
228
def urlsafe_base64_encode(data: bytes) -> str:
229
    """
230
    URL-safe base64 encoding.
231
    
232
    Args:
233
        data: Binary data to encode
234
        
235
    Returns:
236
        URL-safe base64 encoded string
237
    """
238

239
def urlsafe_base64_decode(data: str) -> bytes:
240
    """
241
    URL-safe base64 decoding.
242
    
243
    Args:
244
        data: Base64 string to decode
245
        
246
    Returns:
247
        Decoded binary data
248
    """
249

250
def entry(bucket: str, key: str) -> str:
251
    """
252
    Create entry format for API operations.
253
    
254
    Args:
255
        bucket: Bucket name
256
        key: File key
257
        
258
    Returns:
259
        Base64 encoded entry string
260
    """
261

262
def decode_entry(e: str) -> tuple:
263
    """
264
    Decode entry format to extract bucket and key.
265
    
266
    Args:
267
        e: Encoded entry string
268
        
269
    Returns:
270
        (bucket, key): Decoded bucket name and file key
271
    """
272

273
def canonical_mime_header_key(field_name: str) -> str:
274
    """
275
    Canonicalize HTTP header key.
276
    
277
    Args:
278
        field_name: Header field name
279
        
280
    Returns:
281
        Canonicalized header key
282
    """
283
```
284

285
### Hash and Checksum Functions
286

287
Data integrity verification and hash calculation utilities.
288

289
```python { .api }
290
def file_crc32(file_path: str) -> int:
291
    """
292
    Calculate CRC32 checksum of file.
293
    
294
    Args:
295
        file_path: Path to file
296
        
297
    Returns:
298
        CRC32 checksum value
299
    """
300

301
def io_crc32(io_data) -> int:
302
    """
303
    Calculate CRC32 checksum of IO data.
304
    
305
    Args:
306
        io_data: IO object or binary data
307
        
308
    Returns:
309
        CRC32 checksum value
310
    """
311

312
def io_md5(io_data) -> str:
313
    """
314
    Calculate MD5 hash of IO data.
315
    
316
    Args:
317
        io_data: IO object or binary data
318
        
319
    Returns:
320
        MD5 hash string
321
    """
322

323
def crc32(data: bytes) -> int:
324
    """
325
    Calculate CRC32 checksum of binary data.
326
    
327
    Args:
328
        data: Binary data
329
        
330
    Returns:
331
        CRC32 checksum value
332
    """
333

334
def etag(file_path: str) -> str:
335
    """
336
    Calculate file etag (deprecated for v2 uploads).
337
    
338
    Args:
339
        file_path: Path to file
340
        
341
    Returns:
342
        Etag string
343
    """
344

345
def etag_stream(input_stream) -> str:
346
    """
347
    Calculate stream etag (deprecated for v2 uploads).
348
    
349
    Args:
350
        input_stream: Input stream object
351
        
352
    Returns:
353
        Etag string
354
    """
355
```
356

357
### Date and Time Utilities
358

359
Time format conversion and timestamp utilities.
360

361
```python { .api }
362
def rfc_from_timestamp(timestamp: int) -> str:
363
    """
364
    Convert Unix timestamp to HTTP RFC format.
365
    
366
    Args:
367
        timestamp: Unix timestamp
368
        
369
    Returns:
370
        RFC formatted date string
371
    """
372

373
def dt2ts(dt) -> int:
374
    """
375
    Convert datetime object to Unix timestamp.
376
    
377
    Args:
378
        dt: datetime object
379
        
380
    Returns:
381
        Unix timestamp
382
    """
383
```
384

385
### Development Utilities
386

387
Helper decorators and development tools.
388

389
```python { .api }
390
def deprecated(reason: str):
391
    """
392
    Decorator to mark functions as deprecated.
393
    
394
    Args:
395
        reason: Deprecation reason message
396
        
397
    Returns:
398
        Decorator function
399
    """
400
```
401

402
### HTTP Response Handling
403

404
Response information wrapper for standardized error handling.
405

406
```python { .api }
407
class ResponseInfo:
408
    def __init__(self, response, exception: Exception = None):
409
        """
410
        Initialize response info wrapper.
411
        
412
        Args:
413
            response: HTTP response object
414
            exception: Exception object if error occurred
415
        """
416

417
    @property
418
    def status_code(self) -> int:
419
        """HTTP status code."""
420

421
    @property
422
    def text_body(self) -> str:
423
        """Response body as text."""
424

425
    @property
426
    def req_id(self) -> str:
427
        """Qiniu request ID."""
428

429
    @property
430
    def x_log(self) -> str:
431
        """Qiniu debug log information."""
432

433
    @property
434
    def error(self) -> str:
435
        """Error message."""
436

437
    @property
438
    def url(self) -> str:
439
        """Request URL."""
440

441
    @property
442
    def exception(self) -> Exception:
443
        """Exception object."""
444

445
    def ok(self) -> bool:
446
        """
447
        Check if response indicates success (2xx status).
448
        
449
        Returns:
450
            True if response is successful
451
        """
452

453
    def need_retry(self) -> bool:
454
        """
455
        Check if request should be retried.
456
        
457
        Returns:
458
            True if request should be retried
459
        """
460

461
    def connect_failed(self) -> bool:
462
        """
463
        Check if connection failed.
464
        
465
        Returns:
466
            True if connection failed
467
        """
468

469
    def json(self) -> dict:
470
        """
471
        Parse response body as JSON.
472
        
473
        Returns:
474
            Parsed JSON dictionary
475
        """
476
```
477

478
## Usage Examples
479

480
### Global Configuration Setup
481

482
```python
483
from qiniu import set_default, get_default
484

485
# Configure SDK defaults
486
set_default(
487
    connection_retries=5,
488
    connection_timeout=60,
489
    default_upload_threshold=8 * 1024 * 1024,  # 8MB threshold for resumable upload
490
    preferred_scheme='https',
491
    connection_pool_size=20
492
)
493

494
# Check current configuration
495
print(f"Connection retries: {get_default('connection_retries')}")
496
print(f"Upload threshold: {get_default('default_upload_threshold')}")
497
print(f"Preferred scheme: {get_default('preferred_scheme')}")
498

499
# Custom host configuration for enterprise deployment
500
set_default(
501
    default_rs_host='https://rs.qiniu-enterprise.com',
502
    default_uc_host='https://uc.qiniu-enterprise.com',
503
    default_api_host='https://api.qiniu-enterprise.com'
504
)
505
```
506

507
### Region Configuration
508

509
```python
510
from qiniu import Region, Auth, BucketManager
511

512
# Create custom region configuration
513
custom_region = Region(
514
    up_host='https://upload-z0.qiniup.com',
515
    up_host_backup='https://upload-z0.qiniup.com',
516
    io_host='https://iovip.qbox.me',
517
    scheme='https',
518
    accelerate_uploading=True
519
)
520

521
# Use custom region with bucket manager
522
auth = Auth(access_key, secret_key)
523
bucket_manager = BucketManager(auth, regions=[custom_region])
524

525
# Get region-specific hosts for a bucket
526
hosts = custom_region.get_bucket_hosts(access_key, 'my-bucket', force=True)
527
print(f"Upload hosts: {hosts['up']}")
528
print(f"IO host: {hosts['io']}")
529
print(f"RS host: {hosts['rs']}")
530

531
# Parse upload token to determine region
532
upload_token = auth.upload_token('my-bucket')
533
token_info = custom_region.unmarshal_up_token(upload_token)
534
print(f"Token scope: {token_info['scope']}")
535
print(f"Token deadline: {token_info['deadline']}")
536
```
537

538
### Data Encoding and Transformation
539

540
```python
541
from qiniu import urlsafe_base64_encode, urlsafe_base64_decode, entry, decode_entry
542

543
# Base64 encoding for API parameters
544
data = b"Hello, Qiniu Cloud Storage!"
545
encoded = urlsafe_base64_encode(data)
546
print(f"Encoded: {encoded}")
547

548
decoded = urlsafe_base64_decode(encoded)
549
print(f"Decoded: {decoded.decode('utf-8')}")
550

551
# Entry format for batch operations
552
bucket = 'my-bucket'
553
key = 'path/to/file.jpg'
554
entry_str = entry(bucket, key)
555
print(f"Entry: {entry_str}")
556

557
# Decode entry back to bucket and key
558
decoded_bucket, decoded_key = decode_entry(entry_str)
559
print(f"Bucket: {decoded_bucket}, Key: {decoded_key}")
560

561
# Header canonicalization
562
from qiniu import canonical_mime_header_key
563
header_key = canonical_mime_header_key('content-type')
564
print(f"Canonical header: {header_key}")  # Content-Type
565
```
566

567
### Data Integrity Verification
568

569
```python
570
from qiniu import file_crc32, io_crc32, io_md5, crc32
571
import io
572

573
# File checksum verification
574
file_path = '/path/to/file.jpg'
575
file_checksum = file_crc32(file_path)
576
print(f"File CRC32: {file_checksum}")
577

578
# Data checksum calculation
579
data = b"Sample data for checksum"
580
data_checksum = crc32(data)
581
print(f"Data CRC32: {data_checksum}")
582

583
# IO stream checksum
584
stream = io.BytesIO(data)
585
stream_crc32 = io_crc32(stream)
586
stream_md5 = io_md5(stream)
587
print(f"Stream CRC32: {stream_crc32}")
588
print(f"Stream MD5: {stream_md5}")
589

590
# Verify upload integrity
591
from qiniu import Auth, put_data
592

593
auth = Auth(access_key, secret_key)
594
token = auth.upload_token('my-bucket', 'data.bin')
595

596
# Calculate checksum before upload
597
expected_crc32 = crc32(data)
598

599
# Upload with CRC verification
600
ret, info = put_data(token, 'data.bin', data, check_crc=True)
601

602
if info.ok():
603
    # Verify uploaded file CRC matches
604
    uploaded_crc32 = ret.get('crc32')
605
    if uploaded_crc32 == expected_crc32:
606
        print("Upload integrity verified")
607
    else:
608
        print("Upload integrity check failed")
609
```
610

611
### Time and Date Handling
612

613
```python
614
from qiniu import rfc_from_timestamp, dt2ts
615
import datetime
616
import time
617

618
# Convert timestamp to HTTP date format
619
timestamp = int(time.time())
620
http_date = rfc_from_timestamp(timestamp)
621
print(f"HTTP Date: {http_date}")
622

623
# Convert datetime to timestamp
624
dt = datetime.datetime.now()
625
ts = dt2ts(dt)
626
print(f"Datetime: {dt}")
627
print(f"Timestamp: {ts}")
628

629
# Use in upload token expiration
630
from qiniu import Auth
631

632
auth = Auth(access_key, secret_key)
633

634
# Token expires in 2 hours
635
expire_time = datetime.datetime.now() + datetime.timedelta(hours=2)
636
expire_timestamp = dt2ts(expire_time)
637

638
# Create token with specific expiration
639
token = auth.upload_token('my-bucket', expires=expire_timestamp)
640
print(f"Token expires at: {expire_time}")
641
```
642

643
### Response Handling and Error Management
644

645
```python
646
from qiniu import Auth, BucketManager, ResponseInfo
647

648
auth = Auth(access_key, secret_key)
649
bucket_manager = BucketManager(auth)
650

651
def handle_response(operation_name, ret, info):
652
    """Standard response handling"""
653
    print(f"\n=== {operation_name} ===")
654
    
655
    if info.ok():
656
        print("✓ Operation successful")
657
        print(f"Status: {info.status_code}")
658
        print(f"Request ID: {info.req_id}")
659
        
660
        if ret:
661
            print(f"Result: {ret}")
662
        
663
        return True
664
    else:
665
        print("✗ Operation failed")
666
        print(f"Status: {info.status_code}")
667
        print(f"Error: {info.error}")
668
        
669
        if info.exception:
670
            print(f"Exception: {info.exception}")
671
        
672
        if info.need_retry():
673
            print("⏳ Operation can be retried")
674
        
675
        if info.connect_failed():
676
            print("🔌 Connection failed")
677
        
678
        return False
679

680
# Example operations with error handling
681
ret, info = bucket_manager.stat('my-bucket', 'non-existent-file.jpg')
682
success = handle_response("File Stat", ret, info)
683

684
if not success and info.need_retry():
685
    print("Retrying operation...")
686
    ret, info = bucket_manager.stat('my-bucket', 'non-existent-file.jpg')
687
    handle_response("File Stat Retry", ret, info)
688

689
# JSON response parsing
690
try:
691
    json_data = info.json()
692
    print(f"JSON Response: {json_data}")
693
except ValueError as e:
694
    print(f"Failed to parse JSON: {e}")
695
```
696

697
### Custom Deprecation Warnings
698

699
```python
700
from qiniu import deprecated
701
import warnings
702

703
@deprecated("Use new_function() instead")
704
def old_function():
705
    """This function is deprecated"""
706
    return "old result"
707

708
def new_function():
709
    """New improved function"""
710
    return "new result"
711

712
# Using deprecated function shows warning
713
with warnings.catch_warnings(record=True) as w:
714
    warnings.simplefilter("always")
715
    result = old_function()
716
    
717
    if w:
718
        print(f"Warning: {w[0].message}")
719

720
# Use new function instead
721
result = new_function()
722
print(f"Result: {result}")
723
```
724

725
### Advanced Configuration Management
726

727
```python
728
from qiniu import set_default, get_default, is_customized_default
729
import os
730

731
class QiniuConfig:
732
    """Advanced configuration management"""
733
    
734
    def __init__(self):
735
        self.load_from_environment()
736
    
737
    def load_from_environment(self):
738
        """Load configuration from environment variables"""
739
        env_config = {}
740
        
741
        # Map environment variables to config keys
742
        env_mapping = {
743
            'QINIU_CONNECTION_RETRIES': 'connection_retries',
744
            'QINIU_CONNECTION_TIMEOUT': 'connection_timeout',
745
            'QINIU_PREFERRED_SCHEME': 'preferred_scheme',
746
            'QINIU_UPLOAD_THRESHOLD': 'default_upload_threshold',
747
            'QINIU_RS_HOST': 'default_rs_host',
748
            'QINIU_UC_HOST': 'default_uc_host'
749
        }
750
        
751
        for env_var, config_key in env_mapping.items():
752
            value = os.getenv(env_var)
753
            if value:
754
                # Convert numeric values
755
                if config_key in ['connection_retries', 'connection_timeout', 'default_upload_threshold']:
756
                    value = int(value)
757
                env_config[config_key] = value
758
        
759
        if env_config:
760
            set_default(**env_config)
761
            print(f"Loaded {len(env_config)} settings from environment")
762
    
763
    def show_current_config(self):
764
        """Display current configuration"""
765
        config_keys = [
766
            'connection_retries',
767
            'connection_timeout', 
768
            'preferred_scheme',
769
            'default_upload_threshold',
770
            'default_rs_host',
771
            'default_uc_host'
772
        ]
773
        
774
        print("\n=== Qiniu SDK Configuration ===")
775
        for key in config_keys:
776
            value = get_default(key)
777
            customized = is_customized_default(key)
778
            status = "CUSTOM" if customized else "DEFAULT"
779
            print(f"{key}: {value} ({status})")
780
    
781
    def reset_to_defaults(self):
782
        """Reset configuration to defaults"""
783
        # This would require SDK support for resetting
784
        print("Configuration reset not directly supported")
785

786
# Usage
787
config = QiniuConfig()
788
config.show_current_config()
789
```