使用Python+SAS_TOKEN上传Azure Blob遇异步错误求可行方案

你的错误源于误用**异步版本Azure Blob SDK（azure.storage.blob.aio）**在同步代码环境中。异步方法必须在异步函数内通过await调用，若直接在同步逻辑中执行异步方法，会返回未执行的协程对象或None，进而触发TypeError: object NoneType can't be used in 'await' expression。

以下是几种无需Azure登录、仅用存储账户URL+SAS_TOKEN的可行上传方案：

方案1：同步版Azure Blob SDK上传 #

使用同步版azure.storage.blob包，无需异步环境，适配你的场景：

from azure.storage.blob import BlobClient

# 替换为你的实际信息
storage_account_url = "https://<你的存储账户名>.blob.core.windows.net"
sas_token = "<你的SAS_TOKEN>"
container_name = "<目标容器名>"
blob_name = "<要创建的Blob名称>"
local_file_path = "<本地文件路径>"

# 初始化Blob客户端
blob_client = BlobClient(
    account_url=storage_account_url,
    container_name=container_name,
    blob_name=blob_name,
    credential=sas_token
)

# 上传文件（overwrite=True表示覆盖同名Blob）
with open(local_file_path, "rb") as data:
    blob_client.upload_blob(data, overwrite=True)

print("文件上传完成")

方案2：curl命令上传 #

无需依赖SDK，适合快速测试或脚本场景：

# 替换占位符为实际信息
curl -X PUT \
  "https://<存储账户名>.blob.core.windows.net/<容器名>/<Blob名>?<SAS_TOKEN>" \
  --data-binary "@<本地文件路径>" \
  -H "Content-Type: <文件MIME类型，如application/octet-stream、image/jpeg>"

注意：如果SAS_TOKEN开头带有?，需删除URL中额外的?，确保最终URL格式为https://xxx.blob.core.windows.net/container/blob?sv=xxx&ss=xxx...

方案3：Python requests库上传 #

同步HTTP请求方式，仅需依赖requests包：

import requests

# 替换占位符
storage_account_name = "<存储账户名>"
container_name = "<容器名>"
blob_name = "<Blob名>"
sas_token = "<SAS_TOKEN>"
local_file_path = "<本地文件路径>"
content_type = "application/octet-stream"  # 根据文件类型调整

# 构造上传URL
upload_url = f"https://{storage_account_name}.blob.core.windows.net/{container_name}/{blob_name}?{sas_token}"

# 读取文件并发起上传请求
with open(local_file_path, "rb") as f:
    response = requests.put(upload_url, data=f, headers={"Content-Type": content_type})

response.raise_for_status()  # 捕获HTTP错误
print(f"上传成功，状态码：{response.status_code}")

方案4：Python httpx库上传 #

支持同步/异步的HTTP客户端，以下为同步版本示例：

import httpx

# 替换占位符
storage_account_name = "<存储账户名>"
container_name = "<容器名>"
blob_name = "<Blob名>"
sas_token = "<SAS_TOKEN>"
local_file_path = "<本地文件路径>"
content_type = "application/octet-stream"

upload_url = f"https://{storage_account_name}.blob.core.windows.net/{container_name}/{blob_name}?{sas_token}"

# 上传文件
with open(local_file_path, "rb") as f:
    with httpx.Client() as client:
        response = client.put(upload_url, content=f, headers={"Content-Type": content_type})

response.raise_for_status()
print(f"上传成功，状态码：{response.status_code}")

内容的提问来源于stack exchange，提问作者576i