Microsoft Azure App Configuration Management Client Library for Python
—
Quality
Pending
Does it follow best practices?
Impact
Pending
No eval scenarios have been run
The ReplicasOperations class provides comprehensive management capabilities for Azure App Configuration store replicas. Replicas enable multi-region deployment of your configuration stores, providing improved availability and performance for geographically distributed applications.
class ReplicasOperations:
"""
Operations for managing App Configuration store replicas.
Replicas allow you to create read-only copies of your configuration store in different
Azure regions, enabling improved performance and availability for distributed applications.
Configuration data is automatically synchronized from the primary store to all replicas.
"""def list_by_configuration_store(
self,
resource_group_name: str,
config_store_name: str,
skip_token: Optional[str] = None,
**kwargs: Any
) -> ItemPaged[Replica]:
"""
Lists all replicas for a specific App Configuration store.
Args:
resource_group_name: The name of the resource group containing the store.
config_store_name: The name of the configuration store.
skip_token: A skip token to retrieve the next set of results, if any.
**kwargs: Additional keyword arguments for the request.
Returns:
ItemPaged[Replica]: A paged collection of replicas for the store.
Example:
>>> replicas = client.replicas.list_by_configuration_store("my-rg", "my-store")
>>> for replica in replicas:
... print(f"Replica: {replica.name} in {replica.location}")
... print(f"Status: {replica.provisioning_state}")
... print(f"Endpoint: {replica.endpoint}")
"""def get(
self,
resource_group_name: str,
config_store_name: str,
replica_name: str,
**kwargs: Any
) -> Replica:
"""
Gets the properties of a specific replica.
Args:
resource_group_name: The name of the resource group containing the store.
config_store_name: The name of the configuration store.
replica_name: The name of the replica.
**kwargs: Additional keyword arguments for the request.
Returns:
Replica: The replica details and properties.
Example:
>>> replica = client.replicas.get("my-rg", "my-store", "west-us-replica")
>>> print(f"Replica location: {replica.location}")
>>> print(f"Provisioning state: {replica.provisioning_state}")
>>> print(f"Endpoint: {replica.endpoint}")
"""def begin_create(
self,
resource_group_name: str,
config_store_name: str,
replica_name: str,
replica_creation_parameters: Replica,
**kwargs: Any
) -> LROPoller[Replica]:
"""
Creates a new replica for an App Configuration store.
Args:
resource_group_name: The name of the resource group containing the store.
config_store_name: The name of the configuration store.
replica_name: The name for the new replica (must be unique within the store).
replica_creation_parameters: Parameters for creating the replica.
**kwargs: Additional keyword arguments for the request.
Returns:
LROPoller[Replica]: A poller for the long-running create operation.
Note:
Replica creation is asynchronous and may take several minutes to complete.
The replica will automatically synchronize configuration data from the primary store.
Example:
>>> from azure.mgmt.appconfiguration.models import Replica
>>> replica_params = Replica(location="West US 2")
>>> create_poller = client.replicas.begin_create(
... "my-rg", "my-store", "westus2-replica", replica_params
... )
>>> replica = create_poller.result() # Wait for completion
>>> print(f"Created replica: {replica.name} at {replica.endpoint}")
"""def begin_delete(
self,
resource_group_name: str,
config_store_name: str,
replica_name: str,
**kwargs: Any
) -> LROPoller[None]:
"""
Deletes a replica from an App Configuration store.
Args:
resource_group_name: The name of the resource group containing the store.
config_store_name: The name of the configuration store.
replica_name: The name of the replica to delete.
**kwargs: Additional keyword arguments for the request.
Returns:
LROPoller[None]: A poller for the long-running delete operation.
Warning:
Deleting a replica is permanent and cannot be undone. Applications using
the replica endpoint will need to be reconfigured to use different endpoints.
Example:
>>> delete_poller = client.replicas.begin_delete(
... "my-rg", "my-store", "old-replica"
... )
>>> delete_poller.wait() # Wait for deletion to complete
>>> print("Replica deleted successfully")
"""class Replica:
"""
Represents an App Configuration store replica.
Attributes:
id (str): The resource ID of the replica (read-only).
name (str): The name of the replica (read-only).
type (str): The resource type (read-only).
location (str): The Azure region where the replica is located.
endpoint (str): The endpoint URL for the replica (read-only).
provisioning_state (ReplicaProvisioningState): The provisioning state (read-only).
system_data (SystemData): System metadata for the replica (read-only).
"""
def __init__(
self,
*,
location: str,
**kwargs: Any
) -> None:
"""
Initialize a Replica instance.
Args:
location: The Azure region for the replica (e.g., "West US 2", "East Asia").
**kwargs: Additional properties.
"""class ReplicaProvisioningState:
"""
Enumeration of replica provisioning states.
Values:
CREATING: Replica is being created.
SUCCEEDED: Replica has been successfully created and is operational.
DELETING: Replica is being deleted.
FAILED: Replica creation or operation has failed.
CANCELED: Replica operation was canceled.
"""
CREATING = "Creating"
SUCCEEDED = "Succeeded"
DELETING = "Deleting"
FAILED = "Failed"
CANCELED = "Canceled"from azure.mgmt.appconfiguration import AppConfigurationManagementClient
from azure.mgmt.appconfiguration.models import Replica, ReplicaProvisioningState
from azure.identity import DefaultAzureCredential
# Initialize client
credential = DefaultAzureCredential()
client = AppConfigurationManagementClient(credential, "subscription-id")
resource_group = "my-resource-group"
store_name = "my-config-store"
# Create replicas in multiple regions for global distribution
print("Creating replicas for global distribution...")
regions = {
"westus2-replica": "West US 2",
"eastus2-replica": "East US 2",
"westeurope-replica": "West Europe",
"southeastasia-replica": "Southeast Asia"
}
created_replicas = {}
for replica_name, location in regions.items():
print(f"Creating replica in {location}...")
replica_params = Replica(location=location)
try:
create_poller = client.replicas.begin_create(
resource_group,
store_name,
replica_name,
replica_params
)
# Monitor creation progress
print(f" Replica creation started for {location}")
replica = create_poller.result()
created_replicas[replica_name] = replica
print(f" ✓ Replica created: {replica.endpoint}")
except Exception as e:
print(f" ✗ Failed to create replica in {location}: {e}")
print(f"Successfully created {len(created_replicas)} replicas")def deploy_global_replica_strategy():
"""Deploy replicas using a strategic global distribution approach."""
# Define regions based on application user distribution
deployment_strategy = {
"primary": {
"region": "East US",
"description": "Primary region - no replica needed"
},
"americas": {
"region": "West US 2",
"replica_name": "americas-west-replica",
"description": "Serves western Americas"
},
"europe": {
"region": "West Europe",
"replica_name": "europe-replica",
"description": "Serves European users"
},
"asia": {
"region": "Southeast Asia",
"replica_name": "asia-replica",
"description": "Serves Asian Pacific users"
}
}
print("=== Global Replica Deployment Strategy ===")
# Check current replicas
print("\n1. Current replica status:")
existing_replicas = client.replicas.list_by_configuration_store(
resource_group, store_name
)
existing_names = set()
for replica in existing_replicas:
existing_names.add(replica.name)
print(f" Existing: {replica.name} in {replica.location} ({replica.provisioning_state})")
# Deploy missing replicas
print("\n2. Deploying missing replicas:")
deployment_results = {}
for region_key, config in deployment_strategy.items():
if region_key == "primary":
continue
replica_name = config["replica_name"]
if replica_name in existing_names:
print(f" ✓ {replica_name} already exists")
continue
print(f" Creating {replica_name} in {config['region']}...")
print(f" Purpose: {config['description']}")
try:
replica_params = Replica(location=config["region"])
create_poller = client.replicas.begin_create(
resource_group,
store_name,
replica_name,
replica_params
)
# Store poller for batch monitoring
deployment_results[replica_name] = {
"poller": create_poller,
"config": config
}
except Exception as e:
print(f" ✗ Failed to start creation: {e}")
deployment_results[replica_name] = {"error": str(e)}
# Monitor all deployments
print("\n3. Monitoring deployments:")
for replica_name, result in deployment_results.items():
if "error" in result:
print(f" ✗ {replica_name}: {result['error']}")
continue
try:
poller = result["poller"]
print(f" Waiting for {replica_name}...")
replica = poller.result() # This will wait for completion
print(f" ✓ {replica_name} deployed successfully")
print(f" Endpoint: {replica.endpoint}")
except Exception as e:
print(f" ✗ {replica_name} deployment failed: {e}")
# Final status check
print("\n4. Final deployment status:")
final_replicas = client.replicas.list_by_configuration_store(
resource_group, store_name
)
for replica in final_replicas:
region_info = next(
(f" ({config['description']})"
for config in deployment_strategy.values()
if config.get("region") == replica.location),
""
)
print(f" {replica.name}: {replica.location}{region_info} - {replica.provisioning_state}")
# Execute deployment strategy
deploy_global_replica_strategy()def monitor_replica_health():
"""Monitor the health and status of all replicas."""
print("=== Replica Health Monitor ===")
replicas = client.replicas.list_by_configuration_store(
resource_group, store_name
)
replica_list = list(replicas)
if not replica_list:
print("No replicas found for this configuration store")
return
print(f"Monitoring {len(replica_list)} replicas:")
health_summary = {
"total": len(replica_list),
"healthy": 0,
"creating": 0,
"failed": 0,
"deleting": 0,
"unknown": 0
}
for replica in replica_list:
state = replica.provisioning_state
print(f"\n📍 Replica: {replica.name}")
print(f" Location: {replica.location}")
print(f" Endpoint: {replica.endpoint}")
print(f" Status: {state}")
# Categorize health status
if state == ReplicaProvisioningState.SUCCEEDED:
health_summary["healthy"] += 1
print(" 🟢 Status: Healthy")
elif state == ReplicaProvisioningState.CREATING:
health_summary["creating"] += 1
print(" 🟡 Status: Creating...")
elif state == ReplicaProvisioningState.FAILED:
health_summary["failed"] += 1
print(" 🔴 Status: Failed")
elif state == ReplicaProvisioningState.DELETING:
health_summary["deleting"] += 1
print(" 🟠 Status: Deleting...")
else:
health_summary["unknown"] += 1
print(" ⚪ Status: Unknown")
# Additional health checks could include:
# - Network connectivity tests
# - Configuration synchronization status
# - Performance metrics
if replica.system_data:
print(f" Created: {replica.system_data.created_at}")
print(f" Created by: {replica.system_data.created_by}")
# Health summary
print(f"\n=== Health Summary ===")
print(f"Total Replicas: {health_summary['total']}")
print(f"🟢 Healthy: {health_summary['healthy']}")
print(f"🟡 Creating: {health_summary['creating']}")
print(f"🔴 Failed: {health_summary['failed']}")
print(f"🟠 Deleting: {health_summary['deleting']}")
if health_summary["failed"] > 0:
print(f"\n⚠️ {health_summary['failed']} replica(s) require attention")
healthy_percentage = (health_summary["healthy"] / health_summary["total"]) * 100
print(f"Health Score: {healthy_percentage:.1f}%")
return health_summary
# Run health monitoring
health_status = monitor_replica_health()def optimize_replica_distribution():
"""Analyze and optimize replica distribution for performance."""
print("=== Replica Distribution Analysis ===")
# Get all replicas
replicas = client.replicas.list_by_configuration_store(
resource_group, store_name
)
replica_locations = []
for replica in replicas:
if replica.provisioning_state == ReplicaProvisioningState.SUCCEEDED:
replica_locations.append({
"name": replica.name,
"location": replica.location,
"endpoint": replica.endpoint
})
print(f"Active replicas: {len(replica_locations)}")
# Define region groups for analysis
region_groups = {
"US East": ["East US", "East US 2", "Central US"],
"US West": ["West US", "West US 2", "West Central US"],
"Europe": ["West Europe", "North Europe", "UK South"],
"Asia": ["Southeast Asia", "East Asia", "Japan East"],
"Other": []
}
# Categorize replicas by region group
distribution = {group: [] for group in region_groups.keys()}
for replica in replica_locations:
location = replica["location"]
assigned = False
for group, regions in region_groups.items():
if location in regions:
distribution[group].append(replica)
assigned = True
break
if not assigned:
distribution["Other"].append(replica)
# Analysis results
print("\nDistribution by region:")
for group, replicas in distribution.items():
print(f" {group}: {len(replicas)} replica(s)")
for replica in replicas:
print(f" - {replica['name']} ({replica['location']})")
# Optimization recommendations
print("\n=== Optimization Recommendations ===")
total_replicas = len(replica_locations)
if total_replicas == 0:
print("⚠️ No active replicas found. Consider creating replicas for:")
print(" - Geographic regions where your users are located")
print(" - Disaster recovery purposes")
return
# Check for regional gaps
empty_regions = [group for group, replicas in distribution.items()
if len(replicas) == 0 and group != "Other"]
if empty_regions:
print(f"🌍 Geographic gaps detected in: {', '.join(empty_regions)}")
print(" Consider adding replicas in these regions for better global coverage")
# Check for over-concentration
max_replicas_per_region = max(len(replicas) for replicas in distribution.values())
if max_replicas_per_region > 2:
concentrated_regions = [group for group, replicas in distribution.items()
if len(replicas) > 2]
print(f"📍 High concentration in: {', '.join(concentrated_regions)}")
print(" Consider distributing replicas more evenly across regions")
# Cost optimization
if total_replicas > 5:
print(f"💰 Cost optimization: {total_replicas} replicas may be more than needed")
print(" Review usage patterns and consider consolidating underused replicas")
# Performance recommendations
print("\n🚀 Performance recommendations:")
print(" - Place replicas close to your application users")
print(" - Use replica endpoints in application configuration")
print(" - Implement failover logic for high availability")
print(" - Monitor replica synchronization lag")
return distribution
# Run optimization analysis
distribution_analysis = optimize_replica_distribution()def setup_disaster_recovery_replicas():
"""Set up replicas specifically for disaster recovery purposes."""
print("=== Disaster Recovery Replica Setup ===")
# Get primary store details
primary_store = client.configuration_stores.get(resource_group, store_name)
primary_region = primary_store.location
print(f"Primary store location: {primary_region}")
# Define disaster recovery regions based on primary location
dr_regions = {
"East US": ["West US 2", "Central US"],
"East US 2": ["West US 2", "South Central US"],
"West US": ["East US 2", "Central US"],
"West US 2": ["East US 2", "South Central US"],
"Central US": ["East US 2", "West US 2"],
"West Europe": ["North Europe", "UK South"],
"North Europe": ["West Europe", "UK West"],
"Southeast Asia": ["East Asia", "Australia East"]
}
recommended_dr_regions = dr_regions.get(primary_region, ["East US 2", "West US 2"])
print(f"Recommended DR regions for {primary_region}: {recommended_dr_regions}")
# Check existing replicas
existing_replicas = client.replicas.list_by_configuration_store(
resource_group, store_name
)
existing_locations = {replica.location for replica in existing_replicas
if replica.provisioning_state == ReplicaProvisioningState.SUCCEEDED}
print(f"Existing replica locations: {existing_locations}")
# Create DR replicas
dr_replicas_needed = [region for region in recommended_dr_regions
if region not in existing_locations]
if not dr_replicas_needed:
print("✓ Disaster recovery replicas already exist")
return
print(f"Creating DR replicas in: {dr_replicas_needed}")
dr_results = []
for i, dr_region in enumerate(dr_replicas_needed):
replica_name = f"dr-replica-{i+1}"
print(f"Creating DR replica: {replica_name} in {dr_region}")
try:
replica_params = Replica(location=dr_region)
create_poller = client.replicas.begin_create(
resource_group,
store_name,
replica_name,
replica_params
)
print(f" ⏳ Creation started...")
replica = create_poller.result()
dr_results.append({
"name": replica.name,
"location": replica.location,
"endpoint": replica.endpoint,
"status": "success"
})
print(f" ✓ DR replica created: {replica.endpoint}")
except Exception as e:
print(f" ✗ Failed to create DR replica: {e}")
dr_results.append({
"name": replica_name,
"location": dr_region,
"status": "failed",
"error": str(e)
})
# DR readiness summary
print(f"\n=== DR Readiness Summary ===")
successful_dr = [r for r in dr_results if r["status"] == "success"]
print(f"Primary region: {primary_region}")
print(f"DR replicas created: {len(successful_dr)}")
for replica in successful_dr:
print(f" - {replica['name']}: {replica['location']}")
print(f" Endpoint: {replica['endpoint']}")
if len(successful_dr) >= 1:
print("✅ Basic disaster recovery capability established")
if len(successful_dr) >= 2:
print("✅ Enhanced disaster recovery with multiple regions")
# Generate DR runbook
print(f"\n=== Disaster Recovery Runbook ===")
print("In case of primary region failure:")
print("1. Verify primary store is inaccessible")
print("2. Update application configuration to use DR endpoints:")
for replica in successful_dr:
print(f" - {replica['endpoint']}")
print("3. Monitor application functionality")
print("4. When primary region recovers, switch back to primary endpoint")
print("5. Verify configuration synchronization")
return dr_results
# Set up disaster recovery
dr_setup_results = setup_disaster_recovery_replicas()def maintain_replicas():
"""Perform maintenance tasks on replicas."""
print("=== Replica Maintenance ===")
# Get all replicas
replicas = client.replicas.list_by_configuration_store(
resource_group, store_name
)
replica_list = list(replicas)
maintenance_actions = []
for replica in replica_list:
print(f"\nAnalyzing replica: {replica.name}")
print(f" Location: {replica.location}")
print(f" Status: {replica.provisioning_state}")
# Check for failed replicas
if replica.provisioning_state == ReplicaProvisioningState.FAILED:
maintenance_actions.append({
"action": "delete_failed",
"replica": replica,
"reason": "Replica is in failed state and should be removed"
})
print(f" 🔴 Action needed: Remove failed replica")
# Check for long-running creation
elif replica.provisioning_state == ReplicaProvisioningState.CREATING:
# In a real scenario, you might check creation timestamp
maintenance_actions.append({
"action": "monitor_creation",
"replica": replica,
"reason": "Replica has been creating for extended period"
})
print(f" 🟡 Action needed: Monitor creation progress")
# Check for successful replicas (health check placeholder)
elif replica.provisioning_state == ReplicaProvisioningState.SUCCEEDED:
print(f" ✅ Replica is healthy")
# Execute maintenance actions
if maintenance_actions:
print(f"\n=== Maintenance Actions ({len(maintenance_actions)}) ===")
for action in maintenance_actions:
print(f"\nAction: {action['action']}")
print(f"Replica: {action['replica'].name}")
print(f"Reason: {action['reason']}")
if action['action'] == 'delete_failed':
response = input(f"Delete failed replica '{action['replica'].name}'? (y/N): ")
if response.lower() == 'y':
try:
delete_poller = client.replicas.begin_delete(
resource_group,
store_name,
action['replica'].name
)
delete_poller.wait()
print(f" ✓ Deleted failed replica: {action['replica'].name}")
except Exception as e:
print(f" ✗ Failed to delete replica: {e}")
else:
print(" Skipped deletion")
elif action['action'] == 'monitor_creation':
# Get updated status
try:
current_replica = client.replicas.get(
resource_group, store_name, action['replica'].name
)
print(f" Current status: {current_replica.provisioning_state}")
if current_replica.provisioning_state == ReplicaProvisioningState.SUCCEEDED:
print(f" ✅ Replica creation completed successfully")
elif current_replica.provisioning_state == ReplicaProvisioningState.FAILED:
print(f" ❌ Replica creation failed")
except Exception as e:
print(f" ✗ Could not check replica status: {e}")
else:
print("\n✅ No maintenance actions required")
# Final status summary
print(f"\n=== Final Status ===")
final_replicas = client.replicas.list_by_configuration_store(
resource_group, store_name
)
status_counts = {}
for replica in final_replicas:
state = replica.provisioning_state
status_counts[state] = status_counts.get(state, 0) + 1
for status, count in status_counts.items():
print(f" {status}: {count}")
# Run maintenance
maintain_replicas()from azure.mgmt.appconfiguration.aio import AppConfigurationManagementClient
from azure.identity.aio import DefaultAzureCredential
import asyncio
async def async_replica_operations():
"""Demonstrate asynchronous replica operations."""
credential = DefaultAzureCredential()
async with AppConfigurationManagementClient(credential, "subscription-id") as client:
# Create multiple replicas concurrently
replica_tasks = []
regions = ["West US 2", "West Europe", "Southeast Asia"]
for i, region in enumerate(regions):
replica_name = f"async-replica-{i+1}"
replica_params = Replica(location=region)
task = client.replicas.begin_create(
resource_group, store_name, replica_name, replica_params
)
replica_tasks.append((replica_name, region, task))
print("Creating replicas concurrently...")
# Wait for all creation operations to start
pollers = []
for replica_name, region, task in replica_tasks:
try:
poller = await task
pollers.append((replica_name, region, poller))
print(f"✓ Started creation of {replica_name} in {region}")
except Exception as e:
print(f"✗ Failed to start {replica_name}: {e}")
# Wait for all replicas to be created
print("Waiting for replica creation to complete...")
completed_replicas = []
for replica_name, region, poller in pollers:
try:
replica = await poller.result()
completed_replicas.append(replica)
print(f"✅ {replica_name} created: {replica.endpoint}")
except Exception as e:
print(f"❌ {replica_name} failed: {e}")
# List all replicas
print(f"\nFinal replica list:")
replicas = client.replicas.list_by_configuration_store(
resource_group, store_name
)
async for replica in replicas:
print(f" - {replica.name}: {replica.location} ({replica.provisioning_state})")
# Run async operations
asyncio.run(async_replica_operations())def demonstrate_naming_conventions():
"""Show recommended naming conventions for replicas."""
naming_examples = {
"Geographic": {
"pattern": "{region-code}-replica",
"examples": ["usw2-replica", "euw-replica", "sea-replica"],
"use_case": "Simple geographic identification"
},
"Purpose-based": {
"pattern": "{purpose}-{region}-replica",
"examples": ["dr-usw2-replica", "perf-euw-replica", "test-sea-replica"],
"use_case": "Identify replica purpose and location"
},
"Environment": {
"pattern": "{env}-{region}-replica",
"examples": ["prod-usw2-replica", "stage-euw-replica", "dev-sea-replica"],
"use_case": "Environment-specific replicas"
},
"Sequential": {
"pattern": "replica-{number}",
"examples": ["replica-01", "replica-02", "replica-03"],
"use_case": "Simple sequential numbering"
}
}
print("=== Replica Naming Convention Guidelines ===")
for convention, details in naming_examples.items():
print(f"\n{convention} Convention:")
print(f" Pattern: {details['pattern']}")
print(f" Examples: {', '.join(details['examples'])}")
print(f" Use case: {details['use_case']}")
print(f"\nRecommendations:")
print(f" - Keep names under 64 characters")
print(f" - Use lowercase letters, numbers, and hyphens only")
print(f" - Be consistent across your organization")
print(f" - Include region information for geographic distribution")
print(f" - Consider including purpose or environment when relevant")
demonstrate_naming_conventions()Install with Tessl CLI
npx tessl i tessl/pypi-azure-mgmt-appconfiguration