pip install redis

import redis

接続プールを作成します。
pool = redis.ConnectionPool(
    host='192.xx.xx.2',            # インスタンスのVIPアドレス
    port=6379,                      # インスタンスポート
    password='crs-xxxx:パスワード',       # デフォルトアカウントのパスワード形式：インスタンスID:パスワード
    db=0,                           # データベース番号
    max_connections=200,            # 最大接続数
    socket_connect_timeout=2,       # 接続タイムアウト（秒）
    socket_timeout=2,               # 読み書きタイムアウト（秒）
    decode_responses=True,          # 文字列に自動デコードします。
    health_check_interval=30        # ヘルスチェック間隔（秒）
)

Redisクライアントを作成します。
r = redis.Redis(connection_pool=pool)

# データを書き込みます。
r.set('name', 'Distributed Cache')

# データを読み取ります。
value = r.get('name')
print(f'get name: {value}')

# 有効期限付きのKeyを設定します。
r.setex('session:user123', 3600, 'session_data')

# ハッシュ操作
r.hset('user:1001', mapping={'name': '張三', 'age': '28'})
user_info = r.hgetall('user:1001')
print(f'user info: {user_info}')

# リスト操作
r.lpush('queue:tasks', 'task1', 'task2', 'task3')
task = r.rpop('queue:tasks')
print(f'dequeued task: {task}')

# Pipeline バッチ操作
pipe = r.pipeline()
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.set('key3', 'value3')
pipe.get('key1')
pipe.get('key2')
pipe.get('key3')
results = pipe.execute()
print(f'pipeline results: {results}')

import redis

# カスタムアカウントのパスワード形式：ユーザー名@パスワード
pool = redis.ConnectionPool(
    host='192.xx.xx.2',
    port=6379,
    password='myuser@MyPassword',   # プロキシモードのカスタムアカウント：ユーザー名@パスワード全体をパスワードとして渡します。
    db=0,
    max_connections=200,
    socket_connect_timeout=2,
    socket_timeout=2,
    decode_responses=True
)

r = redis.Redis(connection_pool=pool)
r.set('key', 'value')
print(r.get('key'))

import redis

# SSL暗号化接続
pool = redis.ConnectionPool(
    host='192.xx.xx.2',
    port=6379,
    password='crs-xxxx:パスワード',
    db=0,
    max_connections=200,
    socket_connect_timeout=2,
    socket_timeout=2,
    decode_responses=True,
    ssl=True,                       # SSLを有効にします。
    ssl_ca_certs='/path/to/ca.pem', # CA証明書のパス
    ssl_cert_reqs='required'        # サーバー証明書を検証します。
)

r = redis.Redis(connection_pool=pool)
r.set('ssl_test', 'encrypted_connection')
print(r.get('ssl_test'))

import asyncio
import redis.asyncio as aioredis

async def main():
    # 非同期接続プールを作成します。
    pool = aioredis.ConnectionPool(
        host='192.xx.xx.2',
        port=6379,
        password='crs-xxxx:パスワード',
        db=0,
        max_connections=200,
        socket_connect_timeout=2,
        socket_timeout=2,
        decode_responses=True
    )

    r = aioredis.Redis(connection_pool=pool)

    # 非同期書き込み
    await r.set('async_key', 'async_value')

    # 非同期読み取り
    value = await r.get('async_key')
    print(f'async get: {value}')

    # 非同期Pipeline
    pipe = r.pipeline()
    pipe.set('key1', 'value1')
    pipe.set('key2', 'value2')
    pipe.get('key1')
    pipe.get('key2')
    results = await pipe.execute()
    print(f'async pipeline: {results}')

    # 接続を閉じます。
    await r.aclose()
    await pool.aclose()

asyncio.run(main())