Azure Blob Storage SDK for Python

用于Azure Blob存储（非结构化数据的对象存储）的客户端库。

安装

bash
pip install azure-storage-blob azure-identity

环境变量

bash
AZURESTORAGEACCOUNT_NAME=<你的存储账户名称>

或使用完整URL

AZURESTORAGEACCOUNT_URL=https://<账户>.blob.core.windows.net

身份验证

python
from azure.identity import DefaultAzureCredential
from azure.storage.blob import BlobServiceClient

credential = DefaultAzureCredential()
account_url = https://<账户>.blob.core.windows.net

blobserviceclient = BlobServiceClient(account_url, credential=credential)

客户端层次结构

客户端	用途	获取方式
BlobServiceClient	账户级操作	直接实例化
ContainerClient

容器操作 | blobserviceclient.getcontainerclient() | | BlobClient | 单个Blob操作 | containerclient.getblob_client() |

核心工作流

创建容器

python
containerclient = blobserviceclient.getcontainer_client(mycontainer)
containerclient.createcontainer()

上传Blob

python

从文件路径上传

blobclient = blobserviceclient.getblob_client(
container=mycontainer,
blob=sample.txt
)

with open(./local-file.txt, rb) as data:
blobclient.uploadblob(data, overwrite=True)

从字节/字符串上传

blobclient.uploadblob(bHello, World!, overwrite=True)

从流上传

import io stream = io.BytesIO(bStream content) blobclient.uploadblob(stream, overwrite=True)

下载Blob

python
blobclient = blobserviceclient.getblob_client(
container=mycontainer,
blob=sample.txt
)

下载到文件

with open(./downloaded.txt, wb) as file: downloadstream = blobclient.download_blob() file.write(download_stream.readall())

下载到内存

downloadstream = blobclient.download_blob() content = download_stream.readall() # 字节数据

读取到现有缓冲区

stream = io.BytesIO() numbytes = blobclient.download_blob().readinto(stream)

列出Blob

python
containerclient = blobserviceclient.getcontainer_client(mycontainer)

列出所有Blob

for blob in containerclient.listblobs(): print(f{blob.name} - {blob.size} 字节)

按前缀列出（类似文件夹）

for blob in containerclient.listblobs(namestartswith=logs/): print(blob.name)

遍历Blob层次结构（虚拟目录）

for item in containerclient.walkblobs(delimiter=/): if item.get(prefix): print(f目录: {item[prefix]}) else: print(fBlob: {item.name})

删除Blob

python
blobclient.deleteblob()

包含快照的删除

blobclient.deleteblob(delete_snapshots=include)

性能调优

python

配置大文件上传/下载的块大小

blob_client = BlobClient(
accounturl=accounturl,
container_name=mycontainer,
blob_name=large-file.zip,
credential=credential,
maxblocksize=4 1024 1024, # 4 MiB块
maxsingleput_size=64 1024 1024 # 64 MiB单次上传限制
)

并行上传

blobclient.uploadblob(data, max_concurrency=4)

并行下载

downloadstream = blobclient.downloadblob(maxconcurrency=4)

SAS令牌

python
from datetime import datetime, timedelta, timezone
from azure.storage.blob import generateblobsas, BlobSasPermissions

sastoken = generateblob_sas(
account_name=<账户>,
container_name=mycontainer,
blob_name=sample.txt,
account_key=<账户密钥>, # 或使用用户委托密钥
permission=BlobSasPermissions(read=True),
expiry=datetime.now(timezone.utc) + timedelta(hours=1)
)

使用SAS令牌

bloburl = fhttps://<账户>.blob.core.windows.net/mycontainer/sample.txt?{sastoken}

Blob属性和元数据

python

获取属性

properties = blobclient.getblob_properties()
print(f大小: {properties.size})
print(f内容类型: {properties.contentsettings.contenttype})
print(f最后修改时间: {properties.last_modified})

设置元数据

blobclient.setblob_metadata(metadata={category: logs, year: 2024})

设置内容类型

from azure.storage.blob import ContentSettings blobclient.sethttp_headers( contentsettings=ContentSettings(contenttype=application/json) )

异步客户端

python
from azure.identity.aio import DefaultAzureCredential
from azure.storage.blob.aio import BlobServiceClient

async def upload_async():
credential = DefaultAzureCredential()

async with BlobServiceClient(account_url, credential=credential) as client:
blobclient = client.getblob_client(mycontainer, sample.txt)

with open(./file.txt, rb) as data:
await blobclient.uploadblob(data, overwrite=True)

异步下载

async def download_async(): async with BlobServiceClient(account_url, credential=credential) as client: blobclient = client.getblob_client(mycontainer, sample.txt)

stream = await blobclient.downloadblob()
data = await stream.readall()

最佳实践

1. 1. 使用DefaultAzureCredential 替代连接字符串
2. 使用上下文管理器 管理异步客户端
3. 明确设置overwrite=True 在重新上传时
4. 使用maxconcurrency 处理大文件传输
5. 优先使用readinto() 而非readall()以提高内存效率
6. 使用walkblobs() 进行层次结构列表
7. 设置适当的内容类型 用于Web服务的Blob