如何用Python高效检查ADLS Gen2存储账户是否存在？

我需要读取多个存储账户中的特定文件，但部分存储账户尚未创建。目前我用下面的代码检查存储账户是否存在，但执行耗时超过一分钟才返回ServiceRequestError错误，想缩短执行时间，有没有更优方案？

代码示例：

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

credential=DefaultAzureCredential()
service_client=DataLakeServiceClient("https://somerandomaccountname.dfs.core.windows.net",credential=credential)
file_client=service_client.get_file_client(file_system="non_existing",file_path="file.name")
file_client.exists()

错误信息：

ServiceRequestError: <urllib3.connection.HTTPSConnection object at 0x0000021719BE13D0>: Failed to resolve 'somerandomaccountname.blob.core.windows.net' ([Errno 11001] getaddrinfo failed)

问题根源 #

超时是因为不存在的存储账户域名无法被DNS解析，而Azure SDK默认的重试策略会多次尝试连接，导致耗时过长。

方案1：DNS预解析快速判断 #

直接对存储账户域名做DNS解析，解析失败就直接判定账户不存在，跳过Azure SDK的请求流程，耗时极短：

import socket
from azure.storage.filedatalake import DataLakeServiceClient
from azure.identity import DefaultAzureCredential

def is_storage_account_exists(account_name):
    domain = f"{account_name}.dfs.core.windows.net"
    try:
        # 设置2秒超时，快速判断域名是否可解析
        socket.getaddrinfo(domain, 443, timeout=2)
        return True
    except socket.gaierror:
        return False

# 使用示例
account_name = "somerandomaccountname"
if is_storage_account_exists(account_name):
    credential = DefaultAzureCredential()
    service_client = DataLakeServiceClient(f"https://{account_name}.dfs.core.windows.net", credential=credential)
    file_client = service_client.get_file_client(file_system="non_existing", file_path="file.name")
    print(file_client.exists())
else:
    print("存储账户不存在")

方案2：调整Azure SDK的重试与超时参数 #

如果不想做DNS解析，直接修改SDK的重试策略和请求超时，减少等待时间：

from azure.storage.filedatalake import DataLakeServiceClient
from azure.identity import DefaultAzureCredential
from azure.core.pipeline.policies import RetryPolicy

# 自定义重试策略：只重试特定HTTP状态码，减少重试次数和间隔
retry_policy = RetryPolicy(
    retry_total=1,
    retry_backoff_factor=0.1,
    retry_on_status_codes=[408, 429, 500, 502, 503, 504]
)

credential = DefaultAzureCredential()
service_client = DataLakeServiceClient(
    "https://somerandomaccountname.dfs.core.windows.net",
    credential=credential,
    retry_policy=retry_policy,
    connection_timeout=2,
    read_timeout=2
)

try:
    file_client = service_client.get_file_client(file_system="non_existing", file_path="file.name")
    print(file_client.exists())
except Exception as e:
    print("存储账户不存在或其他错误:", str(e))

方案3：通过Azure Resource Manager(ARM) API查询（需权限） #

如果拥有Azure订阅权限，直接调用ARM API查询存储账户是否存在，结果最准确，不受DNS缓存影响：

from azure.identity import DefaultAzureCredential
from azure.mgmt.storage import StorageManagementClient

def check_storage_account_via_arm(subscription_id, resource_group, account_name):
    credential = DefaultAzureCredential()
    storage_client = StorageManagementClient(credential, subscription_id)
    try:
        storage_client.storage_accounts.get_properties(resource_group, account_name)
        return True
    except Exception:
        # 账户不存在时会抛出ResourceNotFoundError
        return False

# 使用示例
subscription_id = "你的订阅ID"
resource_group = "你的资源组名称"
account_name = "somerandomaccountname"

if check_storage_account_via_arm(subscription_id, resource_group, account_name):
    print("存储账户存在")
else:
    print("存储账户不存在")

注意：使用该方案需要身份凭证拥有Microsoft.Storage/storageAccounts/read权限。

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