使用 Python 版本 2.1 客户端库的 Azure Blob 存储代码示例

项目
09/04/2023

本文介绍使用适用于 Python 的 Azure Blob 存储客户端库版本 2.1 的代码示例。

2023 年 3 月 31 日，我们停用了对不符合当前 Azure SDK 指南的 Azure SDK 库的支持。新的 Azure SDK 库会定期更新，以推动一致的体验并增强安全态势。建议转换到新的 Azure SDK 库，以利用新功能和关键安全更新。

尽管 2023 年 3 月 31 日之后仍然可以使用较旧的库，但它们将不再从 Microsoft 获得官方支持和更新。有关详细信息，请参阅支持停用公告。

使用 Blob 存储构建高可用性应用

相关文章：教程：使用 Blob 存储构建高可用性应用程序

下载示例

下载示例项目并解压缩 storage-python-circuit-breaker-pattern-ha-apps-using-ra-grs.zip 文件。也可使用 git 将应用程序的副本下载到开发环境。示例项目包含一个基本的 Python 应用程序。

git clone https://github.com/Azure-Samples/storage-python-circuit-breaker-pattern-ha-apps-using-ra-grs.git

配置示例

在应用程序中，必须提供存储帐户凭据。可以将此信息存储在运行应用程序的本地计算机的环境变量中。根据操作系统，按照下面的某个示例创建环境变量。

在 Azure 门户中导航到存储帐户。在存储帐户的“设置” 下选择“访问密钥” 。将“存储帐户名称”和“密钥”值粘贴到以下命令中，替换 <youraccountname> 和 <youraccountkey> 占位符。此命令会将环境变量保存到本地计算机。在 Windows 中，必须重载正在使用的命令提示符或 shell，该环境变量才可用。

Linux

export accountname=<youraccountname>
export accountkey=<youraccountkey>

Windows

setx accountname "<youraccountname>"
setx accountkey "<youraccountkey>"

运行控制台应用程序

若要在终端或命令提示符处运行应用程序，请转到 circuitbreaker.py 目录，然后输入 python circuitbreaker.py。应用程序将 HelloWorld.png 映像从解决方案上传到存储帐户。应用程序会进行检查，以确保映像已复制到辅助 RA-GZRS 终结点。然后，它开始下载映像，最多可下载 999 次。每次读取由 P 或 S 表示。其中，P 表示主终结点，S 表示辅助终结点。

Screnshot of console app running.

在示例代码中，使用了 circuitbreaker.py 文件中的 run_circuit_breaker 方法通过 get_blob_to_path 方法来下载存储帐户中的映像。

存储对象重试函数设置为线性重试策略。重试函数决定了是否重试某个请求，并指定了在重试该请求之前需要等待的秒数。如果应该在针对主终结点的初始请求失败以后重试针对次要终结点的请求，请将 retry_to_secondary 值设置为 true。在示例应用程序的存储对象的 retry_callback 函数中，定义了自定义重试策略。

在下载之前，定义了服务对象的 retry_callback 和 response_callback 函数。这些函数定义了在下载成功完成或下载失败并重试时触发的事件处理程序。

了解代码示例

Retry 事件处理程序

当映像下载失败并设置为重试时，将调用 retry_callback 事件处理程序。如果达到应用程序中定义的重试次数上限，请求的 LocationMode 会变为 SECONDARY。此设置强制应用程序从辅助终结点尝试下载该图像。此配置减少了请求获取映像所需的时间，因为不会无限重试主终结点。

def retry_callback(retry_context):
    global retry_count
    retry_count = retry_context.count
    sys.stdout.write(
        "\nRetrying event because of failure reading the primary. RetryCount= {0}".format(retry_count))
    sys.stdout.flush()

    # Check if we have more than n-retries in which case switch to secondary
    if retry_count >= retry_threshold:

        # Check to see if we can fail over to secondary.
        if blob_client.location_mode != LocationMode.SECONDARY:
            blob_client.location_mode = LocationMode.SECONDARY
            retry_count = 0
        else:
            raise Exception("Both primary and secondary are unreachable. "
                            "Check your application's network connection.")

RequestCompleted 事件处理程序

当图像下载成功时，会调用 response_callback 事件处理程序。如果应用程序使用的是辅助终结点，则应用程序会继续使用该终结点，最多 20 次。 20 次以后，应用程序会将 LocationMode 重新设置为 PRIMARY 并重试主终结点。如果请求成功，应用程序会继续从主终结点读取。

def response_callback(response):
    global secondary_read_count
    if blob_client.location_mode == LocationMode.SECONDARY:

        # You're reading the secondary. Let it read the secondary [secondaryThreshold] times,
        # then switch back to the primary and see if it is available now.
        secondary_read_count += 1
        if secondary_read_count >= secondary_threshold:
            blob_client.location_mode = LocationMode.PRIMARY
            secondary_read_count = 0

Share via