我正在尝试将 API 连接从源资源组复制到另一个资源组以提供备份 RG。我当前使用的代码正在将 API 连接移动到另一个 RG,因此目标 RG 无法发挥其作为备份 RG 的作用。我稍微修改了代码以处理 20 个 API 连接,因为我目前管理着大约 200 个 API 连接。
from azure.identity import ClientSecretCredential
from azure.mgmt.resource import ResourceManagementClient
# Define your subscription ID and resource group names
subscription_id = [REDACTED]
tenant_id = [REDACTED]
client_id = [REDACTED]
client_secret = [REDACTED]
# Authenticate using Azure Identity
credential = ClientSecretCredential(tenant_id, client_id, client_secret)
resource_client = ResourceManagementClient(credential, subscription_id)
# Resource groups
# API Connections must be retained in source_api
resource_group_name_source = "azdnaeurgrgp1dev_source_api"
resource_group_name_target = "azmgmteurgrgp_backup_api"
# Get all resources in the source resource group
resources = list(resource_client.resources.list_by_resource_group(
resource_group_name_source,
filter="resourceType eq 'Microsoft.Web/connections'"
))
# Debug: Print the resources retrieved
for resource in resources:
print(resource.id)
# Prepare the target resource group ID
target_resource_group_id = f"/subscriptions/{subscription_id}/resourceGroups/{resource_group_name_target}"
# Move resources in batches of 20
for i in range(0, len(resources), 20):
batch_resources = resources[i:i+20]
resource_ids = [resource.id for resource in batch_resources]
# Move resources to the target resource group
move_resources_poller = resource_client.resources.begin_move_resources(
resource_group_name_source,
{
"targetResourceGroup": target_resource_group_id,
"resources": resource_ids
}
)
# Wait for the move operation to complete
move_resources_poller.result()
print(f"Moved {len(resource_ids)} resources.")
print("All resources moved successfully to", resource_group_name_target)
理想情况下,我希望 API 连接能够被复制而不是被移动。我尝试使用聊天 GPT 来调整代码,但它只是再次建议使用
update
使用 Python 将 API 连接从 Azure 中的资源组 A 复制到资源组 B。
要复制 API 连接而不是移动它们,需要使用与现有脚本不同的方法。 Azure API 本身不支持在资源组之间克隆或复制资源。您需要在目标资源组中手动建立 API 连接的新实例,镜像原始资源组中的配置。
对于源资源组中的每个 API 连接,检索其配置详细信息。然后,使用获取的配置数据在目标资源组中建立新的API连接。
源 RG 中我的 API 连接。
我更新的Python配置:
from azure.identity import ClientSecretCredential
from azure.mgmt.resource import ResourceManagementClient
import requests
# Define your subscription ID and resource group names
subscription_id = ""
client_id = ""
tenant_id = ""
client_secret = ""
# Authenticate using Azure Identity
credential = ClientSecretCredential(tenant_id, client_id, client_secret)
resource_client = ResourceManagementClient(credential, subscription_id)
# Resource groups
resource_group_name_source = "vinay-rg"
resource_group_name_target = "vinay-tar"
# Get all resources in the source resource group
resources = list(resource_client.resources.list_by_resource_group(
resource_group_name_source,
filter="resourceType eq 'Microsoft.Web/connections'"
))
# Function to get API connection details
def get_api_connection_details(subscription_id, resource_group, resource_name):
token = credential.get_token('https://management.azure.com/.default').token
endpoint = f"https://management.azure.com/subscriptions/{subscription_id}/resourceGroups/{resource_group}/providers/Microsoft.Web/connections/{resource_name}?api-version=2016-06-01"
headers = {'Authorization': 'Bearer ' + token}
response = requests.get(endpoint, headers=headers)
if response.status_code != 200:
print(f"Failed to fetch '{resource_name}' details: {response.status_code} - {response.text}")
return response.json()
# Function to create or update an API connection with parameter values included
def create_or_update_api_connection(subscription_id, resource_group, resource_name, properties, location, parameter_values):
properties["parameterValueSet"] = {
"name": "connectionstringauth",
"values": parameter_values
}
payload = {
"location": location,
"properties": properties
}
token = credential.get_token('https://management.azure.com/.default').token
endpoint = f"https://management.azure.com/subscriptions/{subscription_id}/resourceGroups/{resource_group}/providers/Microsoft.Web/connections/{resource_name}?api-version=2016-06-01"
headers = {'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'}
response = requests.put(endpoint, headers=headers, json=payload)
if response.status_code != 200:
print(f"Failed to copy '{resource_name}': {response.status_code} - {response.text}")
return response.status_code
# Processing each API connection
for resource in resources:
details = get_api_connection_details(subscription_id, resource_group_name_source, resource.name)
if 'properties' in details and 'location' in details:
properties = details['properties']
location = details['location']
# Fetch parameters securely, e.g., from Azure Key Vault or configured manually here
parameter_values = {"connectionString": "your_actual_connection_string_here"} # Adjust this as necessary
status_code = create_or_update_api_connection(subscription_id, resource_group_name_target, resource.name, properties, location, parameter_values)
if status_code == 200:
print(f"API Connection '{resource.name}' copied to {resource_group_name_target}")
else:
print(f"Failed to copy '{resource.name}' with status code {status_code}")
else:
print(f"Skipping '{resource.name}': Incomplete data to copy.")
print("Process completed.")
部署成功: