使用Python创建内存文本文件并上传至Azure Blob Storage的实现问题
Python内存文本文件上传Azure Blob Storage实现方案
常见失败原因
绝大多数使用io模块实现上传失败的原因集中在以下几点:
- 内存流写入内容后未将读取指针重置到起始位置,导致SDK读取到空内容,最终上传0字节文件
- 混淆字符流(
io.StringIO)和字节流(io.BytesIO),传入SDK不支持的流类型触发类型报错 - 字符串编码未统一处理,出现乱码或上传过程抛出编码异常
- 多字符串输入未做统一格式处理,导致内容缺失
前置依赖
先安装Azure Blob官方SDK:
pip install azure-storage-blob
完整可运行代码
代码支持传入单个字符串或字符串列表,自动在内存中生成文本流后直接上传,无需生成本地临时文件:
import io from typing import Union, List from azure.storage.blob import BlobServiceClient, ContentSettings def upload_text_to_blob( content: Union[str, List[str]], az_connection_str: str, container_name: str, blob_path: str, file_encoding: str = "utf-8", content_sep: str = "\n", overwrite_existing: bool = True ) -> None: # 统一处理单字符串、多字符串两种入参格式 content_batch = [content] if isinstance(content, str) else content # 拼接全量内容并按指定编码转字节 raw_text = content_sep.join(content_batch) text_bytes = raw_text.encode(file_encoding) # 初始化内存字节流,写入内容后重置指针到起始位 memory_stream = io.BytesIO(text_bytes) memory_stream.seek(0) # 这步是核心,漏掉必传空文件 # 初始化Blob客户端完成上传 service_client = BlobServiceClient.from_connection_string(az_connection_str) blob_client = service_client.get_blob_client(container=container_name, blob=blob_path) blob_client.upload_blob( memory_stream, overwrite=overwrite_existing, content_settings=ContentSettings( content_type=f"text/plain; charset={file_encoding}" ) )
使用示例
# 单个字符串上传 upload_text_to_blob( content="这是单段测试文本", az_connection_str="你的Azure存储连接字符串", container_name="你的容器名", blob_path="test/single.txt" ) # 多字符串上传 upload_text_to_blob( content=["第一行内容", "第二行内容", "第三行内容"], az_connection_str="你的Azure存储连接字符串", container_name="你的容器名", blob_path="test/multi.txt" )
注意事项
- 不要直接传入
io.StringIO对象给upload_blob方法:该方法原生支持字节流、字节对象、本地文件路径,StringIO属于字符流,直接传会触发参数类型错误 - 如果业务场景需要用StringIO处理文本逻辑,可在上传前调用
string_io_obj.getvalue().encode(encoding)转成字节,再传入BytesIO或者直接作为上传参数 - 按需调整
content_sep参数:如果不需要多段字符串之间自动加换行,将该参数设为空字符串即可 - 显式配置
ContentSettings可以避免上传的文件被识别为二进制流,后续公网访问时可以直接在浏览器预览文本内容
内容的提问来源于stack exchange,提问作者Ahmed
相关产品推荐
相关产品推荐

