Python3.5使用DataLakeServiceClient错误凭据无报错 如何验证登录有效性

DataLakeServiceClient 初始化阶段仅在本地构建客户端对象，不会发起网络请求校验存储账号名、访问密钥的合法性，只有实际调用数据读写接口时才会触发认证流程。
你遇到的无限加载无报错问题，是Azure存储Python SDK默认重试策略导致的：错误凭据触发401/403类认证失败后，SDK会默认不断重试，未配置重试上限和超时的场景下就会一直卡住。

分为两步实现：先配置客户端重试&超时规则避免卡住，再增加轻量预校验逻辑提前判断凭据有效性。

• 调整客户端配置，限制重试次数：初始化DataLakeServiceClient时传入自定义重试策略，设置最大重试次数、超时时间，避免异常场景无限等待。
• 增加凭据预校验逻辑：初始化客户端后主动发起一次无副作用的轻量请求（比如拉取文件系统列表、查询指定文件系统属性），请求成功即可证明凭据有效，失败则直接抛出认证相关异常。

from azure.storage.filedatalake import DataLakeServiceClient
from azure.core.pipeline.policies import RetryPolicy
from azure.core.exceptions import HttpResponseError

service_client = None

def initialize_storage_account(storage_account_name, storage_account_key):
    global service_client
    # 自定义重试策略：最多重试2次，总超时时间10秒
    retry_policy = RetryPolicy(max_retries=2, retry_total=2, connection_timeout=10, read_timeout=10)
    try:
        service_client = DataLakeServiceClient(
            account_url="https://{}.dfs.core.windows.net".format(storage_account_name),
            credential=storage_account_key,
            retry_policy=retry_policy
        )
    except Exception as e:
        print("初始化客户端失败：", e)
        raise

def validate_credentials(target_file_system=None):
    """预校验凭据有效性，无副作用"""
    global service_client
    if not service_client:
        raise ValueError("客户端未初始化")
    try:
        if target_file_system:
            # 已知要操作的文件系统时用这个方法，权限要求更低
            fs_client = service_client.get_file_system_client(target_file_system)
            fs_client.get_file_system_properties()
        else:
            # 不知道具体文件系统时，拉取1条文件系统列表即可
            next(service_client.list_file_systems(results_per_page=1), None)
        return True
    except HttpResponseError as e:
        if e.status_code in (401, 403):
            print("凭据无效：认证失败，检查账号名、密钥是否正确")
        elif e.status_code == 404:
            print("指定的存储账号不存在或文件系统不存在")
        else:
            print("校验凭据失败：", e)
        raise
    except Exception as e:
        print("校验凭据异常：", e)
        raise

def create_directory(container_name, dir_name):
    try:
        file_system_client = service_client.get_file_system_client(file_system=container_name)
        file_system_client.create_directory(dir_name)
        print("Created Directory")
    except Exception as e:
        print("创建目录失败：", e)
        raise

if __name__ == "__main__":
    # 替换为你的实际参数
    storage_account = "your_account_name"
    storage_key = "your_account_key"
    target_container = "test_fs"
    
    initialize_storage_account(storage_account, storage_key)
    # 提前校验凭据，这里可以传要操作的容器名提高校验效率
    validate_credentials(target_file_system=target_container)
    # 校验通过后再执行业务逻辑
    create_directory(target_container, "test_dir")

• 预校验逻辑只会发起读请求，不会修改存储内的任何资源
• 配置重试策略后，即使不做预校验，后续调用create_directory等接口时也会在重试次数耗尽后抛出异常，不会再无限卡住
• Python3.5环境需要匹配对应版本的Azure存储SDK，建议使用azure-storage-file-datalake<=12.5.0，更高版本不再支持Python3.5

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