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

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

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

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

2. サービス メニューの [設定] で、[認証] を選びます。

3. [認証] ペインで、[Microsoft Entra 認証を有効にする] チェック ボックスがオンになっているかどうかを調べます。 そうである場合は、次のセクションに進むことができます。

4. そうでない場合は、[Microsoft Entra 認証を有効にする] チェック ボックスをオンにします。 次に、有効なユーザーの名前を入力します。 [保存] を選択します。 入力したユーザー名には、データ所有者アクセス ポリシーが自動的に割り当てられます。

   マネージド ID またはサービス プリンシパルを入力して、キャッシュに接続することもできます。

   

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

   重要

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

Azure CLI での Microsoft Entra ID の使用については、ID のリファレンス ページをご覧ください。

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

Microsoft 認証ライブラリ (MSAL) は、ユーザーの認証を行うために Microsoft ID プラットフォームからセキュリティ トークンを取得します。

MSAL をインストールするには:

1. Python 用の MSAL をインストールします。

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

   pip を使ってこのライブラリをインストールします。

pip install azure-identity

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

1. テキスト ファイルを作成します。 ファイルを PythonApplication1.py として保存します。

2. PythonApplication1.py で、次のスクリプトを追加して変更します。

   スクリプトで:

   • <Your Host Name> を Azure Cache for Redis インスタンスの値に置き換えます。 ホスト名の形式は <DNS name>.redis.cache.windows.net です。
   • <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()
   

3. ターミナルで Python コードを実行する前に、ターミナルで Microsoft Entra ID の使用を認可します。

   azd auth login

4. Python を使って PythonApplication1.py ファイルを実行します。 出力が次の例のようになることを確認します。

   

再認証を使用して Python スクリプトを作成する

Microsoft Entra ID アクセス トークンの有効期間は限られており、約 75 分です。 キャッシュへの接続を維持するには、トークンを更新する必要があります。

この例では、Python を使ってトークンを更新する方法を見ていきます。

1. テキスト ファイルを作成します。 ファイルを PythonApplication2.py として保存します。

2. PythonApplication2.py で、次のスクリプトを追加して変更します。

   スクリプトで:

   • <Your Host Name> を Azure Cache for Redis インスタンスの値に置き換えます。 ホスト名の形式は <DNS name>.redis.cache.windows.net です。
   • <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()
   

3. Python を使って PythonApplication2.py ファイルを実行します。 出力が次の例のようになることを確認します。

   

   前の例とは異なり、トークンの有効期間が切れると、この例のコードはトークンを自動的に更新します。

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

キャッシュ クライアントが Azure Cache for Redis サーバーに接続するには、キャッシュのホスト名、ポート、アクセス キーが必要です。 一部のクライアントでは、これらの項目が少し異なる名前で呼ばれている場合があります。 ホスト名、ポート、キーは Azure portal で取得できます。

• キャッシュのアクセス キーを取得するには:

  1. Azure portal で、キャッシュに移動します。
  2. サービス メニューの [設定] で、[認証] を選びます。
  3. [認証] ペインで、[アクセス キー] タブを選びます。
  4. アクセス キーの値をコピーするには、キー フィールドの [コピー] アイコンを選びます。

  

• キャッシュのホスト名とポートを取得するには:

  1. Azure portal で、キャッシュに移動します。
  2. サービス メニューで、[概要] を選びます。
  3. [基本] の [ホスト名] で、[コピー] アイコンを選んでホスト名の値をコピーします。 ホスト名の値の形式は <DNS name>.redis.cache.windows.net です。
  4. [ポート] で、[コピー] アイコンを選んでポートの値をコピーします。

  

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

キャッシュをテストするには、コマンド ラインから Pythonを実行します。

1. コマンド ラインで「py」と入力して、Python インタープリターを開始します。

2. 次のコードを変更して実行します。

   コードで、<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 スクリプトを作成する

1. 新しいテキスト ファイルを作成します。 ファイルを PythonApplication1.py として保存します。

2. PythonApplication1.py で、次のスクリプトを追加して変更します。

   スクリプトで:

   • <Your Host Name> を Azure Cache for Redis インスタンスの値に置き換えます。 ホスト名の形式は <DNS name>.redis.cache.windows.net です。
   • <Your Access Key> を実際の Microsoft Entra ID ユーザーの値に置き換えます。

   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']}")
   

3. Python を使って PythonApplication1.py ファイルを実行します。 次の例のような結果が出力として表示されることを確認します。