Azure存储账户批量备份脚本动态配置优化方案咨询

最优配置方案选择 #

直接用字典存储源-目标存储的映射关系，比双列表方案更可靠：双列表如果出现顺序错位会直接导致数据备份到错误账户，字典的键值对绑定可以完全避免这个问题，同时支持灵活配置非规则命名的存储映射。

实现步骤 #

1. 配置与业务代码分离 #

把所有存储账户的映射关系放到单独的配置文件中，避免修改业务逻辑代码，同时你用Bicep部署的目标存储可以直接输出该配置文件，无需手动复制连接串。
示例配置文件config.json格式：

{
    "backup_mappings": [
        {
            "source": {
                "account_name": "源存储1名称",
                "connection_string": "源存储1连接串",
                "account_key": "源存储1密钥"
            },
            "target": {
                "account_name": "备份存储1名称",
                "connection_string": "备份存储1连接串",
                "account_key": "备份存储1密钥"
            }
        },
        {
            "source": {
                "account_name": "源存储2名称",
                "connection_string": "源存储2连接串",
                "account_key": "源存储2密钥"
            },
            "target": {
                "account_name": "备份存储2名称",
                "connection_string": "备份存储2连接串",
                "account_key": "备份存储2密钥"
            }
        }
    ]
}

2. 原有逻辑封装为可复用函数 #

把你原本单账户的备份逻辑封装为接收源、目标配置参数的函数，循环遍历映射列表调用即可。
优化后的完整代码示例：

from typing import Container, Dict
import json
from azure.cosmosdb.table.tableservice import TableService,ListGenerator
from azure.storage.blob import BlobClient, BlobServiceClient, ContainerClient
from azure.storage.blob import ResourceTypes, AccountSasPermissions
from azure.storage.blob import generate_account_sas    
from datetime import date, datetime, timedelta

# 全局常量
QUERY_SIZE = 100
TODAY = str(date.today().strftime("%Y%m%d"))

# 加载配置
def load_config(config_path: str = "config.json") -> Dict:
    with open(config_path, "r", encoding="utf-8") as f:
        return json.load(f)

# Table备份逻辑
def query_and_save_all_data(tb_name: str, resp_data: ListGenerator, table_out: TableService, table_in: TableService):
    for item in resp_data:
        del item.etag
        del item.Timestamp
        table_in.insert_or_replace_entity(tb_name, item)
    if resp_data.next_marker:
        data = table_out.query_entities(table_name=tb_name, num_results=QUERY_SIZE, marker=resp_data.next_marker)
        query_and_save_all_data(tb_name, data, table_out, table_in)

# 单组存储账户的备份逻辑
def backup_single_storage_pair(source_config: Dict, target_config: Dict):
    # Blob备份
    source_blob_client = BlobServiceClient.from_connection_string(source_config["connection_string"])
    target_blob_client = BlobServiceClient.from_connection_string(target_config["connection_string"])
    # 生成源账户SAS
    sas_token = generate_account_sas(
        account_name = source_blob_client.account_name,
        account_key = source_config["account_key"], 
        resource_types = ResourceTypes(object=True, container=True),
        permission= AccountSasPermissions(read=True,list=True),
        expiry = datetime.utcnow() + timedelta(hours=4)
    )
    # 遍历Blob容器
    all_containers = source_blob_client.list_containers(include_metadata=True)
    for container in all_containers:
        print(f"处理Blob容器: {container['name']}")
        container_client = source_blob_client.get_container_client(container.name)
        blobs_list = container_client.list_blobs()
        target_container_name = f"{container['name']}{TODAY}"
        # 提前创建目标容器，避免循环内重复创建
        target_blob_client.create_container(target_container_name, exist_ok=True)
        for blob in blobs_list:
            source_blob = BlobClient(
                source_blob_client.url,
                container_name = container['name'],
                blob_name = blob.name,
                credential = sas_token
            )
            target_blob = target_blob_client.get_blob_client(target_container_name, blob.name)
            target_blob.start_copy_from_url(source_blob.url)
            print(f"备份Blob: {blob.name} 到 {target_container_name}")
    
    # Table备份
    table_service_out = TableService(account_name=source_config["account_name"], account_key=source_config["account_key"])
    table_service_in = TableService(account_name=target_config["account_name"], account_key=target_config["account_key"])
    tbs_out = table_service_out.list_tables()
    for tb in tbs_out:
        target_table_name = f"{tb.name}{TODAY}"
        table_service_in.create_table(table_name=target_table_name, fail_on_exist=False)
        data = table_service_out.query_entities(tb.name, num_results=QUERY_SIZE)
        query_and_save_all_data(target_table_name, data, table_service_out, table_service_in)
        print(f"备份Table: {tb.name} 到 {target_table_name}")

if __name__ == "__main__":
    config = load_config()
    for idx, mapping in enumerate(config["backup_mappings"]):
        print(f"开始处理第{idx+1}组存储账户: 源={mapping['source']['account_name']} 目标={mapping['target']['account_name']}")
        backup_single_storage_pair(mapping["source"], mapping["target"])
        print(f"第{idx+1}组存储账户处理完成")

额外优化建议 #

• 配置中的密钥不要明文存储，可以集成Azure Key Vault拉取密钥，避免配置文件泄露
• 可以增加拷贝状态校验逻辑，避免异步拷贝失败没有感知
• 原代码中的空except逻辑已经优化为使用exist_ok=True提前创建容器，避免不必要的异常捕获
• 如果你的源存储和目标存储命名有固定规则（比如源叫xxx，备份叫xxx-backup），可以直接通过Azure Management SDK拉取订阅下所有存储账户自动生成映射关系，完全不需要手动维护配置文件

内容的提问来源于stack exchange，提问作者Nayden Van