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:
redis://
creates a TCP socket connection. https://www.iana.org/assignments/uri-schemes/prov/redisrediss://
creates a SSL wrapped TCP socket connection. https://www.iana.org/assignments/uri-schemes/prov/redissunix://
: creates a Unix Domain Socket connection.
[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")