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

キャッシュで Microsoft Entra ID 認証を使用する

Azure Redis キャッシュでは、Enterprise および Enterprise Flash サービス レベルを除いて、Microsoft Entra 認証が既定で有効になっています。 アクセス キーは既定で無効になっています。

1. Azure portal で、Microsoft Entra トークン ベースの認証を使用するキャッシュを選択します。

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

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

   

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>.<region>.redis.azure.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 = 10000
   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 Managed Redis (プレビュー) インスタンスの値に置き換えます。 ホスト名は、<DNS name>.<region>.redis.azure.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 = 10000
   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=10000, 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>.<region>.redis.azure.net の形式になります。

import redis

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

r = redis.Redis(host=myHostname, port=10000,
                      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 を実行します。 次の例のような結果が表示されるはずです。

キャッシュで Microsoft Entra ID 認証を使用する

Azure Redis キャッシュでは、Enterprise および Enterprise Flash サービス レベルを除いて、Microsoft Entra 認証が既定で有効になっています。 アクセス キーは既定で無効になっています。

1. Azure portal で、Microsoft Entra トークン ベースの認証を使用するキャッシュを選択します。

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

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

   

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 を実行します。 次の例のような結果が表示されるはずです。