Connection Examples#

Connecting to a default Redis instance, running locally.#

[2]:
import redis

connection = redis.Redis()
connection.ping()
[2]:
True

by default Redis return binary responses, to decode them use decode_responses=True#

[3]:
import redis

decoded_connection = redis.Redis(decode_responses=True)
decoded_connection.ping()
[3]:
True

Connecting to a redis instance, specifying a host and port with credentials.#

[4]:
import redis

user_connection = redis.Redis(host='localhost', port=6380, username='dvora', password='redis', decode_responses=True)
user_connection.ping()
[4]:
True

Connecting to a redis instance with username and password credential provider#

[ ]:
import redis

creds_provider = redis.UsernamePasswordCredentialProvider("username", "password")
user_connection = redis.Redis(host="localhost", port=6379, credential_provider=creds_provider)
user_connection.ping()

Connecting to a redis instance with standard credential provider#

[ ]:
from typing import Tuple
import redis

creds_map = {"user_1": "pass_1",
             "user_2": "pass_2"}

class UserMapCredentialProvider(redis.CredentialProvider):
    def __init__(self, username: str):
        self.username = username

    def get_credentials(self) -> Tuple[str, str]:
        return self.username, creds_map.get(self.username)

# Create a default connection to set the ACL user
default_connection = redis.Redis(host="localhost", port=6379)
default_connection.acl_setuser(
    "user_1",
    enabled=True,
    passwords=["+" + "pass_1"],
    keys="~*",
    commands=["+ping", "+command", "+info", "+select", "+flushdb"],
)

# Create a UserMapCredentialProvider instance for user_1
creds_provider = UserMapCredentialProvider("user_1")
# Initiate user connection with the credential provider
user_connection = redis.Redis(host="localhost", port=6379,
                              credential_provider=creds_provider)
user_connection.ping()

Connecting to a redis instance first with an initial credential set and then calling the credential provider#

[ ]:
from typing import Union
import redis

class InitCredsSetCredentialProvider(redis.CredentialProvider):
    def __init__(self, username, password):
        self.username = username
        self.password = password
        self.call_supplier = False

    def call_external_supplier(self) -> Union[Tuple[str], Tuple[str, str]]:
        # Call to an external credential supplier
        raise NotImplementedError

    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:
        if self.call_supplier:
            return self.call_external_supplier()
        # Use the init set only for the first time
        self.call_supplier = True
        return self.username, self.password

cred_provider = InitCredsSetCredentialProvider(username="init_user", password="init_pass")

Connecting to a redis instance with AWS Secrets Manager credential provider.#

[ ]:
import redis
import boto3
import json
import cachetools.func

sm_client = boto3.client('secretsmanager')

def sm_auth_provider(self, secret_id, version_id=None, version_stage='AWSCURRENT'):
    @cachetools.func.ttl_cache(maxsize=128, ttl=24 * 60 * 60) #24h
    def get_sm_user_credentials(secret_id, version_id, version_stage):
        secret = sm_client.get_secret_value(secret_id, version_id)
        return json.loads(secret['SecretString'])
    creds = get_sm_user_credentials(secret_id, version_id, version_stage)
    return creds['username'], creds['password']

secret_id = "EXAMPLE1-90ab-cdef-fedc-ba987SECRET1"
creds_provider = redis.CredentialProvider(supplier=sm_auth_provider, secret_id=secret_id)
user_connection = redis.Redis(host="localhost", port=6379, credential_provider=creds_provider)
user_connection.ping()

Connecting to a redis instance with ElastiCache IAM credential provider.#

[4]:
import redis
import boto3
import cachetools.func

ec_client = boto3.client('elasticache')

def iam_auth_provider(self, user, endpoint, port=6379, region="us-east-1"):
    @cachetools.func.ttl_cache(maxsize=128, ttl=15 * 60) # 15m
    def get_iam_auth_token(user, endpoint, port, region):
        return ec_client.generate_iam_auth_token(user, endpoint, port, region)
    iam_auth_token = get_iam_auth_token(endpoint, port, user, region)
    return iam_auth_token

username = "barshaul"
endpoint = "test-001.use1.cache.amazonaws.com"
creds_provider = redis.CredentialProvider(supplier=iam_auth_provider, user=username,
                                           endpoint=endpoint)
user_connection = redis.Redis(host=endpoint, port=6379, credential_provider=creds_provider)
user_connection.ping()
[4]:
True

Connecting to Redis instances by specifying a URL scheme.#

Parameters are passed to the following schems, as parameters to the url scheme.

Three URL schemes are supported:

[7]:
url_connection = redis.from_url("redis://localhost:6379?decode_responses=True&health_check_interval=2")

url_connection.ping()
[7]:
True

Connecting to a Sentinel instance#

[ ]:
from redis.sentinel import Sentinel
sentinel = Sentinel([('localhost', 26379)], socket_timeout=0.1)
sentinel.discover_master("redis-py-test")