如何使用Python SDK或Azure CLI实现本地文件夹与Azure存储账户同步？

算力阿峰

2026-5-27

当然有啦！Azure提供了很方便的方式来实现本地文件夹和存储账户（具体是Blob存储容器）之间的同步，不管是用CLI还是Python SDK都能搞定，下面给你详细拆解：

使用Azure CLI实现同步

Azure CLI直接提供了az storage blob sync命令，专门用来做本地文件夹和Azure Blob存储容器之间的双向同步，和AWS CLI的sync功能逻辑几乎一致。

举几个常用的场景：

把本地./my-local-folder同步到名为my-container的存储容器：

az storage blob sync --source ./my-local-folder --container-name my-container --account-name my-storage-account

如果要同步时删除目标容器中本地没有的文件（和AWS的--delete参数效果一样），加上--delete选项：
```
az storage blob sync --source ./my-local-folder --container-name my-container --account-name my-storage-account --delete
```

要是想先预览同步操作会做什么，不实际执行，用--dry-run参数：

az storage blob sync --source ./my-local-folder --container-name my-container --account-name my-storage-account --dry-run

注意：执行命令前要确保已经用az login登录了Azure账号，或者通过--account-key、--sas-token提供存储账户的访问权限。

使用Python SDK实现同步

如果需要自定义同步逻辑（比如更复杂的文件过滤规则、自定义对比逻辑），可以用Azure Blob Storage的Python SDK（azure-storage-blob）来实现。

步骤大概是这样：

先安装SDK：
```
pip install azure-storage-blob
```
编写同步代码，下面是一个简单的本地文件夹同步到Blob容器的示例，包含上传新文件/更新修改过的文件，以及可选删除容器中不存在的文件：

from azure.storage.blob import BlobServiceClient
import os
from datetime import datetime

def sync_local_to_azure(local_folder: str, connection_string: str, container_name: str, delete_extra: bool = False):
    # 初始化Blob服务客户端
    blob_service_client = BlobServiceClient.from_connection_string(connection_string)
    container_client = blob_service_client.get_container_client(container_name)

    # 获取容器中所有Blob的名称和最后修改时间
    blob_list = container_client.list_blobs()
    blob_dict = {
        blob.name: blob.last_modified.replace(tzinfo=None)
        for blob in blob_list
    }

    # 遍历本地文件夹
    for root, dirs, files in os.walk(local_folder):
        for file in files:
            local_file_path = os.path.join(root, file)
            # 计算Blob的相对路径（模拟虚拟目录结构）
            blob_name = os.path.relpath(local_file_path, local_folder).replace("\\", "/")
            
            # 获取本地文件的最后修改时间
            local_modified_time = datetime.fromtimestamp(os.path.getmtime(local_file_path))
            
            # 如果Blob不存在，或者本地文件更新时间更晚，就上传
            if blob_name not in blob_dict or local_modified_time > blob_dict[blob_name]:
                print(f"Uploading {local_file_path} to {blob_name}")
                with open(local_file_path, "rb") as data:
                    container_client.upload_blob(name=blob_name, data=data, overwrite=True)
            # 从Blob字典中移除已处理的项，剩下的就是容器中多余的
            if blob_name in blob_dict:
                del blob_dict[blob_name]
    
    # 如果开启了删除多余Blob的选项，删除剩下的项
    if delete_extra:
        for extra_blob in blob_dict.keys():
            print(f"Deleting extra blob: {extra_blob}")
            container_client.delete_blob(extra_blob)

# 使用示例
if __name__ == "__main__":
    CONNECTION_STRING = "your-storage-account-connection-string"
    LOCAL_FOLDER = "./my-local-folder"
    CONTAINER_NAME = "my-container"
    sync_local_to_azure(LOCAL_FOLDER, CONNECTION_STRING, CONTAINER_NAME, delete_extra=True)

这个示例会对比本地文件和Blob的最后修改时间，只上传需要更新的文件；如果开启delete_extra，还会删除容器中本地已经不存在的Blob。你可以根据自己的需求扩展逻辑，比如添加文件后缀过滤、自定义对比规则等。

额外提醒：不管用CLI还是SDK，都需要确保有对应的存储账户权限（比如Blob数据Contributor角色，或者使用存储账户密钥/SAS token）。另外，Azure Blob存储中的“文件夹”是虚拟的，通过Blob名称中的/来模拟，CLI和SDK都会自动处理这种结构，保持本地和容器的目录一致。

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