0
# Utilities and Helpers
1

2
Azure Core provides utility functions and helper classes for common operations used across Azure SDK clients. These utilities handle connection string parsing, case-insensitive operations, messaging, and other foundational functionality.
3

4
## Connection String Utilities
5

6
Functions for parsing and handling Azure connection strings commonly used across Azure services.
7

8
```python { .api }
9
from azure.core.utils import parse_connection_string
10
from typing import Dict
11

12
def parse_connection_string(conn_str: str) -> Dict[str, str]:
13
    """
14
    Parse an Azure connection string into a dictionary.
15
    
16
    Parameters:
17
    - conn_str (str): Connection string in key=value format
18
    
19
    Returns:
20
    Dict[str, str]: Dictionary of connection string components
21
    """
22
    ...
23
```
24

25
## Case-Insensitive Collections
26

27
Utilities for handling case-insensitive operations commonly needed in HTTP and configuration contexts.
28

29
```python { .api }
30
from azure.core.utils import case_insensitive_dict, CaseInsensitiveDict
31
from typing import Dict, Any, Optional
32

33
def case_insensitive_dict(data: Optional[Dict] = None) -> CaseInsensitiveDict:
34
    """Create a case-insensitive dictionary."""
35
    ...
36

37
class CaseInsensitiveDict(dict):
38
    """Dictionary that performs case-insensitive key lookups."""
39
    def __init__(self, data: Optional[Dict] = None): ...
40
    
41
    def __getitem__(self, key: str) -> Any: ...
42
    def __setitem__(self, key: str, value: Any) -> None: ...
43
    def __delitem__(self, key: str) -> None: ...
44
    def __contains__(self, key: str) -> bool: ...
45
    
46
    def get(self, key: str, default: Any = None) -> Any: ...
47
    def pop(self, key: str, default: Any = None) -> Any: ...
48
    def setdefault(self, key: str, default: Any = None) -> Any: ...
49
    def update(self, other: Dict) -> None: ...
50
```
51

52
## Match Conditions and ETags
53

54
Enumeration for conditional request operations commonly used in Azure services.
55

56
```python { .api }
57
from azure.core import MatchConditions
58
from enum import Enum
59

60
class MatchConditions(Enum):
61
    """Conditions for conditional HTTP requests."""
62
    IfMatch = 1          # If-Match header condition
63
    IfNoneMatch = 2      # If-None-Match header condition  
64
    IfModifiedSince = 3  # If-Modified-Since header condition
65
    IfNotModifiedSince = 4  # If-Not-Modified-Since header condition
66
```
67

68
## Case-Insensitive Enum Support
69

70
Metaclass for creating case-insensitive enumerations.
71

72
```python { .api }
73
from azure.core import CaseInsensitiveEnumMeta
74
from enum import EnumMeta
75

76
class CaseInsensitiveEnumMeta(EnumMeta):
77
    """Metaclass for case-insensitive enum lookups."""
78
    def __getitem__(cls, name: str): ...
79
    def __call__(cls, value): ...
80
```
81

82
## Serialization Utilities
83

84
Utilities for handling serialization, null values, and model operations.
85

86
```python { .api }
87
from azure.core.serialization import NULL, AzureJSONEncoder, is_generated_model, as_attribute_dict, attribute_list
88
from typing import Any, Dict, List
89
import json
90

91
# Null sentinel for JSON serialization
92
NULL: object  # Falsy sentinel value for null representation
93

94
class AzureJSONEncoder(json.JSONEncoder):
95
    """JSON encoder for Azure-specific types."""
96
    def default(self, obj: Any) -> Any: ...
97

98
def is_generated_model(obj: Any) -> bool:
99
    """Check if object is a generated model class."""
100
    ...
101

102
def as_attribute_dict(obj: Any) -> Dict[str, Any]:
103
    """Convert object to attribute dictionary."""
104
    ...
105

106
def attribute_list(obj: Any) -> List[str]:
107
    """Get list of object attributes."""
108
    ...
109
```
110

111
## Cloud Events and Messaging
112

113
Support for CloudEvents specification and messaging utilities.
114

115
```python { .api }
116
from azure.core.messaging import CloudEvent
117
from typing import Dict, Any, Optional
118

119
class CloudEvent:
120
    """CloudEvents 1.0 schema implementation."""
121
    def __init__(
122
        self,
123
        source: str,
124
        type: str,
125
        data: Optional[Any] = None,
126
        subject: Optional[str] = None,
127
        **kwargs
128
    ): ...
129
    
130
    @property
131
    def source(self) -> str: ...
132
    
133
    @property
134
    def type(self) -> str: ...
135
    
136
    @property
137
    def data(self) -> Any: ...
138
    
139
    @property
140
    def subject(self) -> Optional[str]: ...
141
    
142
    @property
143
    def id(self) -> str: ...
144
    
145
    @property
146
    def time(self) -> Optional[str]: ...
147
    
148
    @property
149
    def specversion(self) -> str: ...
150
    
151
    @property
152
    def datacontenttype(self) -> Optional[str]: ...
153
    
154
    @property
155
    def dataschema(self) -> Optional[str]: ...
156
    
157
    def to_dict(self) -> Dict[str, Any]:
158
        """Convert CloudEvent to dictionary."""
159
        ...
160
    
161
    @classmethod
162
    def from_dict(cls, data: Dict[str, Any]) -> "CloudEvent":
163
        """Create CloudEvent from dictionary."""
164
        ...
165
```
166

167
## Azure Cloud Configuration
168

169
Configuration constants for different Azure cloud environments.
170

171
```python { .api }
172
from azure.core import AzureClouds
173

174
class AzureClouds:
175
    """Azure cloud environment configurations."""
176
    
177
    # Public Azure Cloud
178
    AZURE_PUBLIC_CLOUD: str
179
    
180
    # Azure Government Cloud
181
    AZURE_GOVERNMENT_CLOUD: str
182
    
183
    # Azure China Cloud  
184
    AZURE_CHINA_CLOUD: str
185
    
186
    # Azure German Cloud (deprecated)
187
    AZURE_GERMAN_CLOUD: str
188
```
189

190
## Global Settings
191

192
Access to global Azure SDK configuration settings.
193

194
```python { .api }
195
from azure.core.settings import settings, Settings
196

197
class Settings:
198
    """Global Azure SDK settings."""
199
    
200
    @property
201
    def log_level(self) -> str: ...
202
    
203
    @log_level.setter  
204
    def log_level(self, value: str) -> None: ...
205
    
206
    @property
207
    def tracing_enabled(self) -> bool: ...
208
    
209
    @tracing_enabled.setter
210
    def tracing_enabled(self, value: bool) -> None: ...
211

212
# Global settings instance
213
settings: Settings
214
```
215

216
## Usage Examples
217

218
### Connection String Parsing
219

220
```python
221
from azure.core.utils import parse_connection_string
222

223
# Parse Azure Storage connection string
224
storage_conn_str = "DefaultEndpointsProtocol=https;AccountName=mystorageaccount;AccountKey=mykey;EndpointSuffix=core.windows.net"
225
parsed = parse_connection_string(storage_conn_str)
226

227
print("Account Name:", parsed.get("AccountName"))
228
print("Account Key:", parsed.get("AccountKey"))
229
print("Protocol:", parsed.get("DefaultEndpointsProtocol"))
230
print("Endpoint Suffix:", parsed.get("EndpointSuffix"))
231

232
# Parse Service Bus connection string
233
servicebus_conn_str = "Endpoint=sb://myservicebus.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=mykey"
234
parsed_sb = parse_connection_string(servicebus_conn_str)
235

236
print("Endpoint:", parsed_sb.get("Endpoint"))
237
print("Key Name:", parsed_sb.get("SharedAccessKeyName"))
238
print("Key:", parsed_sb.get("SharedAccessKey"))
239
```
240

241
### Case-Insensitive Dictionary Operations
242

243
```python
244
from azure.core.utils import CaseInsensitiveDict
245

246
# Create case-insensitive dictionary for HTTP headers
247
headers = CaseInsensitiveDict({
248
    "Content-Type": "application/json",
249
    "Authorization": "Bearer token123",
250
    "User-Agent": "MyApp/1.0"
251
})
252

253
# Access with different casing
254
print(headers["content-type"])     # Works: "application/json"
255
print(headers["AUTHORIZATION"])    # Works: "Bearer token123" 
256
print(headers["user-agent"])       # Works: "MyApp/1.0"
257

258
# Update with different casing
259
headers["CONTENT-TYPE"] = "application/xml"
260
print(headers["content-type"])     # Updated: "application/xml"
261

262
# Check existence with different casing
263
if "authorization" in headers:
264
    print("Authorization header present")
265

266
# Standard dictionary operations work
267
headers.update({"Accept": "application/json"})
268
print(headers.get("ACCEPT", "default"))  # "application/json"
269
```
270

271
### Match Conditions for Conditional Requests
272

273
```python
274
from azure.core import MatchConditions
275
from azure.core.pipeline.transport import HttpRequest
276

277
def create_conditional_request(url, etag, condition):
278
    """Create HTTP request with conditional headers."""
279
    request = HttpRequest("GET", url)
280
    
281
    if condition == MatchConditions.IfMatch:
282
        request.headers["If-Match"] = etag
283
    elif condition == MatchConditions.IfNoneMatch:
284
        request.headers["If-None-Match"] = etag
285
    elif condition == MatchConditions.IfModifiedSince:
286
        request.headers["If-Modified-Since"] = etag  # In this case, etag is a date
287
    elif condition == MatchConditions.IfNotModifiedSince:
288
        request.headers["If-Not-Modified-Since"] = etag
289
    
290
    return request
291

292
# Example usage
293
etag = '"abc123"'
294
request = create_conditional_request(
295
    "https://api.example.com/resource",
296
    etag,
297
    MatchConditions.IfNoneMatch
298
)
299
print("Headers:", request.headers)
300
```
301

302
### Case-Insensitive Enums
303

304
```python
305
from azure.core import CaseInsensitiveEnumMeta
306
from enum import Enum
307

308
class ApiVersion(Enum, metaclass=CaseInsensitiveEnumMeta):
309
    """API version enum with case-insensitive lookup."""
310
    V2020_06_12 = "2020-06-12"
311
    V2021_04_10 = "2021-04-10"
312
    V2022_11_02 = "2022-11-02"
313

314
# Case-insensitive access
315
version1 = ApiVersion["v2020_06_12"]      # Works
316
version2 = ApiVersion["V2020_06_12"]      # Works  
317
version3 = ApiVersion["2020-06-12"]       # Works
318

319
print(f"Version: {version1.value}")  # "2020-06-12"
320
```
321

322
### CloudEvent Handling
323

324
```python
325
from azure.core.messaging import CloudEvent
326
import json
327
from datetime import datetime
328

329
# Create CloudEvent
330
event = CloudEvent(
331
    source="https://myapp.example.com/orders",
332
    type="com.example.order.created",
333
    data={"orderId": "12345", "amount": 99.99},
334
    subject="orders/12345"
335
)
336

337
print(f"Event ID: {event.id}")
338
print(f"Event Source: {event.source}")
339
print(f"Event Type: {event.type}")
340
print(f"Event Data: {event.data}")
341

342
# Convert to dictionary for transmission
343
event_dict = event.to_dict()
344
print("Event as dict:", json.dumps(event_dict, indent=2))
345

346
# Recreate from dictionary
347
received_event = CloudEvent.from_dict(event_dict)
348
print(f"Recreated event data: {received_event.data}")
349

350
# Handle CloudEvent in webhook receiver
351
def handle_webhook_event(event_dict):
352
    """Handle received CloudEvent."""
353
    try:
354
        event = CloudEvent.from_dict(event_dict)
355
        
356
        if event.type == "com.example.order.created":
357
            process_order_created(event.data)
358
        elif event.type == "com.example.order.updated":
359
            process_order_updated(event.data)
360
        else:
361
            print(f"Unknown event type: {event.type}")
362
            
363
    except Exception as e:
364
        print(f"Error processing CloudEvent: {e}")
365

366
def process_order_created(order_data):
367
    print(f"Processing new order: {order_data['orderId']}")
368

369
def process_order_updated(order_data):
370
    print(f"Processing order update: {order_data['orderId']}")
371
```
372

373
### Serialization Utilities
374

375
```python
376
from azure.core.serialization import NULL, AzureJSONEncoder, is_generated_model
377
import json
378

379
# Using NULL sentinel for proper null handling
380
data = {
381
    "name": "example",
382
    "value": NULL,  # Will be serialized as null in JSON
383
    "active": True
384
}
385

386
# Serialize with Azure JSON encoder
387
json_string = json.dumps(data, cls=AzureJSONEncoder)
388
print("Serialized:", json_string)  # {"name": "example", "value": null, "active": true}
389

390
# Check if object is generated model
391
class CustomModel:
392
    def __init__(self):
393
        self.name = "test"
394

395
model = CustomModel()
396
if is_generated_model(model):
397
    print("This is a generated model")
398
else:
399
    print("This is a custom model")
400
```
401

402
### Settings Converter Utilities
403

404
Utility functions for converting environment variables and configuration values to appropriate types.
405

406
```python { .api }
407
from azure.core.settings import convert_bool, convert_logging, convert_azure_cloud
408
from azure.core import AzureClouds
409
from typing import Union
410
import logging
411

412
def convert_bool(value: Union[str, bool]) -> bool:
413
    """Convert string to boolean following Azure SDK conventions.
414
    
415
    Args:
416
        value: The value to convert (bool values returned as-is)
417
        
418
    Returns:
419
        Boolean value matching the intent of the input
420
    """
421

422
def convert_logging(value: Union[str, int]) -> int:
423
    """Convert string to Python logging level.
424
    
425
    Args:
426
        value: String log level name or integer level
427
        
428
    Returns:
429
        Integer log level for use with logging module
430
    """
431

432
def convert_azure_cloud(value: Union[str, AzureClouds]) -> AzureClouds:
433
    """Convert string to AzureClouds enum value.
434
    
435
    Args:
436
        value: String cloud name or AzureClouds enum value
437
        
438
    Returns:
439
        AzureClouds enum value
440
    """
441
```
442

443
#### Usage Examples
444

445
```python
446
from azure.core.settings import convert_bool, convert_logging, convert_azure_cloud
447
from azure.core import AzureClouds
448
import os
449

450
# Convert environment variables to proper types
451
debug_mode = convert_bool(os.environ.get("DEBUG", "false"))
452
print(f"Debug mode: {debug_mode}")  # False
453

454
log_level = convert_logging(os.environ.get("LOG_LEVEL", "INFO"))
455
print(f"Log level: {log_level}")  # 20 (logging.INFO)
456

457
cloud = convert_azure_cloud(os.environ.get("AZURE_CLOUD", "AZURE_PUBLIC_CLOUD"))
458
print(f"Azure cloud: {cloud}")  # AzureClouds.AZURE_PUBLIC_CLOUD
459

460
# String conversion examples
461
print(convert_bool("yes"))     # True
462
print(convert_bool("1"))       # True
463
print(convert_bool("on"))      # True
464
print(convert_bool("true"))    # True
465
print(convert_bool("false"))   # False
466

467
print(convert_logging("DEBUG"))    # 10
468
print(convert_logging("WARNING"))  # 30
469
print(convert_logging("ERROR"))    # 40
470
```
471

472
### Enhanced Connection String Parsing
473

474
Extended connection string parsing with case sensitivity options.
475

476
```python { .api }
477
from azure.core.utils import parse_connection_string
478
from typing import Mapping
479

480
def parse_connection_string(
481
    conn_str: str, 
482
    case_sensitive_keys: bool = False
483
) -> Mapping[str, str]:
484
    """Parse connection string with optional case preservation.
485
    
486
    Args:
487
        conn_str: String with connection details provided by Azure services
488
        case_sensitive_keys: Whether to preserve the original casing of keys
489
        
490
    Returns:
491
        Mapping of connection string key/value pairs
492
    """
493
```
494

495
#### Usage Examples
496

497
```python
498
from azure.core.utils import parse_connection_string
499

500
# Standard case-insensitive parsing (default)
501
conn_str = "Endpoint=sb://test.servicebus.windows.net/;SharedAccessKey=key123"
502
result = parse_connection_string(conn_str)
503
print(result["endpoint"])  # Works - lowercase key
504

505
# Case-sensitive parsing preserves original casing
506
result_sensitive = parse_connection_string(conn_str, case_sensitive_keys=True)
507
print(result_sensitive["Endpoint"])  # Original casing preserved
508
```
509

510
### Additional CaseInsensitiveDict Methods
511

512
Extended functionality for case-insensitive dictionary operations.
513

514
```python { .api }
515
from azure.core.utils import CaseInsensitiveDict
516
from typing import Iterator, Tuple, Any
517

518
class CaseInsensitiveDict:
519
    def copy(self) -> "CaseInsensitiveDict": 
520
        """Create a shallow copy of the case-insensitive dictionary."""
521
        
522
    def lowerkey_items(self) -> Iterator[Tuple[str, Any]]:
523
        """Iterate over (lowercase_key, value) pairs."""
524
```
525

526
#### Usage Examples
527

528
```python
529
from azure.core.utils import CaseInsensitiveDict
530

531
headers = CaseInsensitiveDict({
532
    "Content-Type": "application/json",
533
    "Authorization": "Bearer token123"
534
})
535

536
# Create a copy
537
headers_copy = headers.copy()
538
headers_copy["New-Header"] = "value"
539

540
# Iterate with consistent lowercase keys
541
for key, value in headers.lowerkey_items():
542
    print(f"{key}: {value}")  # Keys will be lowercase
543
    # Output:
544
    # content-type: application/json
545
    # authorization: Bearer token123
546
```
547

548
### Global Settings Configuration
549

550
```python
551
from azure.core.settings import settings
552

553
# Configure global logging level
554
settings.log_level = "DEBUG"
555
print(f"Current log level: {settings.log_level}")
556

557
# Enable/disable tracing
558
settings.tracing_enabled = True
559
print(f"Tracing enabled: {settings.tracing_enabled}")
560

561
# Settings are global and affect all Azure SDK clients
562
def configure_azure_sdk():
563
    """Configure global Azure SDK settings."""
564
    settings.log_level = "INFO"
565
    settings.tracing_enabled = False
566
    
567
    # Additional configuration can be added here
568
    print("Azure SDK configured with custom settings")
569

570
configure_azure_sdk()
571
```
572

573
### Cloud Environment Configuration
574

575
```python
576
from azure.core import AzureClouds
577

578
def get_cloud_endpoints(cloud_name):
579
    """Get endpoints for specific Azure cloud."""
580
    cloud_configs = {
581
        "public": {
582
            "base_url": "https://management.azure.com",
583
            "auth_url": "https://login.microsoftonline.com"
584
        },
585
        "government": {
586
            "base_url": "https://management.usgovcloudapi.net", 
587
            "auth_url": "https://login.microsoftonline.us"
588
        },
589
        "china": {
590
            "base_url": "https://management.chinacloudapi.cn",
591
            "auth_url": "https://login.chinacloudapi.cn"
592
        }
593
    }
594
    
595
    return cloud_configs.get(cloud_name, cloud_configs["public"])
596

597
# Usage
598
public_endpoints = get_cloud_endpoints("public")
599
gov_endpoints = get_cloud_endpoints("government")
600

601
print("Public cloud management URL:", public_endpoints["base_url"])
602
print("Government cloud auth URL:", gov_endpoints["auth_url"])
603
```
604

605
## Best Practices
606

607
### Connection String Handling
608

609
- Never log or expose connection strings in plain text
610
- Use secure storage (Key Vault, environment variables) for connection strings
611
- Validate parsed connection string components before use
612
- Handle parsing errors gracefully for malformed connection strings
613

614
### Case-Insensitive Operations
615

616
- Use `CaseInsensitiveDict` for HTTP headers and configuration data
617
- Be consistent with casing in your application code
618
- Consider performance implications for large dictionaries
619
- Document case-insensitive behavior in your APIs
620

621
### Serialization Best Practices
622

623
- Use the `NULL` sentinel for explicit null values in JSON
624
- Handle serialization errors appropriately
625
- Consider custom encoders for complex types
626
- Validate deserialized data for security and correctness
627

628
### Global Settings Management
629

630
- Configure global settings early in application startup
631
- Be aware that settings affect all Azure SDK clients
632
- Use appropriate log levels for different environments
633
- Consider performance impact of enabling tracing in production