Python上传文件至Azure Blob存储时如何自动设置正确Content-Type
解决方法
直接使用Python标准库内置的mimetypes模块即可自动生成符合MIME规范的Content-Type,无需安装第三方依赖,匹配逻辑和azcopy的后缀识别规则一致,完全满足批量上传多类型文件的需求。
核心逻辑
调用mimetypes.guess_type()传入本地文件路径即可自动匹配对应MIME值:
- 常规后缀均可正确识别:比如
.html返回text/html、.js返回application/javascript、.png返回image/png - 无法识别后缀类型时,兜底使用Azure Blob默认的
application/octet-stream即可
修正后可直接运行的代码
替换你原有逻辑中Content-Type生成部分即可,同时修正了你原代码里的两处语法/逻辑错误:path_generator()循环行末尾缺失的冒号、等待上传任务完成的循环缩进错误(原缩进会导致每提交一个任务就阻塞等待,无法发挥线程池并发优势)
import mimetypes from typing import Dict, Any from multiprocessing.pool import ThreadPool from azure.storage.blob import BlobClient, ContentSettings # 初始化加载内置MIME映射表,全局执行一次即可 mimetypes.init() def upload(self, overwrite=False): with ThreadPool(self._num_threads) as pool: request_processes = [] for local_path, remote_path in path_generator(): blob_client = BlobClient.from_blob_url(blob_url=remote_path, credential=credentials) # 自动匹配Content-Type content_type, _ = mimetypes.guess_type(local_path) if not content_type: content_type = "application/octet-stream" content_settings = ContentSettings(content_type=content_type) kwargs: Dict[str, Any] = { 'blob_type': 'BlockBlob', 'content_settings': content_settings, 'validate_content': True, 'overwrite': overwrite, 'timeout': self._timeout_seconds, 'max_concurrency': 10 } request_processes.append( pool.apply_async( self._sync_upload_file, [blob_client, local_path], kwds=kwargs ) ) # 所有任务提交完成后统一等待结果 for req in request_processes: req.get(self._timeout_seconds) @staticmethod def _sync_upload_file(blob_client, local_path: str, *args, **kwargs): with open(local_path, "rb") as data: resp = blob_client.upload_blob(data, *args, **kwargs) return resp
可选优化
如果存在部分冷门文件后缀识别不准的情况,可以在初始化阶段调用mimetypes.add_type()手动补充映射规则,示例:
# 补充自定义后缀对应的MIME类型 mimetypes.add_type('image/webp', '.webp') mimetypes.add_type('application/wasm', '.wasm')
该方案仅通过文件后缀匹配类型,不需要读取文件内容做特征检测,性能开销极低,适合数千甚至数万级文件批量上传的场景。
内容的提问来源于stack exchange,提问作者Hesam Eskandari
相关产品推荐
相关产品推荐

