azure-storage-file-share-py

Azure Storage File Share Python SDK

管理适用于云原生和直接迁移场景的 SMB 文件共享。

安装

pip install azure-storage-file-share

环境变量

AZURE_STORAGE_CONNECTION_STRING=DefaultEndpointsProtocol=https;AccountName=...;AccountKey=...
# Or
AZURE_STORAGE_ACCOUNT_URL=https://<account>.file.core.windows.net

身份验证

连接字符串

import os
from azure.storage.fileshare import ShareServiceClient

service = ShareServiceClient.from_connection_string(
    os.environ["AZURE_STORAGE_CONNECTION_STRING"]
)

Entra ID

from azure.storage.fileshare import ShareServiceClient
from azure.identity import DefaultAzureCredential

service = ShareServiceClient(
    account_url=os.environ["AZURE_STORAGE_ACCOUNT_URL"],
    credential=DefaultAzureCredential(),
    token_intent="backup",
)

对于 OAuth，请参阅 token_intent 和备份权限；备份意图可以绕过文件/目录 ACL 检查，并且需要明确授权的角色。各示例均为独立片段，需要 account_url、os 和兼容的包。写入/上传可能会替换内容；请确认共享/路径准确无误，保护现有文件，限制下载字节数，并且只删除可丢弃的资源。切勿打印连接字符串或 SAS URL。复制启动是异步的：在报告成功前，轮询目标复制状态。发生故障时，请使用 try/finally 或异步上下文管理器关闭客户端/凭据。

共享操作

创建共享

share = service.create_share("my-share")

列出共享

for share in service.list_shares():
    print(f"{share.name}: {share.quota} GB")

获取共享客户端

share_client = service.get_share_client("my-share")

删除共享

service.delete_share("my-share")

目录操作

创建目录

share_client = service.get_share_client("my-share")
share_client.create_directory("my-directory")

# Nested directory
share_client.create_directory("my-directory/sub-directory")

列出目录和文件

directory_client = share_client.get_directory_client("my-directory")

for item in directory_client.list_directories_and_files():
    if item["is_directory"]:
        print(f"[DIR] {item['name']}")
    else:
        print(f"[FILE] {item['name']} ({item['size']} bytes)")

删除目录

share_client.delete_directory("my-directory")

文件操作

上传文件

file_client = share_client.get_file_client("my-directory/file.txt")

# From string
file_client.upload_file("Hello, World!")

# From file
with open("local-file.txt", "rb") as f:
    file_client.upload_file(f)

# From bytes
file_client.upload_file(b"Binary content")

下载文件

file_client = share_client.get_file_client("my-directory/file.txt")

# To bytes
data = file_client.download_file().readall()

# To file
with open("downloaded.txt", "wb") as f:
    data = file_client.download_file()
    data.readinto(f)

# Stream chunks
download = file_client.download_file()
for chunk in download.chunks():
    process(chunk)

获取文件属性

properties = file_client.get_file_properties()
print(f"Size: {properties.size}")
print(f"Content type: {properties.content_settings.content_type}")
print(f"Last modified: {properties.last_modified}")

删除文件

file_client.delete_file()

复制文件

source_url = "https://account.file.core.windows.net/share/source.txt"
dest_client = share_client.get_file_client("destination.txt")
dest_client.start_copy_from_url(source_url)

范围操作

上传范围

# Upload to specific range
file_client.upload_range(data=b"content", offset=0, length=7)

下载范围

# Download specific range
download = file_client.download_file(offset=0, length=100)
data = download.readall()

快照操作

创建快照

snapshot = share_client.create_snapshot()
print(f"Snapshot: {snapshot['snapshot']}")

访问快照

snapshot_client = service.get_share_client(
    "my-share",
    snapshot=snapshot["snapshot"]
)

异步客户端

from azure.storage.fileshare.aio import ShareServiceClient
from azure.identity.aio import DefaultAzureCredential

async def upload_file():
    credential = DefaultAzureCredential()
    service = ShareServiceClient(account_url, credential=credential, token_intent="backup")
    
    share = service.get_share_client("my-share")
    file_client = share.get_file_client("test.txt")
    
    await file_client.upload_file("Hello!")
    
    await service.close()
    await credential.close()

客户端类型

客户端	用途
ShareServiceClient	账户级操作
ShareClient	共享操作
ShareDirectoryClient	目录操作
ShareFileClient	文件操作

最佳实践

1. 尽可能使用范围受限的身份；连接字符串包含账户凭据
2. 在生产环境中使用 Entra ID 并配置 RBAC
3. 使用 chunks() 流式处理大文件，以避免内存问题
4. 在重大更改前创建快照
5. 设置配额，以避免意外的存储费用
6. 使用范围进行部分文件更新
7. 显式关闭异步客户端

适用场景

当任务涉及执行概述中描述的工作流或操作时，可使用此技能。

限制

• 仅当任务明确符合上述范围时，才使用此技能。
• 不要将输出视为环境特定验证、测试或专家审查的替代品。
• 如果缺少必要的输入、权限、安全边界或成功标准，请停止并请求澄清。