0
# Snapshot Operations
1
2
The `SnapshotsOperations` class provides management capabilities for Azure App Configuration snapshots. Snapshots capture the state of configuration data at a specific point in time, enabling versioning, backup, and rollback scenarios for your configuration management.
3
4
## Operations Class
5
6
```python { .api }
7
class SnapshotsOperations:
8
"""
9
Operations for managing App Configuration snapshots.
10
11
Snapshots provide point-in-time captures of configuration data, allowing you to:
12
- Create versioned backups of configuration state
13
- Enable rollback to previous configurations
14
- Capture configuration for specific releases or deployments
15
- Audit configuration changes over time
16
"""
17
```
18
19
## Snapshot Management
20
21
### Get Snapshot
22
23
```python { .api }
24
def get(
25
self,
26
resource_group_name: str,
27
config_store_name: str,
28
snapshot_name: str,
29
**kwargs: Any
30
) -> Snapshot:
31
"""
32
Gets the properties and status of a configuration snapshot.
33
34
Args:
35
resource_group_name: The name of the resource group containing the store.
36
config_store_name: The name of the configuration store.
37
snapshot_name: The name of the snapshot.
38
**kwargs: Additional keyword arguments for the request.
39
40
Returns:
41
Snapshot: The snapshot details including status, composition type, and filters.
42
43
Example:
44
>>> snapshot = client.snapshots.get("my-rg", "my-store", "release-v1.0")
45
>>> print(f"Snapshot status: {snapshot.status}")
46
>>> print(f"Items count: {snapshot.items_count}")
47
>>> print(f"Size: {snapshot.size_in_bytes} bytes")
48
>>> print(f"Created: {snapshot.created}")
49
"""
50
```
51
52
### Create Snapshot
53
54
```python { .api }
55
def begin_create(
56
self,
57
resource_group_name: str,
58
config_store_name: str,
59
snapshot_name: str,
60
snapshot_creation_parameters: Snapshot,
61
**kwargs: Any
62
) -> LROPoller[Snapshot]:
63
"""
64
Creates a new configuration snapshot with the specified parameters.
65
66
Args:
67
resource_group_name: The name of the resource group containing the store.
68
config_store_name: The name of the configuration store.
69
snapshot_name: The name for the snapshot (must be unique within the store).
70
snapshot_creation_parameters: Parameters defining what to include in the snapshot.
71
**kwargs: Additional keyword arguments for the request.
72
73
Returns:
74
LROPoller[Snapshot]: A poller for the long-running create operation.
75
76
Note:
77
Snapshot creation is asynchronous. The snapshot status will be "Provisioning"
78
during creation and "Ready" when complete. Large configuration sets may take
79
several minutes to snapshot.
80
81
Example:
82
>>> from azure.mgmt.appconfiguration.models import (
83
... Snapshot, KeyValueFilter, CompositionType
84
... )
85
>>> # Create snapshot of all production configs
86
>>> filters = [KeyValueFilter(key="MyApp:*", label="Production")]
87
>>> snapshot_params = Snapshot(
88
... filters=filters,
89
... composition_type=CompositionType.KEY_LABEL,
90
... retention_period=7776000 # 90 days in seconds
91
... )
92
>>> create_poller = client.snapshots.begin_create(
93
... "my-rg", "my-store", "prod-release-v2.1", snapshot_params
94
... )
95
>>> snapshot = create_poller.result() # Wait for completion
96
"""
97
```
98
99
## Snapshot Model
100
101
### Snapshot Class
102
103
```python { .api }
104
class Snapshot:
105
"""
106
Represents a configuration snapshot in Azure App Configuration.
107
108
Attributes:
109
id (str): The resource ID of the snapshot (read-only).
110
name (str): The name of the snapshot (read-only).
111
type (str): The resource type (read-only).
112
filters (List[KeyValueFilter]): Filters that determine which key-values are included.
113
composition_type (CompositionType): How the snapshot composes key-value pairs.
114
retention_period (int): Retention period in seconds (1-7776000, max 90 days).
115
status (SnapshotStatus): Current status of the snapshot (read-only).
116
provisioning_state (str): ARM provisioning state (read-only).
117
created (datetime): When the snapshot was created (read-only).
118
expires (datetime): When the snapshot expires (read-only).
119
size_in_bytes (int): Size of the snapshot in bytes (read-only).
120
items_count (int): Number of key-value pairs in the snapshot (read-only).
121
tags (Dict[str, str]): Tags associated with the snapshot.
122
etag (str): The ETag for concurrency control (read-only).
123
"""
124
125
def __init__(
126
self,
127
*,
128
filters: List[KeyValueFilter],
129
composition_type: CompositionType,
130
retention_period: Optional[int] = None,
131
tags: Optional[Dict[str, str]] = None,
132
**kwargs: Any
133
) -> None:
134
"""
135
Initialize a Snapshot instance.
136
137
Args:
138
filters: List of filters to determine which key-values to include.
139
composition_type: How to compose key-value pairs (KEY or KEY_LABEL).
140
retention_period: Retention period in seconds (default: 2592000 = 30 days).
141
tags: Tags to associate with the snapshot.
142
"""
143
```
144
145
### KeyValueFilter Class
146
147
```python { .api }
148
class KeyValueFilter:
149
"""
150
Filter criteria for including key-value pairs in snapshots.
151
152
Attributes:
153
key (str): Key pattern to match (supports wildcards like "MyApp:*").
154
label (str): Label to match (optional, empty string matches unlabeled keys).
155
"""
156
157
def __init__(
158
self,
159
*,
160
key: str,
161
label: Optional[str] = None,
162
**kwargs: Any
163
) -> None:
164
"""
165
Initialize a KeyValueFilter.
166
167
Args:
168
key: Key pattern to match (use "*" for wildcards).
169
label: Label to match (None or empty string matches unlabeled keys).
170
"""
171
```
172
173
### CompositionType Enumeration
174
175
```python { .api }
176
class CompositionType:
177
"""
178
Enumeration of snapshot composition types.
179
180
Values:
181
KEY: Snapshot composes key-value pairs by key only (ignores labels).
182
KEY_LABEL: Snapshot composes key-value pairs by both key and label.
183
"""
184
KEY = "Key"
185
KEY_LABEL = "Key_Label"
186
```
187
188
### SnapshotStatus Enumeration
189
190
```python { .api }
191
class SnapshotStatus:
192
"""
193
Enumeration of snapshot status values.
194
195
Values:
196
PROVISIONING: Snapshot is being created.
197
READY: Snapshot has been created successfully and is available.
198
ARCHIVED: Snapshot has been archived (past retention period).
199
FAILED: Snapshot creation has failed.
200
"""
201
PROVISIONING = "Provisioning"
202
READY = "Ready"
203
ARCHIVED = "Archived"
204
FAILED = "Failed"
205
```
206
207
## Practical Usage Examples
208
209
### Basic Snapshot Creation
210
211
```python { .api }
212
from azure.mgmt.appconfiguration import AppConfigurationManagementClient
213
from azure.mgmt.appconfiguration.models import (
214
Snapshot,
215
KeyValueFilter,
216
CompositionType,
217
SnapshotStatus
218
)
219
from azure.identity import DefaultAzureCredential
220
import datetime
221
222
# Initialize client
223
credential = DefaultAzureCredential()
224
client = AppConfigurationManagementClient(credential, "subscription-id")
225
226
resource_group = "my-resource-group"
227
store_name = "my-config-store"
228
229
# Create a simple snapshot of all configuration
230
print("Creating snapshot of all configuration...")
231
232
# Define filters - capture all keys
233
all_config_filter = KeyValueFilter(key="*") # All keys
234
235
snapshot_params = Snapshot(
236
filters=[all_config_filter],
237
composition_type=CompositionType.KEY_LABEL, # Include labels
238
retention_period=2592000, # 30 days
239
tags={
240
"Purpose": "Full Backup",
241
"Created": datetime.datetime.now().isoformat(),
242
"Version": "1.0"
243
}
244
)
245
246
# Create the snapshot
247
create_poller = client.snapshots.begin_create(
248
resource_group,
249
store_name,
250
"full-backup-snapshot",
251
snapshot_params
252
)
253
254
print("Snapshot creation started...")
255
snapshot = create_poller.result() # Wait for completion
256
257
print(f"Snapshot created successfully!")
258
print(f"Name: {snapshot.name}")
259
print(f"Status: {snapshot.status}")
260
print(f"Items: {snapshot.items_count}")
261
print(f"Size: {snapshot.size_in_bytes} bytes")
262
print(f"Expires: {snapshot.expires}")
263
```
264
265
### Application Release Snapshots
266
267
```python { .api }
268
def create_release_snapshots():
269
"""Create snapshots for different application releases."""
270
271
# Define release configurations
272
releases = {
273
"v1.0": {
274
"filters": [
275
KeyValueFilter(key="MyApp:*", label="Production"),
276
KeyValueFilter(key="Shared:*", label="Production")
277
],
278
"description": "Production configuration for v1.0 release"
279
},
280
"v1.1-beta": {
281
"filters": [
282
KeyValueFilter(key="MyApp:*", label="Beta"),
283
KeyValueFilter(key="Shared:*", label="Beta"),
284
KeyValueFilter(key="Features:*", label="Beta")
285
],
286
"description": "Beta configuration for v1.1 release"
287
},
288
"v2.0-preview": {
289
"filters": [
290
KeyValueFilter(key="MyApp:v2:*", label="Preview"),
291
KeyValueFilter(key="NewFeatures:*", label="Preview")
292
],
293
"description": "Preview configuration for v2.0 release"
294
}
295
}
296
297
print("=== Creating Release Snapshots ===")
298
299
created_snapshots = {}
300
301
for version, config in releases.items():
302
snapshot_name = f"release-{version}"
303
304
print(f"\nCreating snapshot for {version}...")
305
print(f"Description: {config['description']}")
306
307
# Create snapshot with version-specific filters
308
snapshot_params = Snapshot(
309
filters=config["filters"],
310
composition_type=CompositionType.KEY_LABEL,
311
retention_period=7776000, # 90 days for release snapshots
312
tags={
313
"Release": version,
314
"Type": "Release",
315
"Description": config["description"],
316
"Created": datetime.datetime.now().isoformat()
317
}
318
)
319
320
try:
321
create_poller = client.snapshots.begin_create(
322
resource_group,
323
store_name,
324
snapshot_name,
325
snapshot_params
326
)
327
328
print(f" ⏳ Creating snapshot...")
329
snapshot = create_poller.result()
330
331
created_snapshots[version] = snapshot
332
333
print(f" ✅ Snapshot created for {version}")
334
print(f" Items: {snapshot.items_count}")
335
print(f" Size: {snapshot.size_in_bytes} bytes")
336
print(f" Expires: {snapshot.expires}")
337
338
except Exception as e:
339
print(f" ❌ Failed to create snapshot for {version}: {e}")
340
created_snapshots[version] = None
341
342
# Summary
343
successful_snapshots = [v for v, s in created_snapshots.items() if s is not None]
344
print(f"\n=== Summary ===")
345
print(f"Successfully created {len(successful_snapshots)} release snapshots:")
346
347
for version in successful_snapshots:
348
snapshot = created_snapshots[version]
349
print(f" - {version}: {snapshot.name} ({snapshot.items_count} items)")
350
351
return created_snapshots
352
353
# Create release snapshots
354
release_snapshots = create_release_snapshots()
355
```
356
357
### Environment-Specific Snapshots
358
359
```python { .api }
360
def create_environment_snapshots():
361
"""Create snapshots for different environments."""
362
363
environments = ["Development", "Staging", "Production"]
364
365
print("=== Creating Environment Snapshots ===")
366
367
env_snapshots = {}
368
369
for env in environments:
370
print(f"\nProcessing {env} environment...")
371
372
# Create environment-specific filters
373
env_filters = [
374
KeyValueFilter(key="*", label=env), # All keys with env label
375
KeyValueFilter(key=f"{env}:*") # Keys specific to environment
376
]
377
378
# Adjust retention based on environment
379
retention_days = {
380
"Development": 7, # 7 days
381
"Staging": 30, # 30 days
382
"Production": 90 # 90 days
383
}
384
385
retention_seconds = retention_days[env] * 24 * 60 * 60
386
387
snapshot_params = Snapshot(
388
filters=env_filters,
389
composition_type=CompositionType.KEY_LABEL,
390
retention_period=retention_seconds,
391
tags={
392
"Environment": env,
393
"Type": "Environment Backup",
394
"Retention": f"{retention_days[env]} days",
395
"Created": datetime.datetime.now().isoformat()
396
}
397
)
398
399
snapshot_name = f"env-{env.lower()}-{datetime.datetime.now().strftime('%Y%m%d')}"
400
401
try:
402
create_poller = client.snapshots.begin_create(
403
resource_group,
404
store_name,
405
snapshot_name,
406
snapshot_params
407
)
408
409
snapshot = create_poller.result()
410
env_snapshots[env] = snapshot
411
412
print(f" ✅ {env} snapshot created: {snapshot.name}")
413
print(f" Items: {snapshot.items_count}")
414
print(f" Retention: {retention_days[env]} days")
415
416
except Exception as e:
417
print(f" ❌ Failed to create {env} snapshot: {e}")
418
env_snapshots[env] = None
419
420
return env_snapshots
421
422
# Create environment snapshots
423
env_snapshots = create_environment_snapshots()
424
```
425
426
### Feature Flag Snapshots
427
428
```python { .api }
429
def create_feature_flag_snapshots():
430
"""Create snapshots specifically for feature flags."""
431
432
print("=== Creating Feature Flag Snapshots ===")
433
434
# Feature flag specific filters
435
feature_flag_filters = [
436
KeyValueFilter(key=".appconfig.featureflag/*"), # All feature flags
437
KeyValueFilter(key="FeatureManagement:*") # Alternative feature flag pattern
438
]
439
440
# Create snapshot for current feature flag state
441
timestamp = datetime.datetime.now().strftime("%Y%m%d-%H%M")
442
snapshot_name = f"feature-flags-{timestamp}"
443
444
snapshot_params = Snapshot(
445
filters=feature_flag_filters,
446
composition_type=CompositionType.KEY_LABEL,
447
retention_period=1209600, # 14 days for feature flags
448
tags={
449
"Type": "Feature Flags",
450
"Purpose": "Feature Flag State Backup",
451
"Timestamp": timestamp
452
}
453
)
454
455
print(f"Creating feature flag snapshot: {snapshot_name}")
456
457
try:
458
create_poller = client.snapshots.begin_create(
459
resource_group,
460
store_name,
461
snapshot_name,
462
snapshot_params
463
)
464
465
snapshot = create_poller.result()
466
467
print(f"✅ Feature flag snapshot created:")
468
print(f" Name: {snapshot.name}")
469
print(f" Items: {snapshot.items_count}")
470
print(f" Size: {snapshot.size_in_bytes} bytes")
471
472
# Analyze feature flag snapshot content
473
if snapshot.items_count > 0:
474
print(f" 📊 Analysis:")
475
print(f" - Captured {snapshot.items_count} feature flag configurations")
476
print(f" - Snapshot size: {snapshot.size_in_bytes:,} bytes")
477
print(f" - Average bytes per flag: {snapshot.size_in_bytes // snapshot.items_count:.0f}")
478
479
return snapshot
480
481
except Exception as e:
482
print(f"❌ Failed to create feature flag snapshot: {e}")
483
return None
484
485
# Create feature flag snapshot
486
ff_snapshot = create_feature_flag_snapshots()
487
```
488
489
### Snapshot Monitoring and Management
490
491
```python { .api }
492
def monitor_snapshot_status():
493
"""Monitor the status of existing snapshots."""
494
495
print("=== Snapshot Status Monitor ===")
496
497
# Note: There's no list operation in the ARM API for snapshots
498
# In practice, you would track snapshot names separately
499
# This example shows how to check status of known snapshots
500
501
known_snapshots = [
502
"full-backup-snapshot",
503
"release-v1.0",
504
"release-v1.1-beta",
505
"env-production-20241201",
506
"feature-flags-20241201-1400"
507
]
508
509
snapshot_status = {}
510
511
for snapshot_name in known_snapshots:
512
try:
513
snapshot = client.snapshots.get(resource_group, store_name, snapshot_name)
514
515
snapshot_status[snapshot_name] = {
516
"status": snapshot.status,
517
"created": snapshot.created,
518
"expires": snapshot.expires,
519
"items_count": snapshot.items_count,
520
"size_in_bytes": snapshot.size_in_bytes,
521
"tags": snapshot.tags
522
}
523
524
print(f"\n📸 Snapshot: {snapshot_name}")
525
print(f" Status: {snapshot.status}")
526
print(f" Created: {snapshot.created}")
527
print(f" Expires: {snapshot.expires}")
528
print(f" Items: {snapshot.items_count}")
529
print(f" Size: {snapshot.size_in_bytes:,} bytes")
530
531
# Check expiration warning
532
if snapshot.expires:
533
days_until_expiry = (snapshot.expires - datetime.datetime.now(datetime.timezone.utc)).days
534
if days_until_expiry <= 7:
535
print(f" ⚠️ Expires in {days_until_expiry} days!")
536
elif days_until_expiry <= 1:
537
print(f" 🚨 Expires in {days_until_expiry} day(s)!")
538
539
# Status-specific messages
540
if snapshot.status == SnapshotStatus.PROVISIONING:
541
print(f" ⏳ Still being created...")
542
elif snapshot.status == SnapshotStatus.READY:
543
print(f" ✅ Ready for use")
544
elif snapshot.status == SnapshotStatus.ARCHIVED:
545
print(f" 📦 Archived (past retention period)")
546
elif snapshot.status == SnapshotStatus.FAILED:
547
print(f" ❌ Creation failed")
548
549
except Exception as e:
550
print(f"\n❌ Error checking {snapshot_name}: {e}")
551
snapshot_status[snapshot_name] = {"error": str(e)}
552
553
# Summary statistics
554
print(f"\n=== Summary ===")
555
556
status_counts = {}
557
total_items = 0
558
total_size = 0
559
560
for name, info in snapshot_status.items():
561
if "error" not in info:
562
status = info["status"]
563
status_counts[status] = status_counts.get(status, 0) + 1
564
total_items += info["items_count"]
565
total_size += info["size_in_bytes"]
566
567
print(f"Total snapshots checked: {len(known_snapshots)}")
568
for status, count in status_counts.items():
569
print(f" {status}: {count}")
570
571
if total_items > 0:
572
print(f"Total configuration items: {total_items:,}")
573
print(f"Total storage used: {total_size:,} bytes ({total_size/1024/1024:.2f} MB)")
574
575
return snapshot_status
576
577
# Monitor snapshots
578
snapshot_monitoring = monitor_snapshot_status()
579
```
580
581
### Snapshot Comparison and Analysis
582
583
```python { .api }
584
def analyze_snapshot_differences():
585
"""Analyze differences between snapshots for change tracking."""
586
587
print("=== Snapshot Analysis ===")
588
589
# Compare two snapshots (in practice, you might have utilities to extract and compare)
590
snapshots_to_compare = ["release-v1.0", "release-v1.1-beta"]
591
592
snapshot_data = {}
593
594
for snapshot_name in snapshots_to_compare:
595
try:
596
snapshot = client.snapshots.get(resource_group, store_name, snapshot_name)
597
snapshot_data[snapshot_name] = snapshot
598
599
print(f"\n📊 Snapshot: {snapshot_name}")
600
print(f" Created: {snapshot.created}")
601
print(f" Status: {snapshot.status}")
602
print(f" Items: {snapshot.items_count}")
603
print(f" Size: {snapshot.size_in_bytes:,} bytes")
604
605
if snapshot.tags:
606
print(f" Tags:")
607
for key, value in snapshot.tags.items():
608
print(f" {key}: {value}")
609
610
except Exception as e:
611
print(f"❌ Error loading {snapshot_name}: {e}")
612
613
# Basic comparison (metadata only via ARM API)
614
if len(snapshot_data) == 2:
615
snap1_name, snap2_name = snapshots_to_compare
616
snap1 = snapshot_data.get(snap1_name)
617
snap2 = snapshot_data.get(snap2_name)
618
619
if snap1 and snap2:
620
print(f"\n🔍 Comparison: {snap1_name} vs {snap2_name}")
621
622
item_diff = snap2.items_count - snap1.items_count
623
size_diff = snap2.size_in_bytes - snap1.size_in_bytes
624
625
print(f" Item count change: {item_diff:+d}")
626
print(f" Size change: {size_diff:+,} bytes")
627
628
if item_diff > 0:
629
print(f" 📈 {item_diff} new configuration items added")
630
elif item_diff < 0:
631
print(f" 📉 {abs(item_diff)} configuration items removed")
632
else:
633
print(f" ➡️ Same number of items (possible updates)")
634
635
# Time analysis
636
if snap1.created and snap2.created:
637
time_diff = snap2.created - snap1.created
638
print(f" Time between snapshots: {time_diff}")
639
640
# Recommendations based on analysis
641
print(f"\n💡 Recommendations:")
642
643
ready_snapshots = [name for name, data in snapshot_data.items()
644
if data and data.status == SnapshotStatus.READY]
645
646
if len(ready_snapshots) >= 2:
647
print(f" - Use data plane API to compare actual configuration content")
648
print(f" - Consider automated snapshot comparison for CI/CD pipelines")
649
print(f" - Set up alerts for significant configuration changes")
650
651
return snapshot_data
652
653
# Analyze snapshots
654
snapshot_analysis = analyze_snapshot_differences()
655
```
656
657
### Snapshot Lifecycle Management
658
659
```python { .api }
660
def manage_snapshot_lifecycle():
661
"""Implement a comprehensive snapshot lifecycle management strategy."""
662
663
print("=== Snapshot Lifecycle Management ===")
664
665
# Define lifecycle policies
666
policies = {
667
"daily_backup": {
668
"schedule": "daily",
669
"retention_days": 7,
670
"filters": [KeyValueFilter(key="*")], # All config
671
"name_pattern": "daily-backup-{date}"
672
},
673
"weekly_backup": {
674
"schedule": "weekly",
675
"retention_days": 30,
676
"filters": [KeyValueFilter(key="*")],
677
"name_pattern": "weekly-backup-{date}"
678
},
679
"release_snapshot": {
680
"schedule": "on_demand",
681
"retention_days": 90,
682
"filters": [KeyValueFilter(key="*", label="Production")],
683
"name_pattern": "release-{version}"
684
},
685
"feature_flags": {
686
"schedule": "twice_daily",
687
"retention_days": 14,
688
"filters": [KeyValueFilter(key=".appconfig.featureflag/*")],
689
"name_pattern": "ff-{datetime}"
690
}
691
}
692
693
current_date = datetime.datetime.now()
694
695
# Simulate creating snapshots based on policies
696
for policy_name, policy in policies.items():
697
print(f"\n📋 Policy: {policy_name}")
698
print(f" Schedule: {policy['schedule']}")
699
print(f" Retention: {policy['retention_days']} days")
700
701
# Generate snapshot name based on pattern
702
if "{date}" in policy["name_pattern"]:
703
snapshot_name = policy["name_pattern"].replace("{date}", current_date.strftime("%Y%m%d"))
704
elif "{datetime}" in policy["name_pattern"]:
705
snapshot_name = policy["name_pattern"].replace("{datetime}", current_date.strftime("%Y%m%d-%H%M"))
706
else:
707
snapshot_name = policy["name_pattern"]
708
709
print(f" Snapshot name: {snapshot_name}")
710
711
# Create snapshot based on policy
712
retention_seconds = policy["retention_days"] * 24 * 60 * 60
713
714
snapshot_params = Snapshot(
715
filters=policy["filters"],
716
composition_type=CompositionType.KEY_LABEL,
717
retention_period=retention_seconds,
718
tags={
719
"Policy": policy_name,
720
"Schedule": policy["schedule"],
721
"AutoCreated": "True",
722
"Created": current_date.isoformat()
723
}
724
)
725
726
# In a real implementation, you would check if it's time to create the snapshot
727
# based on the schedule. For this example, we'll just show the process.
728
print(f" ✅ Would create snapshot with {len(policy['filters'])} filter(s)")
729
730
# Cleanup old snapshots (simulation)
731
print(f"\n🧹 Cleanup Process:")
732
print(f" - Check snapshots past retention period")
733
print(f" - Archive or delete expired snapshots")
734
print(f" - Notify administrators of cleanup actions")
735
736
# Best practices summary
737
print(f"\n📚 Best Practices:")
738
print(f" - Automate snapshot creation based on business needs")
739
print(f" - Balance retention periods with storage costs")
740
print(f" - Use descriptive naming conventions")
741
print(f" - Tag snapshots for easy identification and management")
742
print(f" - Monitor snapshot creation success/failure")
743
print(f" - Document snapshot and restore procedures")
744
745
# Run lifecycle management
746
manage_snapshot_lifecycle()
747
```
748
749
## Asynchronous Operations
750
751
```python { .api }
752
from azure.mgmt.appconfiguration.aio import AppConfigurationManagementClient
753
from azure.identity.aio import DefaultAzureCredential
754
import asyncio
755
756
async def async_snapshot_operations():
757
"""Demonstrate asynchronous snapshot operations."""
758
759
credential = DefaultAzureCredential()
760
761
async with AppConfigurationManagementClient(credential, "subscription-id") as client:
762
763
# Create multiple snapshots concurrently
764
snapshot_configs = [
765
{
766
"name": "async-full-backup",
767
"filters": [KeyValueFilter(key="*")],
768
"tags": {"Type": "Full Backup", "Async": "True"}
769
},
770
{
771
"name": "async-app-config",
772
"filters": [KeyValueFilter(key="MyApp:*")],
773
"tags": {"Type": "App Config", "Async": "True"}
774
},
775
{
776
"name": "async-feature-flags",
777
"filters": [KeyValueFilter(key=".appconfig.featureflag/*")],
778
"tags": {"Type": "Feature Flags", "Async": "True"}
779
}
780
]
781
782
print("Creating snapshots concurrently...")
783
784
# Start all snapshot creations
785
create_tasks = []
786
for config in snapshot_configs:
787
snapshot_params = Snapshot(
788
filters=config["filters"],
789
composition_type=CompositionType.KEY_LABEL,
790
retention_period=2592000, # 30 days
791
tags=config["tags"]
792
)
793
794
task = client.snapshots.begin_create(
795
resource_group,
796
store_name,
797
config["name"],
798
snapshot_params
799
)
800
create_tasks.append((config["name"], task))
801
802
# Wait for all create operations to start
803
pollers = []
804
for name, task in create_tasks:
805
try:
806
poller = await task
807
pollers.append((name, poller))
808
print(f"✓ Started creation of {name}")
809
except Exception as e:
810
print(f"✗ Failed to start {name}: {e}")
811
812
# Wait for all snapshots to complete
813
print("Waiting for snapshot creation to complete...")
814
815
results = []
816
for name, poller in pollers:
817
try:
818
snapshot = await poller.result()
819
results.append(snapshot)
820
print(f"✅ {name}: {snapshot.items_count} items, {snapshot.size_in_bytes} bytes")
821
except Exception as e:
822
print(f"❌ {name} failed: {e}")
823
824
print(f"\nCompleted {len(results)} snapshots successfully")
825
826
# Run async operations
827
asyncio.run(async_snapshot_operations())
828
```
829
830
## Integration with CI/CD Pipelines
831
832
```python { .api }
833
def integrate_with_cicd():
834
"""Show how to integrate snapshots with CI/CD pipelines."""
835
836
print("=== CI/CD Pipeline Integration ===")
837
838
pipeline_examples = {
839
"pre_deployment": {
840
"trigger": "Before deployment",
841
"purpose": "Backup current state before changes",
842
"script": """
843
# Pre-deployment snapshot
844
az appconfig snapshot create \\
845
--name "pre-deploy-$(Build.BuildNumber)" \\
846
--config-store-name {store_name} \\
847
--filters "key=*&label=Production" \\
848
--retention-period 604800 # 7 days
849
"""
850
},
851
"post_deployment": {
852
"trigger": "After successful deployment",
853
"purpose": "Capture new configuration state",
854
"script": """
855
# Post-deployment snapshot
856
az appconfig snapshot create \\
857
--name "post-deploy-$(Build.BuildNumber)" \\
858
--config-store-name {store_name} \\
859
--filters "key=*&label=Production" \\
860
--retention-period 2592000 # 30 days
861
"""
862
},
863
"rollback": {
864
"trigger": "On deployment failure",
865
"purpose": "Reference point for rollback procedures",
866
"script": """
867
# Get latest pre-deployment snapshot
868
SNAPSHOT_NAME=$(az appconfig snapshot list \\
869
--config-store-name {store_name} \\
870
--query "[?contains(name, 'pre-deploy')].name | [-1]" -o tsv)
871
872
# Use snapshot as rollback reference
873
echo "Rollback to snapshot: $SNAPSHOT_NAME"
874
"""
875
}
876
}
877
878
for stage, config in pipeline_examples.items():
879
print(f"\n🔄 {stage.replace('_', ' ').title()}:")
880
print(f" Trigger: {config['trigger']}")
881
print(f" Purpose: {config['purpose']}")
882
print(f" Script:")
883
print(f"{config['script']}")
884
885
# Python SDK equivalent for CI/CD
886
print(f"\n🐍 Python SDK Example for CI/CD:")
887
print("""
888
import os
889
from azure.mgmt.appconfiguration import AppConfigurationManagementClient
890
from azure.mgmt.appconfiguration.models import Snapshot, KeyValueFilter, CompositionType
891
from azure.identity import DefaultAzureCredential
892
893
def create_deployment_snapshot(build_number: str, stage: str):
894
client = AppConfigurationManagementClient(
895
DefaultAzureCredential(),
896
os.environ["AZURE_SUBSCRIPTION_ID"]
897
)
898
899
snapshot_name = f"{stage}-{build_number}"
900
filters = [KeyValueFilter(key="*", label="Production")]
901
902
snapshot_params = Snapshot(
903
filters=filters,
904
composition_type=CompositionType.KEY_LABEL,
905
retention_period=2592000, # 30 days
906
tags={
907
"BuildNumber": build_number,
908
"Stage": stage,
909
"Pipeline": "CI/CD"
910
}
911
)
912
913
poller = client.snapshots.begin_create(
914
os.environ["RESOURCE_GROUP"],
915
os.environ["CONFIG_STORE_NAME"],
916
snapshot_name,
917
snapshot_params
918
)
919
920
return poller.result()
921
922
# Usage in pipeline
923
if __name__ == "__main__":
924
build_num = os.environ.get("BUILD_BUILDNUMBER", "manual")
925
stage = os.environ.get("PIPELINE_STAGE", "pre-deploy")
926
927
snapshot = create_deployment_snapshot(build_num, stage)
928
print(f"Created snapshot: {snapshot.name}")
929
""")
930
931
integrate_with_cicd()
932
```