如何用Python上传文件至ADLS Gen2？遇403授权失败问题

一、排查requests实现的403 AuthorizationFailure错误 #

从你提供的示例路径abfss://fs-adls@adlsprod.dfs.core.windows.net/FileStore/abc.txt，可提取正确参数：

• account_name: adlsprod
• fs_name: fs-adls

1. 确认URL结构正确性 #

创建文件系统（mkfs）的REST API URL必须遵循以下格式：
https://{account_name}.dfs.core.windows.net/{fs_name}?resource=filesystem
核心注意点：account_name在域名段，fs_name在路径开头，切勿颠倒顺序

2. 排查权限与认证问题 #

• 若使用账户密钥，需确保Authorization头的HMAC-SHA256签名生成正确，签名内容必须包含HTTP方法、UTC日期、资源路径等关键参数，签名错误会直接返回403。
• 若使用SAS token，需确认token包含filesystem资源的create权限，且处于有效期内，同时要将token拼接在URL末尾（格式：?resource=filesystem&<sas_token>）。
• 检查操作账户/服务主体是否拥有ADLS Gen2的Storage Blob Data Contributor角色，仅Reader角色无法创建文件系统。

正确的requests示例片段 #

import requests
import hmac
import hashlib
import base64
from datetime import datetime

account_name = "adlsprod"
fs_name = "fs-adls"
account_key = "your_account_key"

# 构建请求URL
url = f"https://{account_name}.dfs.core.windows.net/{fs_name}?resource=filesystem"

# 生成Authorization头
date = datetime.utcnow().strftime('%a, %d %b %Y %H:%M:%S GMT')
string_to_sign = f"PUT\n\n\n{date}\n/{account_name}/{fs_name}\nresource:filesystem"
signature = base64.b64encode(hmac.new(base64.b64decode(account_key), string_to_sign.encode('utf-8'), hashlib.sha256).digest()).decode()
auth_header = f"SharedKey {account_name}:{signature}"

# 发送请求
response = requests.put(url, headers={"Authorization": auth_header, "x-ms-date": date, "x-ms-version": "2020-04-08"})
print(response.status_code, response.text)

二、其他Python上传文件至ADLS Gen2的方法 #

1. 使用官方azure-storage-file-datalake库（推荐） #

微软官方维护的库，封装了所有ADLS Gen2操作，无需手动处理签名和REST细节。

安装命令：pip install azure-storage-file-datalake

示例代码（账户密钥认证）：

from azure.storage.filedatalake import DataLakeServiceClient

account_name = "adlsprod"
account_key = "your_account_key"
fs_name = "fs-adls"
local_file_path = "path/to/your/local/file.txt"
adls_file_path = "FileStore/uploaded_file.txt"

# 初始化服务客户端
service_client = DataLakeServiceClient(account_url=f"https://{account_name}.dfs.core.windows.net", credential=account_key)

# 创建文件系统（如果不存在）
file_system_client = service_client.get_file_system_client(file_system=fs_name)
if not file_system_client.exists():
    file_system_client.create_file_system()

# 上传文件
directory_client = file_system_client.get_directory_client("FileStore")
file_client = directory_client.create_file("uploaded_file.txt")

with open(local_file_path, "rb") as f:
    file_client.upload_data(f, overwrite=True)

Azure AD认证（服务主体）示例：

from azure.storage.filedatalake import DataLakeServiceClient
from azure.identity import ClientSecretCredential

tenant_id = "your_tenant_id"
client_id = "your_client_id"
client_secret = "your_client_secret"
account_name = "adlsprod"

credential = ClientSecretCredential(tenant_id, client_id, client_secret)
service_client = DataLakeServiceClient(account_url=f"https://{account_name}.dfs.core.windows.net", credential=credential)
# 后续文件上传操作与上述一致

2. 调用AzCopy命令行（适合大文件/批量上传） #

通过Python的subprocess调用AzCopy，自动处理断点续传、并行上传，适合大文件或批量迁移场景。

示例代码：

import subprocess

# AzCopy批量上传命令
command = [
    "azcopy", "copy",
    "path/to/local/files/*",
    "abfss://fs-adls@adlsprod.dfs.core.windows.net/FileStore",
    "--recursive"
]

result = subprocess.run(command, capture_output=True, text=True)
print(result.stdout)
if result.stderr:
    print(result.stderr)

注意：需提前安装AzCopy并完成认证（如通过az login登录Azure账号，或使用SAS token）。

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