【免费下载链接】context-hub
本指南基于 Context Hub 仓库中的 Azure Blob Storage Python SDK 文档(content/azure/docs/storage-blob/python/DOC.md),面向需要对接 Azure 对象存储的 Python 开发者,系统讲解azure-storage-blob12.28.0 的安装、四种认证方式、三层客户端模型,以及上传、下载、列出的完整可运行示例,并深入剖析 12.28.0 版本特有的行为差异(Azurite 连接串简写、start_from、decompress、默认块大小变更等)。阅读完本文,你将能独立完成从认证配置到生产级传输调优的 Azure Blob 集成。
版本前提与 Golden Rule
本文档对应的 SDK 版本为azure-storage-blob==12.28.0。使用该版本时有一个必须遵守的「黄金法则」:Blob 操作一律使用azure-storage-blob,并在真实 Azure 环境中配合azure-identity的DefaultAzureCredential进行认证。依据包概览与 PyPI 元数据,12.28.0 要求Python >= 3.9(注意:部分 Microsoft Learn 旧文章仍写 Python 3.8+,属于过期信息,应以包元数据为准)。
该文档属于 Context Hub 的维护者(source: maintainer)来源内容,语言标记为 python、版本 12.28.0、修订号 1,更新于 2026-03-12,其 frontmatter 格式遵循 内容指南 中定义的 DOC.md 规范。
安装
固定版本安装(推荐,避免行为漂移):
python -m pip install "azure-storage-blob==12.28.0" azure-identity如需异步用法,额外安装异步传输层(如aiohttp):
python -m pip install "azure-storage-blob==12.28.0" azure-identity aiohttp常用环境变量
export AZURE_STORAGE_ACCOUNT_URL="https://<account>.blob.core.windows.net" export AZURE_STORAGE_CONTAINER="documents"当你在本地开发 Azurite 或使用连接字符串认证时,用AZURE_STORAGE_CONTAINER之外还要注意:连接字符串认证场景下,应使用AZURE_STORAGE_CONNECTION_STRING替代AZURE_STORAGE_ACCOUNT_URL。
认证与初始化
首选:Microsoft Entra ID +DefaultAzureCredential
这是官方推荐的 Azure 托管应用与常规本地开发路径。本地配置三步走:
- 安装
azure-identity; - 通过
az login或其他受支持的开发者凭据源登录; - 为主体验证主体授予存储账户或容器上的 Azure RBAC 权限。
角色指引:
- 只读流程(列出、下载等)需要
Storage Blob Data Reader或更高权限; - 读写流程(上传、创建、删除等)需要
Storage Blob Data Contributor或更高权限。
RBAC 分配可能需要几分钟传播。如果新建的角色分配立即失败,稍作等待重试即可,无需重写认证流程。
import os from azure.identity import DefaultAzureCredential from azure.storage.blob import BlobServiceClient account_url = os.environ["AZURE_STORAGE_ACCOUNT_URL"] credential = DefaultAzureCredential() service = BlobServiceClient(account_url=account_url, credential=credential)连接字符串或 Azurite
连接字符串在本地存储模拟器、迁移工具以及已持有存储密钥的环境中仍然实用:
import os from azure.storage.blob import BlobServiceClient service = BlobServiceClient.from_connection_string( os.environ["AZURE_STORAGE_CONNECTION_STRING"] )12.28.0 新增的 Azurite 简写:from_connection_string()现在直接接受"UseDevelopmentStorage=true;":
service = BlobServiceClient.from_connection_string("UseDevelopmentStorage=true;")SAS 令牌或账户密钥
SDK 同样支持 SAS 令牌与账户密钥认证。在确有需要时使用它们,但应用代码应优先选择 Entra ID,以避免在代码中嵌入长期有效的存储密钥。
客户端模型:三层结构
SDK 围绕三个客户端层组织:
BlobServiceClient:账户级作用域;ContainerClient:单个容器;BlobClient:单个 Blob。
典型获取流程:
container = service.get_container_client("documents") blob = container.get_blob_client("reports/hello.txt")注意:SDK 不会自动创建缺失的容器,在调用上传或下载前必须先创建容器。这一设计在 registry.js 中对应的是文档路径按entry → language → version → path逐层解析的思路——客户端层级同样需要逐层获取,任何一层缺失都会导致后续操作失败。
核心用法
基础同步流程(完整可运行示例)
import os from azure.core.exceptions import ResourceExistsError from azure.identity import DefaultAzureCredential from azure.storage.blob import BlobServiceClient account_url = os.environ["AZURE_STORAGE_ACCOUNT_URL"] container_name = os.getenv("AZURE_STORAGE_CONTAINER", "documents") service = BlobServiceClient(account_url=account_url, credential=DefaultAzureCredential()) container = service.get_container_client(container_name) try: container.create_container() except ResourceExistsError: pass blob = container.get_blob_client("reports/hello.txt") blob.upload_blob(b"hello from azure-storage-blob\n", overwrite=True) text = blob.download_blob(encoding="UTF-8").readall() print(text) for item in container.list_blobs(name_starts_with="reports/"): print(item.name, item.size)上传
upload_blob()会根据对象大小与传输设置自动选择单请求上传或基于块的(block-based)上传:
from azure.storage.blob import ContentSettings with open("report.json", "rb") as data: blob.upload_blob( data, overwrite=True, tags={"kind": "report"}, content_settings=ContentSettings(content_type="application/json"), )实用上传要点:
overwrite=True通常是幂等应用代码的必需项;standard_blob_tier=可为块 Blob 设置Hot、Cool、Cold或Archive访问层;- 手动分块暂存请使用
stage_block()配合commit_block_list()。
下载
download_blob()返回StorageStreamDownloader。小负载用readall(),写文件用readinto(),流式处理用chunks():
download_stream = blob.download_blob(max_concurrency=4) with open("report.bin", "wb") as fh: download_stream.readinto(fh)分块处理:
stream = blob.download_blob() for chunk in stream.chunks(): process(chunk)列出
扁平列出:
for item in container.list_blobs(name_starts_with="reports/2026/"): print(item.name)带虚拟目录的层级列出:
from azure.storage.blob import BlobPrefix for item in container.walk_blobs(name_starts_with="reports/", delimiter="/"): if isinstance(item, BlobPrefix): print("dir", item.name) else: print("blob", item.name)需要理解的关键事实:Blob Storage 本质上是扁平的,类文件夹行为来自 Blob 命名加分隔符(delimiter)。另外,Blob 快照(snapshots)无法通过层级列出方式列出。列出过滤务必使用name_starts_with=关键字——这是当前文档与示例使用的标准关键字。
异步模式
使用azure.storage.blob.aio下的异步客户端,且凭据与客户端都需要显式关闭。顶层服务客户端通常用async with管理,子客户端共享其连接池:
import os from azure.identity.aio import DefaultAzureCredential from azure.storage.blob.aio import BlobServiceClient async def main() -> None: account_url = os.environ["AZURE_STORAGE_ACCOUNT_URL"] async with DefaultAzureCredential() as credential: async with BlobServiceClient(account_url, credential=credential) as service: container = service.get_container_client("documents") async for item in container.list_blobs(name_starts_with="reports/"): print(item.name)生产环境配置要点
以下设置在官方概览与任务指南中被重点提及,值得在生产中关注:
| 配置类别 | 相关参数 | 作用 |
|---|---|---|
| 重试策略 | retry_total、retry_connect、retry_read、retry_status、retry_to_secondary | 控制不同失败场景下的重试次数与是否故障转移至副本 |
| 超时 | connection_timeout、read_timeout | 控制建连与读超时 |
| 日志 | logging_enable=True | 输出请求诊断信息 |
| 上传调优 | max_block_size、max_single_put_size、per-callmax_concurrency | 决定分块大小与并发 |
| 下载调优 | max_chunk_get_size、max_single_get_size、per-callmax_concurrency | 决定获取块大小与并发 |
默认值是合理起点:除非你针对特定工作负载有证据表明需要调整传输或重试行为,否则先使用默认配置。
常见陷阱清单
- 部分 Microsoft Learn 任务文章仍标注 Python
3.8+,但 12.28.0 的包概览与 PyPI 元数据要求Python >= 3.9,请以包元数据为准; - 较老的概览片段仍提到 Python
3.5+的异步支持,该描述对当前稳定版已过时; upload_blob()不保护并发写入者,微软明确文档化客户端库不支持对同一 Blob 的并发写入;- 容器必须存在,否则上传/下载示例会失败;
- 列出过滤使用
name_starts_with=关键字; DefaultAzureCredential失败通常是环境问题:未登录、租户错误或 RBAC 传播延迟;- 调试日志可能包含请求与响应体,生产环境不要随意开启。
12.28.0 版本敏感行为
12.28.0是12.28.0b1功能集的稳定发布版;ContainerClient.list_blobs()、list_blob_names()、walk_blobs()在 12.28.0 功能线中新增了start_from关键字;BlobClient.download_blob()在 12.28.0 功能线中新增了decompress关键字;BlobServiceClient.from_connection_string()现在接受UseDevelopmentStorage=true;用于 Azurite;- 默认
connection_data_block_size从4 KiB 改为 256 KiB,官方 changelog 指出这应能提升许多大文件下载的吞吐。
在 Context Hub 中获取与使用本文档
本文档以 DOC.md 形式存放于仓库的 content/azure/docs/storage-blob/python/ 目录,其 frontmatter(name: storage-blob、languages: python、versions: 12.28.0)符合 内容指南 中 DOC.md 的字段规范——name构成条目 IDazure/storage-blob,versions记录的是 PyPI 包版本,revision与updated-on提供内容新鲜度信号。
当你的编码 Agent 需要最新、准确的 Blob 存储 API 信息时,可以按 get-api-docs 技能 的指引,通过 Context Hub CLI 获取(而不是依赖可能过时的训练数据):
chub search "azure blob storage" --json # 找到最佳匹配 id,例如 azure/storage-blob chub get azure/storage-blob --lang py # 获取本文档对应语言的 DOC.md获取后可先用--lang指定语言、--version 12.28.0锁定版本(若未指定版本,CLI 默认取recommendedVersion,见 registry.js 的版本回退逻辑)。使用过程中若发现文档未覆盖的坑点,可用chub annotate保存本地笔记;对文档质量有意见可用chub feedback反馈,完整命令参考见 CLI 参考文档。
官方资料
本文档内容基于以下官方来源整理(仓库内文档引用的外部参考资料):Microsoft Learn 包概览、API 根、快速入门、上传/下载/列出手册,PyPI 包页面,以及 Azure SDK changelog。编写代码时若遇到版本相关疑问,优先核对 changelog 与包元数据。
【免费下载链接】context-hub
相关推荐
kOps 中 Azure Blob Storage Go 客户端模块(azblob)完全指南:认证、上传下载与源码级实战
kOps 中 Azure Blob Storage Go 客户端模块(azblob)完全指南:认证、上传下载与源码级实战 导读 本篇技术指南以 kOps 仓库所
云原生集群管理运维IaCContext Hub 技术文档精读:Azure Blob Storage JavaScript 客户端(@azure/storage-blob 12.31.0)认证、上传、下载与列表演练
Context Hub 技术文档精读:Azure Blob Storage JavaScript 客户端(@azure/storage blob 12.31.0
在 distribution 中驾驭 Azure Blob Storage:Go azblob SDK 的认证、客户端模型与实战
在 distribution 中驾驭 Azure Blob Storage:Go azblob SDK 的认证、客户端模型与实战 本文以当前仓库 vendor
镜像仓库后端存储云原生
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考