快速入门：将 Azure Redis 与 Python 配合使用

本文介绍如何使用带有 Python 语言的 Azure Redis 缓存并使用 Microsoft Entra ID 进行连接。

先决条件

• Azure 订阅 - 创建免费帐户
• 安装 Python 3.7+ 语言环境
• 将这些导入添加到项目和开发环境中
  • redis - Redis Python 客户端
  • redis-entraid - Redis Microsoft Entra ID 身份验证扩展
  • azure-identity - Azure 身份验证库

创建 Azure 托管 Redis 实例

首先，创建缓存。 可以通过 Azure 门户使用 Azure 托管 Redis 或 Azure Redis 缓存创建缓存。 在本快速入门中，我们使用 Azure 托管 Redis。

创建缓存时，默认情况下会启用 Microsoft Entra ID，使其从头开始安全。 你的缓存在此 QuickStart 中还必须使用公共终结点。

若要使用门户创建缓存，请遵循以下过程之一：

• Azure 托管 Redis
• 用于 Redis 的 Azure 缓存

（可选）可以使用 Azure CLI、PowerShell（无论你喜欢哪种方式）创建缓存。

用于连接到 Redis 缓存的代码

在代码示例的第一部分，设置您的连接至缓存。

• Azure 托管 Redis 和企业版缓存的端口：10000
• Azure Redis 缓存实例的端口：6380

import redis
from azure.identity import DefaultAzureCredential
from redis_entraid.cred_provider import create_from_default_azure_credential

# Connection details for your cache
# Get the connection details for the Redis instance
redis_host = "contosob116.westus3.redis.azure.net"
redis_port = 10000  #For an Azure 

print("🚀 Starting Azure Redis Cache connection test...")
print(f"📡 Connecting to: {redis_host}:{redis_port}")

# Validate configuration
if not redis_host or not redis_port:
    print("❌ Error: Redis host and port must be configured")
    exit(1)

print()  # Add a new line

try:
    # Create credential provider using DefaultAzureCredential for Azure Entra ID authentication
    credential_provider = create_from_default_azure_credential(
         ("https://redis.azure.com/.default",),)

    # Create a Redis client with Azure Entra ID authentication
    r = redis.Redis(host=redis_host, 
                    port=redis_port, 
                    ssl=True, 
                    decode_responses=True, 
                    credential_provider=credential_provider,
                    socket_timeout=10,
                    socket_connect_timeout=10
                    )

测试连接的代码

在下一部分中，使用返回ping值的 Redis 命令True测试连接。

# Ping the Redis server to test the connection
result = r.ping()
if result:
    print("Ping returned: ", result)

代码设置密钥，获取密钥

在本部分中，使用基本 set 和 get 序列以最简单的方式开始使用 Redis 缓存。

    # Create a simple set and get operation
    result = r.set("Message", "Hello, The cache is working with Python!")
    print("✅ SET Message succeeded: " + str(result))
    print()  # Add a new line

    value = r.get("Message")

    if value is not None:
        print("✅ GET Message returned : " + str(value))
        print()  # Add a new line
    else:
        print("⚠️  GET Message returned None")
        print()  # Add a new line
    
    print("🎉 All Redis operations completed successfully!")
    print()  # Add a new line

在运行此代码之前，您必须先将自己添加为 Redis 缓存的用户。

您还必须通过命令行使用 Azure 命令行工具或 Azure 开发人员命令行工具（azd）授权连接到 Azure。

还应 将用户或系统主体添加到缓存。 添加任何可能在 Redis 缓存中以用户身份运行程序的人。

结果如下所示：

C:\utils\python-quickstart>python quickstart-amr.py
🚀 Starting Azure Redis Cache connection test...
📡 Connecting to: contosob116.westus3.redis.azure.net:10000

✅ Ping returned : True

✅ SET Message succeeded: True

✅ GET Message returned : Hello, The cache is working with Python!

🎉 All Redis operations completed successfully!

🔐 Redis connection closed

在这里，你可以完整地查看此代码示例。 代码包含从前面的代码说明中省略的一些错误检查，以便于简单起见。 最后一步是关闭与缓存的连接。

# Python Quickstart using Azure Entra ID authentication
# Azure Managed Redis cache that you created using the Azure portal, or CLI
# This script demonstrates secure connection using Microsoft Entra ID authentication
# This script demonstrates secure connection using the default Azure credential provider
# You should be a user on the cache and logged in to Azure CLI with the same account using `az login`

import redis
from azure.identity import DefaultAzureCredential
from redis_entraid.cred_provider import create_from_default_azure_credential

# Connection details for your cache
# Get the connection details for the Redis instance
redis_host = "<host-url>"  # Replace with your cache info
redis_port = <port number>  # Replace with your cache info 

print("🚀 Starting Azure Redis Cache connection test...")
print(f"📡 Connecting to: {redis_host}:{redis_port}")

# Validate configuration
if not redis_host or not redis_port:
    print("❌ Error: Redis host and port must be configured")
    exit(1)

print()  # Add a new line

try:
    # Create credential provider using DefaultAzureCredential for Azure Entra ID authentication
    credential_provider = create_from_default_azure_credential(
         ("https://redis.azure.com/.default",),)

    # Create a Redis client with Azure Entra ID authentication
    r = redis.Redis(host=redis_host, 
                    port=redis_port, 
                    ssl=True, 
                    decode_responses=True, 
                    credential_provider=credential_provider,
                    socket_timeout=10,
                    socket_connect_timeout=10
                    )

    # Test connection 
    result = r.ping()
    print("✅ Ping returned : " + str(result))
    print()  # Add a new line

    # Create a simple set and get operation
    result = r.set("Message", "Hello, The cache is working with Python!")
    print("✅ SET Message succeeded: " + str(result))
    print()  # Add a new line

    value = r.get("Message")

    if value is not None:
        print("✅ GET Message returned : " + str(value))
        print()  # Add a new line
    else:
        print("⚠️  GET Message returned None")
        print()  # Add a new line
    
    print("🎉 All Redis operations completed successfully!")
    print()  # Add a new line

except redis.ConnectionError as e:
    print(f"❌ Connection error: {e}")
    print("💡 Check if Redis host and port are correct, and ensure network connectivity")
    print()  # Add a new line
except redis.AuthenticationError as e:
    print(f"❌ Authentication error: {e}")
    print("💡 Check if Azure Entra ID authentication is properly configured")
    print()  # Add a new line
except redis.TimeoutError as e:
    print(f"❌ Timeout error: {e}")
    print("💡 Check network latency and Redis server performance")
    print()  # Add a new line
except Exception as e:
    print(f"❌ Unexpected error: {e}")
    if "999" in str(e):
        print("💡 Error 999 typically indicates a network connectivity issue or firewall restriction")
        print("   - Verify the Redis hostname is correct")
        print("   - Verify that you have logged in with Az CLI")
        print("   - Ensure the Redis cache is running and accessible")
    print()  # Add a new line
finally:
    # Clean up connection if it exists
    if 'r' in locals():
        try:
            r.close()
            print("🔐 Redis connection closed")
        except Exception as e:
            print(f"❌ Error closing connection: {e}")

清理资源

要继续使用在本文中创建的资源，请保留资源组。

否则，如果您已不再需要使用这些资源，可以删除您创建的 Azure 资源组以避免产生费用。

删除资源组的步骤

1. 登录到 Azure 门户，然后选择“资源组”。

2. 选择要删除的资源组。

   如果有多个资源组，请使用“筛选任何字段...”框，键入为本文创建的资源组的名称。 在结果列表中选择资源组。

   

3. 选择“删除资源组”。

4. 系统会要求确认是否删除资源组。 键入资源组的名称进行确认，然后选择“删除”。

   

片刻之后，将会删除该资源组及其所有资源。

相关内容

• 用于使用 Microsoft Entra ID 进行连接的 Redis 扩展
• redis-py 指南