クイックスタート: Python で Azure Cache for Redis を使用する

キャッシュで Microsoft Entra ID の認証を有効にする

既にキャッシュがある場合は、まず Microsoft Entra 認証が有効になっているかどうかを確認します。 そうでない場合は、有効にします。 アプリケーションには Microsoft Entra ID を使用することをお勧めします。

1. Azure portal で、Microsoft Entra トークンベースの認証を使用する対象の Azure Cache for Redis インスタンスを選択します。

2. リソース メニューで [認証] を選択します。

3. 作業ペインで、[Microsoft Entra 認証を有効にする] がオンになっているかどうかを確認します。 その場合は、次に進みます。

4. [Enable Microsoft Entra 認証] を選択し、有効なユーザー名を入力します。 [保存] を選択した場合は、既定で入力したユーザーに [データ所有者アクセス ポリシー] が自動的に割り当てられます。 マネージド ID またはサービス プリンシパルを入力して、キャッシュ インスタンスに接続することもできます。

   

5. 構成を更新するかどうか確認するポップアップ ダイアログ ボックスが表示されて、数分かかることが通知されます。 [はい] を選択します。

   重要

   有効化の操作が完了すると、キャッシュ インスタンス内のノードが再起動して新しい構成が読み込まれます。 そのため、この操作はメンテナンス期間中またはピーク営業時間外に実行することをお勧めします。 この操作には最大 30 分かかることがあります。

Azure CLI で Microsoft Entra ID を使用する方法については、ID のリファレンス ページを参照してください。

Microsoft 認証ライブラリをインストールする

1. Microsoft 認証ライブラリ (MSAL) をインストールします。 このライブラリを使用すると、Microsoft ID からセキュリティ トークンを取得してユーザーを認証できます。

2. MSAL を使用してトークン認証サポートを提供する、Python Azure ID クライアント ライブラリを使用できます。 pip を使用して、このライブラリをインストールします。

pip install azure-identity

Microsoft Entra ID を使用して Python スクリプトを作成する

1. 新しいテキスト ファイルを作成し、次のスクリプトを追加して、ファイルを PythonApplication1.py として保存します。

2. <Your Host Name> を Azure Cache for Redis インスタンスの値に置き換えます。 ホスト名は、<DNS name>.redis.cache.windows.net の形式になります。

3. <Your Username> を Microsoft Entra ID ユーザーの値に置き換えます。

   import redis
   from azure.identity import DefaultAzureCredential
   
   scope = "https://redis.azure.com/.default"
   host = "<Your Host Name>"
   port = 6380
   user_name = "<Your Username>"
   
   
   def hello_world():
       cred = DefaultAzureCredential()
       token = cred.get_token(scope)
       r = redis.Redis(host=host,
                       port=port,
                       ssl=True,    # ssl connection is required.
                       username=user_name,
                       password=token.token,
                       decode_responses=True)
       result = r.ping()
       print("Ping returned : " + str(result))
   
       result = r.set("Message", "Hello!, The cache is working with Python!")
       print("SET Message returned : " + str(result))
   
       result = r.get("Message")
       print("GET Message returned : " + result)
   
       result = r.client_list()
       print("CLIENT LIST returned : ")
       for c in result:
           print(f"id : {c['id']}, addr : {c['addr']}")
   
   if __name__ == '__main__':
       hello_world()
   

4. ターミナルから Python コードを実行する前に、ターミナルで Microsoft Entra ID を使用することを認可していることを確認します。

   azd auth login

5. Python で PythonApplication1.py を実行します。 次の例のような結果が表示されるはずです。

   

再認証を使用した Python スクリプトの作成

Microsoft Entra ID アクセス トークンの有効期限は限られており、平均 75 分です。 キャッシュへの接続を維持するには、トークンを更新する必要があります。 この例では、Python を使用してこれを行う方法を示します。

1. 新しいテキスト ファイルを作成し、次のスクリプトを追加します。 次に、ファイルを PythonApplication2.py として保存します。

2. <Your Host Name> を Azure Cache for Redis インスタンスの値に置き換えます。 ホスト名は、<DNS name>.redis.cache.windows.net の形式になります。

3. <Your Username> を Microsoft Entra ID ユーザーの値に置き換えます。

   import time
   import logging
   import redis
   from azure.identity import DefaultAzureCredential
   
   scope = "https://redis.azure.com/.default"
   host = "<Your Host Name>"
   port = 6380
   user_name = "<Your Username>"
   
   def re_authentication():
       _LOGGER = logging.getLogger(__name__)
       cred = DefaultAzureCredential()
       token = cred.get_token(scope)
       r = redis.Redis(host=host,
                       port=port,
                       ssl=True,   # ssl connection is required.
                       username=user_name,
                       password=token.token,
                       decode_responses=True)
       max_retry = 3
       for index in range(max_retry):
           try:
               if _need_refreshing(token):
                   _LOGGER.info("Refreshing token...")
                   tmp_token = cred.get_token(scope)
                   if tmp_token:
                       token = tmp_token
                   r.execute_command("AUTH", user_name, token.token)
               result = r.ping()
               print("Ping returned : " + str(result))
   
               result = r.set("Message", "Hello!, The cache is working with Python!")
               print("SET Message returned : " + str(result))
   
               result = r.get("Message")
               print("GET Message returned : " + result)
   
               result = r.client_list()
               print("CLIENT LIST returned : ")
               for c in result:
                   print(f"id : {c['id']}, addr : {c['addr']}")
               break
           except redis.ConnectionError:
               _LOGGER.info("Connection lost. Reconnecting.")
               token = cred.get_token(scope)
               r = redis.Redis(host=host,
                               port=port,
                               ssl=True,   # ssl connection is required.
                               username=user_name,
                               password=token.token,
                               decode_responses=True)
           except Exception:
               _LOGGER.info("Unknown failures.")
               break
   
   
   def _need_refreshing(token, refresh_offset=300):
       return not token or token.expires_on - time.time() < refresh_offset
   
   if __name__ == '__main__':
       re_authentication()
   

4. Python で PythonApplication2.py を実行します。 次の例のような結果が表示されるはずです。

   

   最初の例とは異なり、トークンの有効期限が切れた場合、この例ではトークンが自動的に更新されます。

Azure portal からホスト名、ポート、アクセス キーを取得する

Azure Cache for Redis サーバーに接続するには、キャッシュ クライアントにキャッシュのホスト名、ポート、およびキーが必要です。 クライアントによっては、これらの項目を参照するための名前が若干異なる場合があります。 ホスト名、ポート、およびキーは Azure portal から取得できます。

• アクセス キーを取得するには、[リソース] メニューから [認証] を選択します。 その後、[アクセス キー] タブを選択します。

  

• キャッシュのホスト名とポートを取得するには、[リソース] メニューから [概要] を選択します。 ホスト名は、<DNS 名>.redis.cache.windows.net の形式になります。

  

コマンド ラインからキャッシュへの読み取りと書き込みを行う

コマンド ラインから Python を実行して、キャッシュをテストします。 まず、コマンド ラインで「py」と入力して Python インタープリターを開始し、次のコードを使用します。 <Your Host Name> と <Your Access Key> を、Azure Cache for Redis インスタンスの値に置き換えます。 ホスト名は、<DNS name>.redis.cache.windows.net の形式になります。

>>> import redis
>>> r = redis.Redis(host='<Your Host Name>',
        port=6380, db=0, password='<Your Access Key>', ssl=True)
>>> r.set('foo', 'bar')
True
>>> r.get('foo')
b'bar'

アクセス キーを使用して Python スクリプトを作成する

新しいテキスト ファイルを作成し、次のスクリプトを追加して、ファイルを PythonApplication1.py として保存します。 <Your Host Name> と <Your Access Key> を、Azure Cache for Redis インスタンスの値に置き換えます。 ホスト名は、<DNS name>.redis.cache.windows.net の形式になります。

import redis

myHostname = "<Your Host Name>"
myPassword = "<Your Access Key>"

r = redis.Redis(host=myHostname, port=6380,
                      password=myPassword, ssl=True)

result = r.ping()
print("Ping returned : " + str(result))

result = r.set("Message", "Hello!, The cache is working with Python!")
print("SET Message returned : " + str(result))

result = r.get("Message")
print("GET Message returned : " + result)

result = r.client_list()
print("CLIENT LIST returned : ")
for c in result:
    print(f"id : {c['id']}, addr : {c['addr']}")

Python で PythonApplication1.py を実行します。 次の例のような結果が表示されるはずです。