pip install redis

import redis

# Create a connection pool
pool = redis.ConnectionPool(
    host='192.xx.xx.2',            # Instance VIP address
    port=6379,                      # Instance port
    password='crs-xxxx:password',       # Default account password format: Instance ID:Password
    db=0,                           # Database number
    max_connections=200,            # Maximum number of connections
    socket_connect_timeout=2,       # Connection timeout (in seconds)
    socket_timeout=2,               # Read/write timeout (in seconds)
    decode_responses=True,          # Automatically decode as strings
    health_check_interval=30        # Health check interval (in seconds)
)

# Create a Redis client
r = redis.Redis(connection_pool=pool)

# Write data.
r.set('name', 'Distributed Cache')

# Read data.
value = r.get('name')
print(f'get name: {value}')

# Set a Key with an expiration time
r.setex('session:user123', 3600, 'session_data')

# Hash operations
r.hset('user:1001', mapping={'name': 'Zhang San', 'age': '28'})
user_info = r.hgetall('user:1001')
print(f'user info: {user_info}')

# List operations
r.lpush('queue:tasks', 'task1', 'task2', 'task3')
task = r.rpop('queue:tasks')
print(f'dequeued task: {task}')

# Pipeline batch operations
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

# Custom account password format: username@password
pool = redis.ConnectionPool(
    host='192.xx.xx.2',
    port=6379,
    password='myuser@MyPassword',   # Proxy mode custom account: the entire "username@password" string is passed as the password.
    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 Encryption Connection
pool = redis.ConnectionPool(
    host='192.xx.xx.2',
    port=6379,
    password='crs-xxxx:password',
    db=0,
    max_connections=200,
    socket_connect_timeout=2,
    socket_timeout=2,
    decode_responses=True,
    ssl=True,                       # Enable SSL.
    ssl_ca_certs='/path/to/ca.pem', # Path to the CA certificate
    ssl_cert_reqs='required'        # Verify the server certificate.
)

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():
    # Create an asynchronous connection pool
    pool = aioredis.ConnectionPool(
        host='192.xx.xx.2',
        port=6379,
        password='crs-xxxx:password',
        db=0,
        max_connections=200,
        socket_connect_timeout=2,
        socket_timeout=2,
        decode_responses=True
    )

    r = aioredis.Redis(connection_pool=pool)

    # Asynchronous write
    await r.set('async_key', 'async_value')

    # Asynchronous read
    value = await r.get('async_key')
    print(f'async get: {value}')

    # Asynchronous 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}')

    # Close the connection.
    await r.aclose()
    await pool.aclose()

asyncio.run(main())