0
# Key-Value Operations
1
2
The `KeyValuesOperations` class provides direct management of configuration key-value pairs through the Azure Resource Manager API. This enables programmatic CRUD operations on individual configuration keys within App Configuration stores.
3
4
## Operations Class
5
6
```python { .api }
7
class KeyValuesOperations:
8
"""
9
Operations for managing key-value pairs in Azure App Configuration stores.
10
11
This class provides methods for creating, reading, updating, and deleting individual
12
key-value pairs through the ARM API. Note that this is different from the data plane
13
API used by the azure-appconfiguration package.
14
"""
15
```
16
17
## Key-Value Management
18
19
### Get Key-Value
20
21
```python { .api }
22
def get(
23
self,
24
resource_group_name: str,
25
config_store_name: str,
26
key_value_name: str,
27
**kwargs: Any
28
) -> KeyValue:
29
"""
30
Gets the properties of a key-value pair in an App Configuration store.
31
32
Args:
33
resource_group_name: The name of the resource group containing the store.
34
config_store_name: The name of the configuration store.
35
key_value_name: The name of the key-value pair. Use '$' prefix for reserved keys.
36
**kwargs: Additional keyword arguments for the request.
37
38
Returns:
39
KeyValue: The key-value pair with its properties.
40
41
Raises:
42
HttpResponseError: If the key-value is not found or access is denied.
43
44
Example:
45
>>> kv = client.key_values.get("my-rg", "my-store", "MyApp:Settings:DatabaseUrl")
46
>>> print(f"Value: {kv.value}")
47
>>> print(f"Content Type: {kv.content_type}")
48
>>> print(f"ETag: {kv.etag}")
49
>>> print(f"Last Modified: {kv.last_modified}")
50
"""
51
```
52
53
### Create or Update Key-Value
54
55
```python { .api }
56
def create_or_update(
57
self,
58
resource_group_name: str,
59
config_store_name: str,
60
key_value_name: str,
61
key_value_parameters: Optional[KeyValue] = None,
62
**kwargs: Any
63
) -> KeyValue:
64
"""
65
Creates or updates a key-value pair in an App Configuration store.
66
67
Args:
68
resource_group_name: The name of the resource group containing the store.
69
config_store_name: The name of the configuration store.
70
key_value_name: The name of the key-value pair.
71
key_value_parameters: The key-value parameters for creation or update.
72
**kwargs: Additional keyword arguments for the request.
73
74
Returns:
75
KeyValue: The created or updated key-value pair.
76
77
Note:
78
If key_value_parameters is None, creates a key with null value.
79
Use ETags in key_value_parameters for conditional updates.
80
81
Example:
82
>>> from azure.mgmt.appconfiguration.models import KeyValue
83
>>> kv_params = KeyValue(
84
... value="production-database-url",
85
... content_type="text/plain",
86
... tags={"Environment": "Production"}
87
... )
88
>>> result = client.key_values.create_or_update(
89
... "my-rg", "my-store", "MyApp:Settings:DatabaseUrl", kv_params
90
... )
91
>>> print(f"Created key-value: {result.key}")
92
"""
93
```
94
95
### Delete Key-Value
96
97
```python { .api }
98
def begin_delete(
99
self,
100
resource_group_name: str,
101
config_store_name: str,
102
key_value_name: str,
103
**kwargs: Any
104
) -> LROPoller[None]:
105
"""
106
Deletes a key-value pair from an App Configuration store.
107
108
Args:
109
resource_group_name: The name of the resource group containing the store.
110
config_store_name: The name of the configuration store.
111
key_value_name: The name of the key-value pair to delete.
112
**kwargs: Additional keyword arguments for the request.
113
114
Returns:
115
LROPoller[None]: A poller for the long-running delete operation.
116
117
Example:
118
>>> delete_poller = client.key_values.begin_delete(
119
... "my-rg", "my-store", "MyApp:Settings:ObsoleteKey"
120
... )
121
>>> delete_poller.wait() # Wait for deletion to complete
122
>>> print("Key-value deleted successfully")
123
"""
124
```
125
126
## Key-Value Models and Properties
127
128
### KeyValue Model
129
130
```python { .api }
131
class KeyValue:
132
"""
133
Represents a key-value pair in Azure App Configuration.
134
135
Attributes:
136
key (str): The key name (read-only after creation).
137
label (str): The label for the key-value pair.
138
value (str): The value of the key-value pair.
139
content_type (str): The content type of the value.
140
etag (str): The ETag for concurrency control (read-only).
141
last_modified (datetime): The last modification time (read-only).
142
locked (bool): Whether the key-value is locked (read-only).
143
tags (Dict[str, str]): Tags associated with the key-value pair.
144
"""
145
146
def __init__(
147
self,
148
*,
149
value: Optional[str] = None,
150
content_type: Optional[str] = None,
151
tags: Optional[Dict[str, str]] = None,
152
**kwargs: Any
153
) -> None:
154
"""
155
Initialize a KeyValue instance.
156
157
Args:
158
value: The value of the key-value pair.
159
content_type: The content type of the value (e.g., "application/json").
160
tags: Tags to associate with the key-value pair.
161
**kwargs: Additional properties.
162
"""
163
```
164
165
## Practical Usage Examples
166
167
### Basic Key-Value Operations
168
169
```python { .api }
170
from azure.mgmt.appconfiguration import AppConfigurationManagementClient
171
from azure.mgmt.appconfiguration.models import KeyValue
172
from azure.identity import DefaultAzureCredential
173
174
# Initialize client
175
credential = DefaultAzureCredential()
176
client = AppConfigurationManagementClient(credential, "subscription-id")
177
178
resource_group = "my-resource-group"
179
store_name = "my-config-store"
180
181
# Create a simple key-value pair
182
print("Creating basic key-value...")
183
basic_kv = KeyValue(value="Hello, World!")
184
result = client.key_values.create_or_update(
185
resource_group,
186
store_name,
187
"MyApp:Greeting",
188
basic_kv
189
)
190
print(f"Created: {result.key} = {result.value}")
191
192
# Create a structured configuration value
193
print("Creating JSON configuration...")
194
json_config = KeyValue(
195
value='{"timeout": 30, "retries": 3, "endpoint": "https://api.example.com"}',
196
content_type="application/json",
197
tags={"Type": "Configuration", "Environment": "Production"}
198
)
199
result = client.key_values.create_or_update(
200
resource_group,
201
store_name,
202
"MyApp:ApiSettings",
203
json_config
204
)
205
print(f"Created JSON config: {result.key}")
206
207
# Retrieve a key-value
208
print("Retrieving key-value...")
209
retrieved_kv = client.key_values.get(resource_group, store_name, "MyApp:Greeting")
210
print(f"Retrieved: {retrieved_kv.key} = {retrieved_kv.value}")
211
print(f"Content-Type: {retrieved_kv.content_type}")
212
print(f"Last Modified: {retrieved_kv.last_modified}")
213
```
214
215
### Working with Labels
216
217
```python { .api }
218
# Key-value pairs can have labels for environment-specific configurations
219
# Note: Labels are part of the key name in ARM API, formatted as "key$label"
220
221
# Production configuration
222
prod_kv = KeyValue(
223
value="prod-database-connection-string",
224
content_type="text/plain",
225
tags={"Environment": "Production"}
226
)
227
client.key_values.create_or_update(
228
resource_group,
229
store_name,
230
"MyApp:Database:ConnectionString$Production", # Key$Label format
231
prod_kv
232
)
233
234
# Development configuration
235
dev_kv = KeyValue(
236
value="dev-database-connection-string",
237
content_type="text/plain",
238
tags={"Environment": "Development"}
239
)
240
client.key_values.create_or_update(
241
resource_group,
242
store_name,
243
"MyApp:Database:ConnectionString$Development", # Key$Label format
244
dev_kv
245
)
246
247
# Retrieve environment-specific values
248
prod_config = client.key_values.get(
249
resource_group, store_name, "MyApp:Database:ConnectionString$Production"
250
)
251
dev_config = client.key_values.get(
252
resource_group, store_name, "MyApp:Database:ConnectionString$Development"
253
)
254
255
print(f"Production DB: {prod_config.value}")
256
print(f"Development DB: {dev_config.value}")
257
```
258
259
### Feature Flag Management
260
261
```python { .api }
262
# Create feature flags using reserved key prefix
263
import json
264
265
# Define feature flag structure
266
feature_flag_value = {
267
"id": "BetaFeature",
268
"description": "Beta feature for advanced users",
269
"enabled": False,
270
"conditions": {
271
"client_filters": [
272
{
273
"name": "Microsoft.Percentage",
274
"parameters": {
275
"Value": 50
276
}
277
}
278
]
279
}
280
}
281
282
# Create feature flag (use .appconfig.featureflag/ prefix)
283
feature_flag_kv = KeyValue(
284
value=json.dumps(feature_flag_value),
285
content_type="application/vnd.microsoft.appconfig.ff+json;charset=utf-8",
286
tags={"Type": "FeatureFlag", "Owner": "FeatureTeam"}
287
)
288
289
result = client.key_values.create_or_update(
290
resource_group,
291
store_name,
292
".appconfig.featureflag/BetaFeature", # Feature flag key format
293
feature_flag_kv
294
)
295
print(f"Created feature flag: {result.key}")
296
297
# Retrieve and parse feature flag
298
feature_flag = client.key_values.get(
299
resource_group, store_name, ".appconfig.featureflag/BetaFeature"
300
)
301
flag_config = json.loads(feature_flag.value)
302
print(f"Feature '{flag_config['id']}' enabled: {flag_config['enabled']}")
303
```
304
305
### Conditional Updates with ETags
306
307
```python { .api }
308
# ETags provide optimistic concurrency control
309
key_name = "MyApp:Settings:MaxConnections"
310
311
# Get current value with ETag
312
current_kv = client.key_values.get(resource_group, store_name, key_name)
313
print(f"Current value: {current_kv.value}, ETag: {current_kv.etag}")
314
315
# Update with ETag for conditional update
316
updated_kv = KeyValue(
317
value="100", # New value
318
content_type="text/plain",
319
# Note: ETags are handled automatically by the service in ARM API
320
# For conditional updates, you would typically use If-Match headers in kwargs
321
)
322
323
try:
324
# Perform conditional update
325
result = client.key_values.create_or_update(
326
resource_group,
327
store_name,
328
key_name,
329
updated_kv,
330
# For conditional operations, use headers
331
headers={"If-Match": current_kv.etag}
332
)
333
print(f"Updated successfully: {result.value}")
334
except Exception as e:
335
print(f"Conditional update failed: {e}")
336
# Handle conflict - value was modified by another client
337
```
338
339
### Bulk Key Management
340
341
```python { .api }
342
# Managing multiple keys efficiently
343
import time
344
345
def bulk_create_configuration(config_dict: dict, key_prefix: str = ""):
346
"""Create multiple configuration keys from a dictionary."""
347
results = []
348
349
for key, value in config_dict.items():
350
full_key = f"{key_prefix}:{key}" if key_prefix else key
351
352
# Determine content type based on value type
353
if isinstance(value, (dict, list)):
354
content_type = "application/json"
355
str_value = json.dumps(value)
356
elif isinstance(value, bool):
357
content_type = "text/plain"
358
str_value = str(value).lower()
359
elif isinstance(value, (int, float)):
360
content_type = "text/plain"
361
str_value = str(value)
362
else:
363
content_type = "text/plain"
364
str_value = str(value)
365
366
kv = KeyValue(
367
value=str_value,
368
content_type=content_type,
369
tags={"CreatedBy": "BulkImport", "Timestamp": str(int(time.time()))}
370
)
371
372
try:
373
result = client.key_values.create_or_update(
374
resource_group, store_name, full_key, kv
375
)
376
results.append(result)
377
print(f"✓ Created: {result.key}")
378
except Exception as e:
379
print(f"✗ Failed to create {full_key}: {e}")
380
381
return results
382
383
# Example configuration
384
app_config = {
385
"Database": {
386
"ConnectionString": "Server=prod-db;Database=MyApp;",
387
"Timeout": 30,
388
"RetryCount": 3
389
},
390
"Api": {
391
"BaseUrl": "https://api.myapp.com",
392
"ApiKey": "secret-api-key",
393
"RateLimitPerMinute": 1000
394
},
395
"Features": {
396
"EnableCaching": True,
397
"EnableLogging": True,
398
"DebugMode": False
399
}
400
}
401
402
# Create all configurations
403
print("Creating bulk configuration...")
404
created_keys = bulk_create_configuration(app_config, "MyApp:Settings")
405
print(f"Created {len(created_keys)} configuration keys")
406
```
407
408
### Configuration Validation and Cleanup
409
410
```python { .api }
411
def validate_and_cleanup_config(prefix: str = ""):
412
"""Validate configuration keys and clean up invalid ones."""
413
print(f"Validating configuration with prefix: {prefix}")
414
415
# Note: ARM API doesn't provide list operation for key-values
416
# You would need to track keys separately or use the data plane API
417
# This example shows how to validate known keys
418
419
known_keys = [
420
"MyApp:Settings:Database:ConnectionString",
421
"MyApp:Settings:Api:BaseUrl",
422
"MyApp:Settings:Features:EnableCaching"
423
]
424
425
valid_keys = []
426
invalid_keys = []
427
428
for key_name in known_keys:
429
try:
430
kv = client.key_values.get(resource_group, store_name, key_name)
431
432
# Validate based on content type and value
433
if kv.content_type == "application/json":
434
try:
435
json.loads(kv.value) # Validate JSON
436
valid_keys.append(key_name)
437
except json.JSONDecodeError:
438
print(f"✗ Invalid JSON in key: {key_name}")
439
invalid_keys.append(key_name)
440
else:
441
# Basic validation for non-JSON values
442
if kv.value is not None and len(kv.value.strip()) > 0:
443
valid_keys.append(key_name)
444
else:
445
print(f"✗ Empty value in key: {key_name}")
446
invalid_keys.append(key_name)
447
448
except Exception as e:
449
print(f"✗ Error accessing key {key_name}: {e}")
450
invalid_keys.append(key_name)
451
452
print(f"Validation complete: {len(valid_keys)} valid, {len(invalid_keys)} invalid")
453
454
# Optionally clean up invalid keys
455
if invalid_keys:
456
response = input(f"Delete {len(invalid_keys)} invalid keys? (y/N): ")
457
if response.lower() == 'y':
458
for key_name in invalid_keys:
459
try:
460
delete_poller = client.key_values.begin_delete(
461
resource_group, store_name, key_name
462
)
463
delete_poller.wait()
464
print(f"✓ Deleted invalid key: {key_name}")
465
except Exception as e:
466
print(f"✗ Failed to delete {key_name}: {e}")
467
468
# Run validation
469
validate_and_cleanup_config()
470
```
471
472
## Error Handling
473
474
```python { .api }
475
from azure.core.exceptions import HttpResponseError, ResourceNotFoundError
476
477
def safe_key_operations():
478
"""Demonstrate error handling for key-value operations."""
479
480
try:
481
# Try to get a non-existent key
482
kv = client.key_values.get(resource_group, store_name, "NonExistent:Key")
483
except ResourceNotFoundError:
484
print("Key not found - this is expected for non-existent keys")
485
except HttpResponseError as e:
486
print(f"HTTP error occurred: {e.status_code} - {e.reason}")
487
if hasattr(e, 'error') and e.error:
488
print(f"Error details: {e.error}")
489
490
try:
491
# Try to create a key with invalid characters
492
invalid_kv = KeyValue(value="test-value")
493
result = client.key_values.create_or_update(
494
resource_group,
495
store_name,
496
"invalid/key/name", # Forward slashes may not be allowed
497
invalid_kv
498
)
499
except HttpResponseError as e:
500
print(f"Invalid key name error: {e.status_code}")
501
# Handle validation errors appropriately
502
503
try:
504
# Attempt operation on non-existent store
505
kv = KeyValue(value="test")
506
result = client.key_values.create_or_update(
507
resource_group,
508
"non-existent-store",
509
"TestKey",
510
kv
511
)
512
except HttpResponseError as e:
513
if e.status_code == 404:
514
print("Configuration store not found")
515
elif e.status_code == 403:
516
print("Access denied - check permissions")
517
else:
518
print(f"Unexpected error: {e.status_code}")
519
520
safe_key_operations()
521
```
522
523
## Asynchronous Operations
524
525
```python { .api }
526
from azure.mgmt.appconfiguration.aio import AppConfigurationManagementClient
527
from azure.identity.aio import DefaultAzureCredential
528
import asyncio
529
530
async def async_key_value_operations():
531
"""Demonstrate asynchronous key-value operations."""
532
credential = DefaultAzureCredential()
533
534
async with AppConfigurationManagementClient(credential, "subscription-id") as client:
535
# Create multiple keys concurrently
536
tasks = []
537
538
for i in range(5):
539
kv = KeyValue(
540
value=f"async-value-{i}",
541
content_type="text/plain",
542
tags={"Batch": "AsyncCreation"}
543
)
544
545
task = client.key_values.create_or_update(
546
resource_group,
547
store_name,
548
f"AsyncTest:Key{i}",
549
kv
550
)
551
tasks.append(task)
552
553
# Wait for all creates to complete
554
results = await asyncio.gather(*tasks, return_exceptions=True)
555
556
successful_creates = [r for r in results if not isinstance(r, Exception)]
557
print(f"Successfully created {len(successful_creates)} keys concurrently")
558
559
# Retrieve keys concurrently
560
get_tasks = [
561
client.key_values.get(resource_group, store_name, f"AsyncTest:Key{i}")
562
for i in range(5)
563
]
564
565
retrieved_kvs = await asyncio.gather(*get_tasks, return_exceptions=True)
566
567
for i, kv in enumerate(retrieved_kvs):
568
if not isinstance(kv, Exception):
569
print(f"Retrieved: {kv.key} = {kv.value}")
570
571
# Run async operations
572
asyncio.run(async_key_value_operations())
573
```
574
575
## Integration with Azure App Configuration Data Plane
576
577
```python { .api }
578
# Note: This ARM API is different from the data plane API
579
# For comprehensive key-value management, you might use both APIs
580
581
from azure.appconfiguration import AzureAppConfigurationClient
582
583
def compare_arm_vs_dataplane():
584
"""Compare ARM API vs Data Plane API for key-value operations."""
585
586
# ARM API client (management operations)
587
mgmt_client = AppConfigurationManagementClient(credential, subscription_id)
588
589
# Data Plane API client (configuration operations)
590
# Requires connection string from the store
591
store = mgmt_client.configuration_stores.get(resource_group, store_name)
592
keys = mgmt_client.configuration_stores.list_keys(resource_group, store_name)
593
connection_string = next(iter(keys)).connection_string
594
595
data_client = AzureAppConfigurationClient.from_connection_string(connection_string)
596
597
print("=== ARM API (Management) ===")
598
# ARM API: Individual key operations
599
arm_kv = KeyValue(value="arm-api-value", content_type="text/plain")
600
arm_result = mgmt_client.key_values.create_or_update(
601
resource_group, store_name, "TestKey", arm_kv
602
)
603
print(f"ARM API created: {arm_result.key}")
604
605
print("=== Data Plane API (Configuration) ===")
606
# Data Plane API: Rich querying and batch operations
607
from azure.appconfiguration import ConfigurationSetting
608
609
# Set configuration
610
setting = ConfigurationSetting(key="TestKey", value="data-plane-value")
611
data_client.set_configuration_setting(setting)
612
613
# Query with filters
614
settings = data_client.list_configuration_settings(key_filter="Test*")
615
for setting in settings:
616
print(f"Data Plane found: {setting.key} = {setting.value}")
617
618
print("\nUse ARM API for: Store management, access control, metadata")
619
print("Use Data Plane API for: Configuration CRUD, queries, real-time updates")
620
621
# compare_arm_vs_dataplane()
622
```