共用方式為

使用 Python 2.1 版用戶端連結庫 Azure Blob 儲存體 程式代碼範例

  • 發行項

本文說明使用適用於 Python 之 Azure Blob 儲存體 客戶端連結庫 2.1 版的程式代碼範例。

在 2023 年 3 月 31 日，我們已淘汰不支援不符合 目前 Azure SDK 指導方針的 Azure SDK 連結庫。 新的 Azure SDK 連結庫會定期更新，以推動一致的體驗，並強化您的安全性狀態。 建議您轉換至新的 Azure SDK 連結庫，以利用新功能和重大安全性更新。

雖然舊版連結庫仍然可以在 2023 年 3 月 31 日之後使用，但將不再收到 Microsoft 的官方支援和更新。 如需詳細資訊，請參閱 支援淘汰公告。

使用 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 中，必須重新載入命令提示字元或您所使用的殼層，才能使用此環境變數。

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 則代表次要端點。

在程式碼範例中，會使用 circuitbreaker.py 檔案中的 run_circuit_breaker 方法，透過 get_blob_to_path 方法從儲存體帳戶下載影像。

儲存物件重試函式會設為線性重試原則。 重試函式會決定是否要重試要求，並指定重試要求之前所等待的秒數。 如果對主要端點的初始要求失敗時應對次要端點重試該要求，請將 retry_to_secondary 值設為 true。 在範例應用程式中，自訂重試原則會定義於儲存體物件的 retry_callback 函式中。

在下載之前，服務物件 retry_callback 和 response_callback 函式會先加以定義。 這些函式會定義事件處理常式，當下載作業成功完成，或下載作業失敗而正在重試時，便會引發這些處理常式。

瞭解程式代碼範例

重試事件處理常式

當影像下載失敗，並設定為重試時，便會呼叫 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.")

要求已完成的事件處理常式

當影像下載成功時，便會呼叫 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