Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-27 02:58
📄 __init__.py import sys if sys.version_info >= (3, 8): from importlib import metadata else: import importlib_metadata as metadata from redis.client import Redis, StrictRedis from redis.cluster import RedisCluster from redis.connection import ( BlockingConnectionPool, Connection, ConnectionPool, SSLConnection, UnixDomainSocketConnection, ) from redis.exceptions import ( AuthenticationError, AuthenticationWrongNumberOfArgsError, BusyLoadingError, ChildDeadlockedError, ConnectionError, DataError, InvalidResponse, PubSubError, ReadOnlyError, RedisError, ResponseError, TimeoutError, WatchError, ) from redis.sentinel import ( Sentinel, SentinelConnectionPool, SentinelManagedConnection, SentinelManagedSSLConnection, ) from redis.utils import from_url def int_or_str(value): try: return int(value) except ValueError: return value try: __version__ = metadata.version("redis") except metadata.PackageNotFoundError: __version__ = "99.99.99" VERSION = tuple(map(int_or_str, __version__.split("."))) __all__ = [ "AuthenticationError", "AuthenticationWrongNumberOfArgsError", "BlockingConnectionPool", "BusyLoadingError", "ChildDeadlockedError", "Connection", "ConnectionError", "ConnectionPool", "DataError", "from_url", "InvalidResponse", "PubSubError", "ReadOnlyError", "Redis", "RedisCluster", "RedisError", "ResponseError", "Sentinel", "SentinelConnectionPool", "SentinelManagedConnection", "SentinelManagedSSLConnection", "SSLConnection", "StrictRedis", "TimeoutError", "UnixDomainSocketConnection", "WatchError", ]	1.71 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 backoff.py import random from abc import ABC, abstractmethod class AbstractBackoff(ABC): """Backoff interface""" def reset(self): """ Reset internal state before an operation. `reset` is called once at the beginning of every call to `Retry.call_with_retry` """ pass @abstractmethod def compute(self, failures): """Compute backoff in seconds upon failure""" pass class ConstantBackoff(AbstractBackoff): """Constant backoff upon failure""" def __init__(self, backoff): """`backoff`: backoff time in seconds""" self._backoff = backoff def compute(self, failures): return self._backoff class NoBackoff(ConstantBackoff): """No backoff upon failure""" def __init__(self): super().__init__(0) class ExponentialBackoff(AbstractBackoff): """Exponential backoff upon failure""" def __init__(self, cap, base): """ `cap`: maximum backoff time in seconds `base`: base backoff time in seconds """ self._cap = cap self._base = base def compute(self, failures): return min(self._cap, self._base * 2 ** failures) class FullJitterBackoff(AbstractBackoff): """Full jitter backoff upon failure""" def __init__(self, cap, base): """ `cap`: maximum backoff time in seconds `base`: base backoff time in seconds """ self._cap = cap self._base = base def compute(self, failures): return random.uniform(0, min(self._cap, self._base * 2 ** failures)) class EqualJitterBackoff(AbstractBackoff): """Equal jitter backoff upon failure""" def __init__(self, cap, base): """ `cap`: maximum backoff time in seconds `base`: base backoff time in seconds """ self._cap = cap self._base = base def compute(self, failures): temp = min(self._cap, self._base * 2 ** failures) / 2 return temp + random.uniform(0, temp) class DecorrelatedJitterBackoff(AbstractBackoff): """Decorrelated jitter backoff upon failure""" def __init__(self, cap, base): """ `cap`: maximum backoff time in seconds `base`: base backoff time in seconds """ self._cap = cap self._base = base self._previous_backoff = 0 def reset(self): self._previous_backoff = 0 def compute(self, failures): max_backoff = max(self._base, self._previous_backoff * 3) temp = random.uniform(self._base, max_backoff) self._previous_backoff = min(self._cap, temp) return self._previous_backoff	2.61 KB	2025-05-25 12:26
📄 client.py import copy import datetime import re import threading import time import warnings from itertools import chain from redis.commands import ( CoreCommands, RedisModuleCommands, SentinelCommands, list_or_args, ) from redis.connection import ConnectionPool, SSLConnection, UnixDomainSocketConnection from redis.exceptions import ( ConnectionError, ExecAbortError, ModuleError, PubSubError, RedisError, ResponseError, TimeoutError, WatchError, ) from redis.lock import Lock from redis.utils import safe_str, str_if_bytes SYM_EMPTY = b"" EMPTY_RESPONSE = "EMPTY_RESPONSE" # some responses (ie. dump) are binary, and just meant to never be decoded NEVER_DECODE = "NEVER_DECODE" def timestamp_to_datetime(response): "Converts a unix timestamp to a Python datetime object" if not response: return None try: response = int(response) except ValueError: return None return datetime.datetime.fromtimestamp(response) def string_keys_to_dict(key_string, callback): return dict.fromkeys(key_string.split(), callback) class CaseInsensitiveDict(dict): "Case insensitive dict implementation. Assumes string keys only." def __init__(self, data): for k, v in data.items(): self[k.upper()] = v def __contains__(self, k): return super().__contains__(k.upper()) def __delitem__(self, k): super().__delitem__(k.upper()) def __getitem__(self, k): return super().__getitem__(k.upper()) def get(self, k, default=None): return super().get(k.upper(), default) def __setitem__(self, k, v): super().__setitem__(k.upper(), v) def update(self, data): data = CaseInsensitiveDict(data) super().update(data) def parse_debug_object(response): "Parse the results of Redis's DEBUG OBJECT command into a Python dict" # The 'type' of the object is the first item in the response, but isn't # prefixed with a name response = str_if_bytes(response) response = "type:" + response response = dict(kv.split(":") for kv in response.split()) # parse some expected int values from the string response # note: this cmd isn't spec'd so these may not appear in all redis versions int_fields = ("refcount", "serializedlength", "lru", "lru_seconds_idle") for field in int_fields: if field in response: response[field] = int(response[field]) return response def parse_object(response, infotype): """Parse the results of an OBJECT command""" if infotype in ("idletime", "refcount"): return int_or_none(response) return response def parse_info(response): """Parse the result of Redis's INFO command into a Python dict""" info = {} response = str_if_bytes(response) def get_value(value): if "," not in value or "=" not in value: try: if "." in value: return float(value) else: return int(value) except ValueError: return value else: sub_dict = {} for item in value.split(","): k, v = item.rsplit("=", 1) sub_dict[k] = get_value(v) return sub_dict for line in response.splitlines(): if line and not line.startswith("#"): if line.find(":") != -1: # Split, the info fields keys and values. # Note that the value may contain ':'. but the 'host:' # pseudo-command is the only case where the key contains ':' key, value = line.split(":", 1) if key == "cmdstat_host": key, value = line.rsplit(":", 1) if key == "module": # Hardcode a list for key 'modules' since there could be # multiple lines that started with 'module' info.setdefault("modules", []).append(get_value(value)) else: info[key] = get_value(value) else: # if the line isn't splittable, append it to the "__raw__" key info.setdefault("__raw__", []).append(line) return info def parse_memory_stats(response, kwargs): """Parse the results of MEMORY STATS""" stats = pairs_to_dict(response, decode_keys=True, decode_string_values=True) for key, value in stats.items(): if key.startswith("db."): stats[key] = pairs_to_dict( value, decode_keys=True, decode_string_values=True ) return stats SENTINEL_STATE_TYPES = { "can-failover-its-master": int, "config-epoch": int, "down-after-milliseconds": int, "failover-timeout": int, "info-refresh": int, "last-hello-message": int, "last-ok-ping-reply": int, "last-ping-reply": int, "last-ping-sent": int, "master-link-down-time": int, "master-port": int, "num-other-sentinels": int, "num-slaves": int, "o-down-time": int, "pending-commands": int, "parallel-syncs": int, "port": int, "quorum": int, "role-reported-time": int, "s-down-time": int, "slave-priority": int, "slave-repl-offset": int, "voted-leader-epoch": int, } def parse_sentinel_state(item): result = pairs_to_dict_typed(item, SENTINEL_STATE_TYPES) flags = set(result["flags"].split(",")) for name, flag in ( ("is_master", "master"), ("is_slave", "slave"), ("is_sdown", "s_down"), ("is_odown", "o_down"), ("is_sentinel", "sentinel"), ("is_disconnected", "disconnected"), ("is_master_down", "master_down"), ): result[name] = flag in flags return result def parse_sentinel_master(response): return parse_sentinel_state(map(str_if_bytes, response)) def parse_sentinel_masters(response): result = {} for item in response: state = parse_sentinel_state(map(str_if_bytes, item)) result[state["name"]] = state return result def parse_sentinel_slaves_and_sentinels(response): return [parse_sentinel_state(map(str_if_bytes, item)) for item in response] def parse_sentinel_get_master(response): return response and (response[0], int(response[1])) or None def pairs_to_dict(response, decode_keys=False, decode_string_values=False): """Create a dict given a list of key/value pairs""" if response is None: return {} if decode_keys or decode_string_values: # the iter form is faster, but I don't know how to make that work # with a str_if_bytes() map keys = response[::2] if decode_keys: keys = map(str_if_bytes, keys) values = response[1::2] if decode_string_values: values = map(str_if_bytes, values) return dict(zip(keys, values)) else: it = iter(response) return dict(zip(it, it)) def pairs_to_dict_typed(response, type_info): it = iter(response) result = {} for key, value in zip(it, it): if key in type_info: try: value = type_info[key](value) except Exception: # if for some reason the value can't be coerced, just use # the string value pass result[key] = value return result def zset_score_pairs(response, options): """ If ``withscores`` is specified in the options, return the response as a list of (value, score) pairs """ if not response or not options.get("withscores"): return response score_cast_func = options.get("score_cast_func", float) it = iter(response) return list(zip(it, map(score_cast_func, it))) def sort_return_tuples(response, *options): """ If ``groups`` is specified, return the response as a list of n-element tuples with n being the value found in options['groups'] """ if not response or not options.get("groups"): return response n = options["groups"] return list(zip([response[i::n] for i in range(n)])) def int_or_none(response): if response is None: return None return int(response) def parse_stream_list(response): if response is None: return None data = [] for r in response: if r is not None: data.append((r[0], pairs_to_dict(r[1]))) else: data.append((None, None)) return data def pairs_to_dict_with_str_keys(response): return pairs_to_dict(response, decode_keys=True) def parse_list_of_dicts(response): return list(map(pairs_to_dict_with_str_keys, response)) def parse_xclaim(response, options): if options.get("parse_justid", False): return response return parse_stream_list(response) def parse_xautoclaim(response, options): if options.get("parse_justid", False): return response[1] return parse_stream_list(response[1]) def parse_xinfo_stream(response, options): data = pairs_to_dict(response, decode_keys=True) if not options.get("full", False): first = data["first-entry"] if first is not None: data["first-entry"] = (first[0], pairs_to_dict(first[1])) last = data["last-entry"] if last is not None: data["last-entry"] = (last[0], pairs_to_dict(last[1])) else: data["entries"] = {_id: pairs_to_dict(entry) for _id, entry in data["entries"]} data["groups"] = [ pairs_to_dict(group, decode_keys=True) for group in data["groups"] ] return data def parse_xread(response): if response is None: return [] return [[r[0], parse_stream_list(r[1])] for r in response] def parse_xpending(response, options): if options.get("parse_detail", False): return parse_xpending_range(response) consumers = [{"name": n, "pending": int(p)} for n, p in response[3] or []] return { "pending": response[0], "min": response[1], "max": response[2], "consumers": consumers, } def parse_xpending_range(response): k = ("message_id", "consumer", "time_since_delivered", "times_delivered") return [dict(zip(k, r)) for r in response] def float_or_none(response): if response is None: return None return float(response) def bool_ok(response): return str_if_bytes(response) == "OK" def parse_zadd(response, options): if response is None: return None if options.get("as_score"): return float(response) return int(response) def parse_client_list(response, options): clients = [] for c in str_if_bytes(response).splitlines(): # Values might contain '=' clients.append(dict(pair.split("=", 1) for pair in c.split(" "))) return clients def parse_config_get(response, options): response = [str_if_bytes(i) if i is not None else None for i in response] return response and pairs_to_dict(response) or {} def parse_scan(response, options): cursor, r = response return int(cursor), r def parse_hscan(response, options): cursor, r = response return int(cursor), r and pairs_to_dict(r) or {} def parse_zscan(response, options): score_cast_func = options.get("score_cast_func", float) cursor, r = response it = iter(r) return int(cursor), list(zip(it, map(score_cast_func, it))) def parse_zmscore(response, options): # zmscore: list of scores (double precision floating point number) or nil return [float(score) if score is not None else None for score in response] def parse_slowlog_get(response, options): space = " " if options.get("decode_responses", False) else b" " def parse_item(item): result = { "id": item[0], "start_time": int(item[1]), "duration": int(item[2]), } # Redis Enterprise injects another entry at index [3], which has # the complexity info (i.e. the value N in case the command has # an O(N) complexity) instead of the command. if isinstance(item[3], list): result["command"] = space.join(item[3]) else: result["complexity"] = item[3] result["command"] = space.join(item[4]) return result return [parse_item(item) for item in response] def parse_stralgo(response, options): """ Parse the response from `STRALGO` command. Without modifiers the returned value is string. When LEN is given the command returns the length of the result (i.e integer). When IDX is given the command returns a dictionary with the LCS length and all the ranges in both the strings, start and end offset for each string, where there are matches. When WITHMATCHLEN is given, each array representing a match will also have the length of the match at the beginning of the array. """ if options.get("len", False): return int(response) if options.get("idx", False): if options.get("withmatchlen", False): matches = [ [(int(match[-1]))] + list(map(tuple, match[:-1])) for match in response[1] ] else: matches = [list(map(tuple, match)) for match in response[1]] return { str_if_bytes(response[0]): matches, str_if_bytes(response[2]): int(response[3]), } return str_if_bytes(response) def parse_cluster_info(response, options): response = str_if_bytes(response) return dict(line.split(":") for line in response.splitlines() if line) def _parse_node_line(line): line_items = line.split(" ") node_id, addr, flags, master_id, ping, pong, epoch, connected = line.split(" ")[:8] addr = addr.split("@")[0] slots = [sl.split("-") for sl in line_items[8:]] node_dict = { "node_id": node_id, "flags": flags, "master_id": master_id, "last_ping_sent": ping, "last_pong_rcvd": pong, "epoch": epoch, "slots": slots, "connected": True if connected == "connected" else False, } return addr, node_dict def parse_cluster_nodes(response, options): """ @see: https://redis.io/commands/cluster-nodes # string @see: https://redis.io/commands/cluster-replicas # list of string """ if isinstance(response, str): response = response.splitlines() return dict(_parse_node_line(str_if_bytes(node)) for node in response) def parse_geosearch_generic(response, options): """ Parse the response of 'GEOSEARCH', GEORADIUS' and 'GEORADIUSBYMEMBER' commands according to 'withdist', 'withhash' and 'withcoord' labels. """ if options["store"] or options["store_dist"]: # `store` and `store_dist` cant be combined # with other command arguments. # relevant to 'GEORADIUS' and 'GEORADIUSBYMEMBER' return response if type(response) != list: response_list = [response] else: response_list = response if not options["withdist"] and not options["withcoord"] and not options["withhash"]: # just a bunch of places return response_list cast = { "withdist": float, "withcoord": lambda ll: (float(ll[0]), float(ll[1])), "withhash": int, } # zip all output results with each casting function to get # the properly native Python value. f = [lambda x: x] f += [cast[o] for o in ["withdist", "withhash", "withcoord"] if options[o]] return [list(map(lambda fv: fv[0](fv[1]), zip(f, r))) for r in response_list] def parse_command(response, options): commands = {} for command in response: cmd_dict = {} cmd_name = str_if_bytes(command[0]) cmd_dict["name"] = cmd_name cmd_dict["arity"] = int(command[1]) cmd_dict["flags"] = [str_if_bytes(flag) for flag in command[2]] cmd_dict["first_key_pos"] = command[3] cmd_dict["last_key_pos"] = command[4] cmd_dict["step_count"] = command[5] commands[cmd_name] = cmd_dict return commands def parse_pubsub_numsub(response, options): return list(zip(response[0::2], response[1::2])) def parse_client_kill(response, options): if isinstance(response, int): return response return str_if_bytes(response) == "OK" def parse_acl_getuser(response, options): if response is None: return None data = pairs_to_dict(response, decode_keys=True) # convert everything but user-defined data in 'keys' to native strings data["flags"] = list(map(str_if_bytes, data["flags"])) data["passwords"] = list(map(str_if_bytes, data["passwords"])) data["commands"] = str_if_bytes(data["commands"]) # split 'commands' into separate 'categories' and 'commands' lists commands, categories = [], [] for command in data["commands"].split(" "): if "@" in command: categories.append(command) else: commands.append(command) data["commands"] = commands data["categories"] = categories data["enabled"] = "on" in data["flags"] return data def parse_acl_log(response, options): if response is None: return None if isinstance(response, list): data = [] for log in response: log_data = pairs_to_dict(log, True, True) client_info = log_data.get("client-info", "") log_data["client-info"] = parse_client_info(client_info) # float() is lossy comparing to the "double" in C log_data["age-seconds"] = float(log_data["age-seconds"]) data.append(log_data) else: data = bool_ok(response) return data def parse_client_info(value): """ Parsing client-info in ACL Log in following format. "key1=value1 key2=value2 key3=value3" """ client_info = {} infos = str_if_bytes(value).split(" ") for info in infos: key, value = info.split("=") client_info[key] = value # Those fields are defined as int in networking.c for int_key in { "id", "age", "idle", "db", "sub", "psub", "multi", "qbuf", "qbuf-free", "obl", "argv-mem", "oll", "omem", "tot-mem", }: client_info[int_key] = int(client_info[int_key]) return client_info def parse_module_result(response): if isinstance(response, ModuleError): raise response return True def parse_set_result(response, options): """ Handle SET result since GET argument is available since Redis 6.2. Parsing SET result into: - BOOL - String when GET argument is used """ if options.get("get"): # Redis will return a getCommand result. # See `setGenericCommand` in t_string.c return response return response and str_if_bytes(response) == "OK" class Redis(RedisModuleCommands, CoreCommands, SentinelCommands): """ Implementation of the Redis protocol. This abstract class provides a Python interface to all Redis commands and an implementation of the Redis protocol. Pipelines derive from this, implementing how the commands are sent and received to the Redis server. Based on configuration, an instance will either use a ConnectionPool, or Connection object to talk to redis. """ RESPONSE_CALLBACKS = { string_keys_to_dict( "AUTH COPY EXPIRE EXPIREAT PEXPIRE PEXPIREAT " "HEXISTS HMSET MOVE MSETNX PERSIST " "PSETEX RENAMENX SISMEMBER SMOVE SETEX SETNX", bool, ), string_keys_to_dict( "BITCOUNT BITPOS DECRBY DEL EXISTS GEOADD GETBIT HDEL HLEN " "HSTRLEN INCRBY LINSERT LLEN LPUSHX PFADD PFCOUNT RPUSHX SADD " "SCARD SDIFFSTORE SETBIT SETRANGE SINTERSTORE SREM STRLEN " "SUNIONSTORE UNLINK XACK XDEL XLEN XTRIM ZCARD ZLEXCOUNT ZREM " "ZREMRANGEBYLEX ZREMRANGEBYRANK ZREMRANGEBYSCORE", int, ), string_keys_to_dict("INCRBYFLOAT HINCRBYFLOAT", float), string_keys_to_dict( # these return OK, or int if redis-server is >=1.3.4 "LPUSH RPUSH", lambda r: isinstance(r, int) and r or str_if_bytes(r) == "OK", ), string_keys_to_dict("SORT", sort_return_tuples), string_keys_to_dict("ZSCORE ZINCRBY GEODIST", float_or_none), string_keys_to_dict( "FLUSHALL FLUSHDB LSET LTRIM MSET PFMERGE READONLY READWRITE " "RENAME SAVE SELECT SHUTDOWN SLAVEOF SWAPDB WATCH UNWATCH ", bool_ok, ), string_keys_to_dict("BLPOP BRPOP", lambda r: r and tuple(r) or None), string_keys_to_dict( "SDIFF SINTER SMEMBERS SUNION", lambda r: r and set(r) or set() ), string_keys_to_dict( "ZPOPMAX ZPOPMIN ZINTER ZDIFF ZUNION ZRANGE ZRANGEBYSCORE " "ZREVRANGE ZREVRANGEBYSCORE", zset_score_pairs, ), string_keys_to_dict( "BZPOPMIN BZPOPMAX", lambda r: r and (r[0], r[1], float(r[2])) or None ), string_keys_to_dict("ZRANK ZREVRANK", int_or_none), string_keys_to_dict("XREVRANGE XRANGE", parse_stream_list), string_keys_to_dict("XREAD XREADGROUP", parse_xread), string_keys_to_dict("BGREWRITEAOF BGSAVE", lambda r: True), "ACL CAT": lambda r: list(map(str_if_bytes, r)), "ACL DELUSER": int, "ACL GENPASS": str_if_bytes, "ACL GETUSER": parse_acl_getuser, "ACL HELP": lambda r: list(map(str_if_bytes, r)), "ACL LIST": lambda r: list(map(str_if_bytes, r)), "ACL LOAD": bool_ok, "ACL LOG": parse_acl_log, "ACL SAVE": bool_ok, "ACL SETUSER": bool_ok, "ACL USERS": lambda r: list(map(str_if_bytes, r)), "ACL WHOAMI": str_if_bytes, "CLIENT GETNAME": str_if_bytes, "CLIENT ID": int, "CLIENT KILL": parse_client_kill, "CLIENT LIST": parse_client_list, "CLIENT INFO": parse_client_info, "CLIENT SETNAME": bool_ok, "CLIENT UNBLOCK": lambda r: r and int(r) == 1 or False, "CLIENT PAUSE": bool_ok, "CLIENT GETREDIR": int, "CLIENT TRACKINGINFO": lambda r: list(map(str_if_bytes, r)), "CLUSTER ADDSLOTS": bool_ok, "CLUSTER COUNT-FAILURE-REPORTS": lambda x: int(x), "CLUSTER COUNTKEYSINSLOT": lambda x: int(x), "CLUSTER DELSLOTS": bool_ok, "CLUSTER FAILOVER": bool_ok, "CLUSTER FORGET": bool_ok, "CLUSTER INFO": parse_cluster_info, "CLUSTER KEYSLOT": lambda x: int(x), "CLUSTER MEET": bool_ok, "CLUSTER NODES": parse_cluster_nodes, "CLUSTER REPLICATE": bool_ok, "CLUSTER RESET": bool_ok, "CLUSTER SAVECONFIG": bool_ok, "CLUSTER SET-CONFIG-EPOCH": bool_ok, "CLUSTER SETSLOT": bool_ok, "CLUSTER SLAVES": parse_cluster_nodes, "CLUSTER REPLICAS": parse_cluster_nodes, "COMMAND": parse_command, "COMMAND COUNT": int, "COMMAND GETKEYS": lambda r: list(map(str_if_bytes, r)), "CONFIG GET": parse_config_get, "CONFIG RESETSTAT": bool_ok, "CONFIG SET": bool_ok, "DEBUG OBJECT": parse_debug_object, "GEOHASH": lambda r: list(map(str_if_bytes, r)), "GEOPOS": lambda r: list( map(lambda ll: (float(ll[0]), float(ll[1])) if ll is not None else None, r) ), "GEOSEARCH": parse_geosearch_generic, "GEORADIUS": parse_geosearch_generic, "GEORADIUSBYMEMBER": parse_geosearch_generic, "HGETALL": lambda r: r and pairs_to_dict(r) or {}, "HSCAN": parse_hscan, "INFO": parse_info, "LASTSAVE": timestamp_to_datetime, "MEMORY PURGE": bool_ok, "MEMORY STATS": parse_memory_stats, "MEMORY USAGE": int_or_none, "MODULE LOAD": parse_module_result, "MODULE UNLOAD": parse_module_result, "MODULE LIST": lambda r: [pairs_to_dict(m) for m in r], "OBJECT": parse_object, "PING": lambda r: str_if_bytes(r) == "PONG", "QUIT": bool_ok, "STRALGO": parse_stralgo, "PUBSUB NUMSUB": parse_pubsub_numsub, "RANDOMKEY": lambda r: r and r or None, "RESET": str_if_bytes, "SCAN": parse_scan, "SCRIPT EXISTS": lambda r: list(map(bool, r)), "SCRIPT FLUSH": bool_ok, "SCRIPT KILL": bool_ok, "SCRIPT LOAD": str_if_bytes, "SENTINEL CKQUORUM": bool_ok, "SENTINEL FAILOVER": bool_ok, "SENTINEL FLUSHCONFIG": bool_ok, "SENTINEL GET-MASTER-ADDR-BY-NAME": parse_sentinel_get_master, "SENTINEL MASTER": parse_sentinel_master, "SENTINEL MASTERS": parse_sentinel_masters, "SENTINEL MONITOR": bool_ok, "SENTINEL RESET": bool_ok, "SENTINEL REMOVE": bool_ok, "SENTINEL SENTINELS": parse_sentinel_slaves_and_sentinels, "SENTINEL SET": bool_ok, "SENTINEL SLAVES": parse_sentinel_slaves_and_sentinels, "SET": parse_set_result, "SLOWLOG GET": parse_slowlog_get, "SLOWLOG LEN": int, "SLOWLOG RESET": bool_ok, "SSCAN": parse_scan, "TIME": lambda x: (int(x[0]), int(x[1])), "XCLAIM": parse_xclaim, "XAUTOCLAIM": parse_xautoclaim, "XGROUP CREATE": bool_ok, "XGROUP DELCONSUMER": int, "XGROUP DESTROY": bool, "XGROUP SETID": bool_ok, "XINFO CONSUMERS": parse_list_of_dicts, "XINFO GROUPS": parse_list_of_dicts, "XINFO STREAM": parse_xinfo_stream, "XPENDING": parse_xpending, "ZADD": parse_zadd, "ZSCAN": parse_zscan, "ZMSCORE": parse_zmscore, } @classmethod def from_url(cls, url, kwargs): """ Return a Redis client object configured from the given URL For example:: redis://[[username]:[password]]@localhost:6379/0 rediss://[[username]:[password]]@localhost:6379/0 unix://[[username]:[password]]@/path/to/socket.sock?db=0 Three URL schemes are supported: - `redis://` creates a TCP socket connection. See more at: <https://www.iana.org/assignments/uri-schemes/prov/redis> - `rediss://` creates a SSL wrapped TCP socket connection. See more at: <https://www.iana.org/assignments/uri-schemes/prov/rediss> - ``unix://``: creates a Unix Domain Socket connection. The username, password, hostname, path and all querystring values are passed through urllib.parse.unquote in order to replace any percent-encoded values with their corresponding characters. There are several ways to specify a database number. The first value found will be used: 1. A ``db`` querystring option, e.g. redis://localhost?db=0 2. If using the redis:// or rediss:// schemes, the path argument of the url, e.g. redis://localhost/0 3. A ``db`` keyword argument to this function. If none of these options are specified, the default db=0 is used. All querystring options are cast to their appropriate Python types. Boolean arguments can be specified with string values "True"/"False" or "Yes"/"No". Values that cannot be properly cast cause a ``ValueError`` to be raised. Once parsed, the querystring arguments and keyword arguments are passed to the ``ConnectionPool``'s class initializer. In the case of conflicting arguments, querystring arguments always win. """ connection_pool = ConnectionPool.from_url(url, kwargs) return cls(connection_pool=connection_pool) def __init__( self, host="localhost", port=6379, db=0, password=None, socket_timeout=None, socket_connect_timeout=None, socket_keepalive=None, socket_keepalive_options=None, connection_pool=None, unix_socket_path=None, encoding="utf-8", encoding_errors="strict", charset=None, errors=None, decode_responses=False, retry_on_timeout=False, retry_on_error=[], ssl=False, ssl_keyfile=None, ssl_certfile=None, ssl_cert_reqs="required", ssl_ca_certs=None, ssl_ca_path=None, ssl_check_hostname=False, ssl_password=None, ssl_validate_ocsp=False, ssl_validate_ocsp_stapled=False, ssl_ocsp_context=None, ssl_ocsp_expected_cert=None, max_connections=None, single_connection_client=False, health_check_interval=0, client_name=None, username=None, retry=None, redis_connect_func=None, ): """ Initialize a new Redis client. To specify a retry policy for specific errors, first set `retry_on_error` to a list of the error/s to retry on, then set `retry` to a valid `Retry` object. To retry on TimeoutError, `retry_on_timeout` can also be set to `True`. """ if not connection_pool: if charset is not None: warnings.warn( DeprecationWarning( '"charset" is deprecated. Use "encoding" instead' ) ) encoding = charset if errors is not None: warnings.warn( DeprecationWarning( '"errors" is deprecated. Use "encoding_errors" instead' ) ) encoding_errors = errors if retry_on_timeout is True: retry_on_error.append(TimeoutError) kwargs = { "db": db, "username": username, "password": password, "socket_timeout": socket_timeout, "encoding": encoding, "encoding_errors": encoding_errors, "decode_responses": decode_responses, "retry_on_error": retry_on_error, "retry": copy.deepcopy(retry), "max_connections": max_connections, "health_check_interval": health_check_interval, "client_name": client_name, "redis_connect_func": redis_connect_func, } # based on input, setup appropriate connection args if unix_socket_path is not None: kwargs.update( { "path": unix_socket_path, "connection_class": UnixDomainSocketConnection, } ) else: # TCP specific options kwargs.update( { "host": host, "port": port, "socket_connect_timeout": socket_connect_timeout, "socket_keepalive": socket_keepalive, "socket_keepalive_options": socket_keepalive_options, } ) if ssl: kwargs.update( { "connection_class": SSLConnection, "ssl_keyfile": ssl_keyfile, "ssl_certfile": ssl_certfile, "ssl_cert_reqs": ssl_cert_reqs, "ssl_ca_certs": ssl_ca_certs, "ssl_check_hostname": ssl_check_hostname, "ssl_password": ssl_password, "ssl_ca_path": ssl_ca_path, "ssl_validate_ocsp_stapled": ssl_validate_ocsp_stapled, "ssl_validate_ocsp": ssl_validate_ocsp, "ssl_ocsp_context": ssl_ocsp_context, "ssl_ocsp_expected_cert": ssl_ocsp_expected_cert, } ) connection_pool = ConnectionPool(kwargs) self.connection_pool = connection_pool self.connection = None if single_connection_client: self.connection = self.connection_pool.get_connection("_") self.response_callbacks = CaseInsensitiveDict(self.__class__.RESPONSE_CALLBACKS) def __repr__(self): return f"{type(self).__name__}<{repr(self.connection_pool)}>" def get_encoder(self): """Get the connection pool's encoder""" return self.connection_pool.get_encoder() def get_connection_kwargs(self): """Get the connection's key-word arguments""" return self.connection_pool.connection_kwargs def set_response_callback(self, command, callback): """Set a custom Response Callback""" self.response_callbacks[command] = callback def load_external_module( self, funcname, func, ): """ This function can be used to add externally defined redis modules, and their namespaces to the redis client. funcname - A string containing the name of the function to create func - The function, being added to this class. ex: Assume that one has a custom redis module named foomod that creates command named 'foo.dothing' and 'foo.anotherthing' in redis. To load function functions into this namespace: from redis import Redis from foomodule import F r = Redis() r.load_external_module("foo", F) r.foo().dothing('your', 'arguments') For a concrete example see the reimport of the redisjson module in tests/test_connection.py::test_loading_external_modules """ setattr(self, funcname, func) def pipeline(self, transaction=True, shard_hint=None): """ Return a new pipeline object that can queue multiple commands for later execution. ``transaction`` indicates whether all commands should be executed atomically. Apart from making a group of operations atomic, pipelines are useful for reducing the back-and-forth overhead between the client and server. """ return Pipeline( self.connection_pool, self.response_callbacks, transaction, shard_hint ) def transaction(self, func, watches, kwargs): """ Convenience method for executing the callable `func` as a transaction while watching all keys specified in `watches`. The 'func' callable should expect a single argument which is a Pipeline object. """ shard_hint = kwargs.pop("shard_hint", None) value_from_callable = kwargs.pop("value_from_callable", False) watch_delay = kwargs.pop("watch_delay", None) with self.pipeline(True, shard_hint) as pipe: while True: try: if watches: pipe.watch(watches) func_value = func(pipe) exec_value = pipe.execute() return func_value if value_from_callable else exec_value except WatchError: if watch_delay is not None and watch_delay > 0: time.sleep(watch_delay) continue def lock( self, name, timeout=None, sleep=0.1, blocking_timeout=None, lock_class=None, thread_local=True, ): """ Return a new Lock object using key ``name`` that mimics the behavior of threading.Lock. If specified, ``timeout`` indicates a maximum life for the lock. By default, it will remain locked until release() is called. ``sleep`` indicates the amount of time to sleep per loop iteration when the lock is in blocking mode and another client is currently holding the lock. ``blocking_timeout`` indicates the maximum amount of time in seconds to spend trying to acquire the lock. A value of ``None`` indicates continue trying forever. ``blocking_timeout`` can be specified as a float or integer, both representing the number of seconds to wait. ``lock_class`` forces the specified lock implementation. ``thread_local`` indicates whether the lock token is placed in thread-local storage. By default, the token is placed in thread local storage so that a thread only sees its token, not a token set by another thread. Consider the following timeline: time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds. thread-1 sets the token to "abc" time: 1, thread-2 blocks trying to acquire `my-lock` using the Lock instance. time: 5, thread-1 has not yet completed. redis expires the lock key. time: 5, thread-2 acquired `my-lock` now that it's available. thread-2 sets the token to "xyz" time: 6, thread-1 finishes its work and calls release(). if the token is not stored in thread local storage, then thread-1 would see the token value as "xyz" and would be able to successfully release the thread-2's lock. In some use cases it's necessary to disable thread local storage. For example, if you have code where one thread acquires a lock and passes that lock instance to a worker thread to release later. If thread local storage isn't disabled in this case, the worker thread won't see the token set by the thread that acquired the lock. Our assumption is that these cases aren't common and as such default to using thread local storage.""" if lock_class is None: lock_class = Lock return lock_class( self, name, timeout=timeout, sleep=sleep, blocking_timeout=blocking_timeout, thread_local=thread_local, ) def pubsub(self, kwargs): """ Return a Publish/Subscribe object. With this object, you can subscribe to channels and listen for messages that get published to them. """ return PubSub(self.connection_pool, kwargs) def monitor(self): return Monitor(self.connection_pool) def client(self): return self.__class__( connection_pool=self.connection_pool, single_connection_client=True ) def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.close() def __del__(self): self.close() def close(self): conn = self.connection if conn: self.connection = None self.connection_pool.release(conn) def _send_command_parse_response(self, conn, command_name, args, options): """ Send a command and parse the response """ conn.send_command(args) return self.parse_response(conn, command_name, *options) def _disconnect_raise(self, conn, error): """ Close the connection and raise an exception if retry_on_error is not set or the error is not one of the specified error types """ conn.disconnect() if ( conn.retry_on_error is None or isinstance(error, tuple(conn.retry_on_error)) is False ): raise error # COMMAND EXECUTION AND PROTOCOL PARSING def execute_command(self, args, options): """Execute a command and return a parsed response""" pool = self.connection_pool command_name = args[0] conn = self.connection or pool.get_connection(command_name, options) try: return conn.retry.call_with_retry( lambda: self._send_command_parse_response( conn, command_name, args, options ), lambda error: self._disconnect_raise(conn, error), ) finally: if not self.connection: pool.release(conn) def parse_response(self, connection, command_name, options): """Parses a response from the Redis server""" try: if NEVER_DECODE in options: response = connection.read_response(disable_decoding=True) else: response = connection.read_response() except ResponseError: if EMPTY_RESPONSE in options: return options[EMPTY_RESPONSE] raise if command_name in self.response_callbacks: return self.response_callbacks[command_name](response, options) return response StrictRedis = Redis class Monitor: """ Monitor is useful for handling the MONITOR command to the redis server. next_command() method returns one command from monitor listen() method yields commands from monitor. """ monitor_re = re.compile(r"\[(\d+) (.)\] (.)") command_re = re.compile(r'"(.?)(?<!\\)"') def __init__(self, connection_pool): self.connection_pool = connection_pool self.connection = self.connection_pool.get_connection("MONITOR") def __enter__(self): self.connection.send_command("MONITOR") # check that monitor returns 'OK', but don't return it to user response = self.connection.read_response() if not bool_ok(response): raise RedisError(f"MONITOR failed: {response}") return self def __exit__(self, args): self.connection.disconnect() self.connection_pool.release(self.connection) def next_command(self): """Parse the response from a monitor command""" response = self.connection.read_response() if isinstance(response, bytes): response = self.connection.encoder.decode(response, force=True) command_time, command_data = response.split(" ", 1) m = self.monitor_re.match(command_data) db_id, client_info, command = m.groups() command = " ".join(self.command_re.findall(command)) # Redis escapes double quotes because each piece of the command # string is surrounded by double quotes. We don't have that # requirement so remove the escaping and leave the quote. command = command.replace('\\"', '"') if client_info == "lua": client_address = "lua" client_port = "" client_type = "lua" elif client_info.startswith("unix"): client_address = "unix" client_port = client_info[5:] client_type = "unix" else: # use rsplit as ipv6 addresses contain colons client_address, client_port = client_info.rsplit(":", 1) client_type = "tcp" return { "time": float(command_time), "db": int(db_id), "client_address": client_address, "client_port": client_port, "client_type": client_type, "command": command, } def listen(self): """Listen for commands coming to the server.""" while True: yield self.next_command() class PubSub: """ PubSub provides publish, subscribe and listen support to Redis channels. After subscribing to one or more channels, the listen() method will block until a message arrives on one of the subscribed channels. That message will be returned and it's safe to start listening again. """ PUBLISH_MESSAGE_TYPES = ("message", "pmessage") UNSUBSCRIBE_MESSAGE_TYPES = ("unsubscribe", "punsubscribe") HEALTH_CHECK_MESSAGE = "redis-py-health-check" def __init__( self, connection_pool, shard_hint=None, ignore_subscribe_messages=False, encoder=None, ): self.connection_pool = connection_pool self.shard_hint = shard_hint self.ignore_subscribe_messages = ignore_subscribe_messages self.connection = None self.subscribed_event = threading.Event() # we need to know the encoding options for this connection in order # to lookup channel and pattern names for callback handlers. self.encoder = encoder if self.encoder is None: self.encoder = self.connection_pool.get_encoder() self.health_check_response_b = self.encoder.encode(self.HEALTH_CHECK_MESSAGE) if self.encoder.decode_responses: self.health_check_response = ["pong", self.HEALTH_CHECK_MESSAGE] else: self.health_check_response = [b"pong", self.health_check_response_b] self.reset() def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.reset() def __del__(self): try: # if this object went out of scope prior to shutting down # subscriptions, close the connection manually before # returning it to the connection pool self.reset() except Exception: pass def reset(self): if self.connection: self.connection.disconnect() self.connection.clear_connect_callbacks() self.connection_pool.release(self.connection) self.connection = None self.channels = {} self.health_check_response_counter = 0 self.pending_unsubscribe_channels = set() self.patterns = {} self.pending_unsubscribe_patterns = set() self.subscribed_event.clear() def close(self): self.reset() def on_connect(self, connection): "Re-subscribe to any channels and patterns previously subscribed to" # NOTE: for python3, we can't pass bytestrings as keyword arguments # so we need to decode channel/pattern names back to unicode strings # before passing them to [p]subscribe. self.pending_unsubscribe_channels.clear() self.pending_unsubscribe_patterns.clear() if self.channels: channels = {} for k, v in self.channels.items(): channels[self.encoder.decode(k, force=True)] = v self.subscribe(channels) if self.patterns: patterns = {} for k, v in self.patterns.items(): patterns[self.encoder.decode(k, force=True)] = v self.psubscribe(patterns) @property def subscribed(self): """Indicates if there are subscriptions to any channels or patterns""" return self.subscribed_event.is_set() def execute_command(self, args): """Execute a publish/subscribe command""" # NOTE: don't parse the response in this function -- it could pull a # legitimate message off the stack if the connection is already # subscribed to one or more channels if self.connection is None: self.connection = self.connection_pool.get_connection( "pubsub", self.shard_hint ) # register a callback that re-subscribes to any channels we # were listening to when we were disconnected self.connection.register_connect_callback(self.on_connect) connection = self.connection kwargs = {"check_health": not self.subscribed} if not self.subscribed: self.clean_health_check_responses() self._execute(connection, connection.send_command, args, kwargs) def clean_health_check_responses(self): """ If any health check responses are present, clean them """ ttl = 10 conn = self.connection while self.health_check_response_counter > 0 and ttl > 0: if self._execute(conn, conn.can_read, timeout=conn.socket_timeout): response = self._execute(conn, conn.read_response) if self.is_health_check_response(response): self.health_check_response_counter -= 1 else: raise PubSubError( "A non health check response was cleaned by " "execute_command: {0}".format(response) ) ttl -= 1 def _disconnect_raise_connect(self, conn, error): """ Close the connection and raise an exception if retry_on_timeout is not set or the error is not a TimeoutError. Otherwise, try to reconnect """ conn.disconnect() if not (conn.retry_on_timeout and isinstance(error, TimeoutError)): raise error conn.connect() def _execute(self, conn, command, args, *kwargs): """ Connect manually upon disconnection. If the Redis server is down, this will fail and raise a ConnectionError as desired. After reconnection, the ``on_connect`` callback should have been called by the # connection to resubscribe us to any channels and patterns we were previously listening to """ return conn.retry.call_with_retry( lambda: command(args, *kwargs), lambda error: self._disconnect_raise_connect(conn, error), ) def parse_response(self, block=True, timeout=0): """Parse the response from a publish/subscribe command""" conn = self.connection if conn is None: raise RuntimeError( "pubsub connection not set: " "did you forget to call subscribe() or psubscribe()?" ) self.check_health() if not block and not self._execute(conn, conn.can_read, timeout=timeout): return None response = self._execute(conn, conn.read_response) if self.is_health_check_response(response): # ignore the health check message as user might not expect it self.health_check_response_counter -= 1 return None return response def is_health_check_response(self, response): """ Check if the response is a health check response. If there are no subscriptions redis responds to PING command with a bulk response, instead of a multi-bulk with "pong" and the response. """ return response in [ self.health_check_response, # If there was a subscription self.health_check_response_b, # If there wasn't ] def check_health(self): conn = self.connection if conn is None: raise RuntimeError( "pubsub connection not set: " "did you forget to call subscribe() or psubscribe()?" ) if conn.health_check_interval and time.time() > conn.next_health_check: conn.send_command("PING", self.HEALTH_CHECK_MESSAGE, check_health=False) self.health_check_response_counter += 1 def _normalize_keys(self, data): """ normalize channel/pattern names to be either bytes or strings based on whether responses are automatically decoded. this saves us from coercing the value for each message coming in. """ encode = self.encoder.encode decode = self.encoder.decode return {decode(encode(k)): v for k, v in data.items()} def psubscribe(self, args, *kwargs): """ Subscribe to channel patterns. Patterns supplied as keyword arguments expect a pattern name as the key and a callable as the value. A pattern's callable will be invoked automatically when a message is received on that pattern rather than producing a message via ``listen()``. """ if args: args = list_or_args(args[0], args[1:]) new_patterns = dict.fromkeys(args) new_patterns.update(kwargs) ret_val = self.execute_command("PSUBSCRIBE", new_patterns.keys()) # update the patterns dict AFTER we send the command. we don't want to # subscribe twice to these patterns, once for the command and again # for the reconnection. new_patterns = self._normalize_keys(new_patterns) self.patterns.update(new_patterns) if not self.subscribed: # Set the subscribed_event flag to True self.subscribed_event.set() # Clear the health check counter self.health_check_response_counter = 0 self.pending_unsubscribe_patterns.difference_update(new_patterns) return ret_val def punsubscribe(self, args): """ Unsubscribe from the supplied patterns. If empty, unsubscribe from all patterns. """ if args: args = list_or_args(args[0], args[1:]) patterns = self._normalize_keys(dict.fromkeys(args)) else: patterns = self.patterns self.pending_unsubscribe_patterns.update(patterns) return self.execute_command("PUNSUBSCRIBE", args) def subscribe(self, args, kwargs): """ Subscribe to channels. Channels supplied as keyword arguments expect a channel name as the key and a callable as the value. A channel's callable will be invoked automatically when a message is received on that channel rather than producing a message via ``listen()`` or ``get_message()``. """ if args: args = list_or_args(args[0], args[1:]) new_channels = dict.fromkeys(args) new_channels.update(kwargs) ret_val = self.execute_command("SUBSCRIBE", new_channels.keys()) # update the channels dict AFTER we send the command. we don't want to # subscribe twice to these channels, once for the command and again # for the reconnection. new_channels = self._normalize_keys(new_channels) self.channels.update(new_channels) if not self.subscribed: # Set the subscribed_event flag to True self.subscribed_event.set() # Clear the health check counter self.health_check_response_counter = 0 self.pending_unsubscribe_channels.difference_update(new_channels) return ret_val def unsubscribe(self, args): """ Unsubscribe from the supplied channels. If empty, unsubscribe from all channels """ if args: args = list_or_args(args[0], args[1:]) channels = self._normalize_keys(dict.fromkeys(args)) else: channels = self.channels self.pending_unsubscribe_channels.update(channels) return self.execute_command("UNSUBSCRIBE", args) def listen(self): "Listen for messages on channels this client has been subscribed to" while self.subscribed: response = self.handle_message(self.parse_response(block=True)) if response is not None: yield response def get_message(self, ignore_subscribe_messages=False, timeout=0): """ Get the next message if one is available, otherwise None. If timeout is specified, the system will wait for `timeout` seconds before returning. Timeout should be specified as a floating point number. """ if not self.subscribed: # Wait for subscription start_time = time.time() if self.subscribed_event.wait(timeout) is True: # The connection was subscribed during the timeout time frame. # The timeout should be adjusted based on the time spent # waiting for the subscription time_spent = time.time() - start_time timeout = max(0.0, timeout - time_spent) else: # The connection isn't subscribed to any channels or patterns, # so no messages are available return None response = self.parse_response(block=False, timeout=timeout) if response: return self.handle_message(response, ignore_subscribe_messages) return None def ping(self, message=None): """ Ping the Redis server """ message = "" if message is None else message return self.execute_command("PING", message) def handle_message(self, response, ignore_subscribe_messages=False): """ Parses a pub/sub message. If the channel or pattern was subscribed to with a message handler, the handler is invoked instead of a parsed message being returned. """ if response is None: return None message_type = str_if_bytes(response[0]) if message_type == "pmessage": message = { "type": message_type, "pattern": response[1], "channel": response[2], "data": response[3], } elif message_type == "pong": message = { "type": message_type, "pattern": None, "channel": None, "data": response[1], } else: message = { "type": message_type, "pattern": None, "channel": response[1], "data": response[2], } # if this is an unsubscribe message, remove it from memory if message_type in self.UNSUBSCRIBE_MESSAGE_TYPES: if message_type == "punsubscribe": pattern = response[1] if pattern in self.pending_unsubscribe_patterns: self.pending_unsubscribe_patterns.remove(pattern) self.patterns.pop(pattern, None) else: channel = response[1] if channel in self.pending_unsubscribe_channels: self.pending_unsubscribe_channels.remove(channel) self.channels.pop(channel, None) if not self.channels and not self.patterns: # There are no subscriptions anymore, set subscribed_event flag # to false self.subscribed_event.clear() if message_type in self.PUBLISH_MESSAGE_TYPES: # if there's a message handler, invoke it if message_type == "pmessage": handler = self.patterns.get(message["pattern"], None) else: handler = self.channels.get(message["channel"], None) if handler: handler(message) return None elif message_type != "pong": # this is a subscribe/unsubscribe message. ignore if we don't # want them if ignore_subscribe_messages or self.ignore_subscribe_messages: return None return message def run_in_thread(self, sleep_time=0, daemon=False, exception_handler=None): for channel, handler in self.channels.items(): if handler is None: raise PubSubError(f"Channel: '{channel}' has no handler registered") for pattern, handler in self.patterns.items(): if handler is None: raise PubSubError(f"Pattern: '{pattern}' has no handler registered") thread = PubSubWorkerThread( self, sleep_time, daemon=daemon, exception_handler=exception_handler ) thread.start() return thread class PubSubWorkerThread(threading.Thread): def __init__(self, pubsub, sleep_time, daemon=False, exception_handler=None): super().__init__() self.daemon = daemon self.pubsub = pubsub self.sleep_time = sleep_time self.exception_handler = exception_handler self._running = threading.Event() def run(self): if self._running.is_set(): return self._running.set() pubsub = self.pubsub sleep_time = self.sleep_time while self._running.is_set(): try: pubsub.get_message(ignore_subscribe_messages=True, timeout=sleep_time) except BaseException as e: if self.exception_handler is None: raise self.exception_handler(e, pubsub, self) pubsub.close() def stop(self): # trip the flag so the run loop exits. the run loop will # close the pubsub connection, which disconnects the socket # and returns the connection to the pool. self._running.clear() class Pipeline(Redis): """ Pipelines provide a way to transmit multiple commands to the Redis server in one transmission. This is convenient for batch processing, such as saving all the values in a list to Redis. All commands executed within a pipeline are wrapped with MULTI and EXEC calls. This guarantees all commands executed in the pipeline will be executed atomically. Any command raising an exception does not halt the execution of subsequent commands in the pipeline. Instead, the exception is caught and its instance is placed into the response list returned by execute(). Code iterating over the response list should be able to deal with an instance of an exception as a potential value. In general, these will be ResponseError exceptions, such as those raised when issuing a command on a key of a different datatype. """ UNWATCH_COMMANDS = {"DISCARD", "EXEC", "UNWATCH"} def __init__(self, connection_pool, response_callbacks, transaction, shard_hint): self.connection_pool = connection_pool self.connection = None self.response_callbacks = response_callbacks self.transaction = transaction self.shard_hint = shard_hint self.watching = False self.reset() def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.reset() def __del__(self): try: self.reset() except Exception: pass def __len__(self): return len(self.command_stack) def __bool__(self): """Pipeline instances should always evaluate to True""" return True def reset(self): self.command_stack = [] self.scripts = set() # make sure to reset the connection state in the event that we were # watching something if self.watching and self.connection: try: # call this manually since our unwatch or # immediate_execute_command methods can call reset() self.connection.send_command("UNWATCH") self.connection.read_response() except ConnectionError: # disconnect will also remove any previous WATCHes self.connection.disconnect() # clean up the other instance attributes self.watching = False self.explicit_transaction = False # we can safely return the connection to the pool here since we're # sure we're no longer WATCHing anything if self.connection: self.connection_pool.release(self.connection) self.connection = None def multi(self): """ Start a transactional block of the pipeline after WATCH commands are issued. End the transactional block with `execute`. """ if self.explicit_transaction: raise RedisError("Cannot issue nested calls to MULTI") if self.command_stack: raise RedisError( "Commands without an initial WATCH have already " "been issued" ) self.explicit_transaction = True def execute_command(self, args, kwargs): if (self.watching or args[0] == "WATCH") and not self.explicit_transaction: return self.immediate_execute_command(args, *kwargs) return self.pipeline_execute_command(args, *kwargs) def _disconnect_reset_raise(self, conn, error): """ Close the connection, reset watching state and raise an exception if we were watching, retry_on_timeout is not set, or the error is not a TimeoutError """ conn.disconnect() # if we were already watching a variable, the watch is no longer # valid since this connection has died. raise a WatchError, which # indicates the user should retry this transaction. if self.watching: self.reset() raise WatchError( "A ConnectionError occurred on while " "watching one or more keys" ) # if retry_on_timeout is not set, or the error is not # a TimeoutError, raise it if not (conn.retry_on_timeout and isinstance(error, TimeoutError)): self.reset() raise def immediate_execute_command(self, args, *options): """ Execute a command immediately, but don't auto-retry on a ConnectionError if we're already WATCHing a variable. Used when issuing WATCH or subsequent commands retrieving their values but before MULTI is called. """ command_name = args[0] conn = self.connection # if this is the first call, we need a connection if not conn: conn = self.connection_pool.get_connection(command_name, self.shard_hint) self.connection = conn return conn.retry.call_with_retry( lambda: self._send_command_parse_response( conn, command_name, args, *options ), lambda error: self._disconnect_reset_raise(conn, error), ) def pipeline_execute_command(self, args, options): """ Stage a command to be executed when execute() is next called Returns the current Pipeline object back so commands can be chained together, such as: pipe = pipe.set('foo', 'bar').incr('baz').decr('bang') At some other point, you can then run: pipe.execute(), which will execute all commands queued in the pipe. """ self.command_stack.append((args, options)) return self def _execute_transaction(self, connection, commands, raise_on_error): cmds = chain([(("MULTI",), {})], commands, [(("EXEC",), {})]) all_cmds = connection.pack_commands( [args for args, options in cmds if EMPTY_RESPONSE not in options] ) connection.send_packed_command(all_cmds) errors = [] # parse off the response for MULTI # NOTE: we need to handle ResponseErrors here and continue # so that we read all the additional command messages from # the socket try: self.parse_response(connection, "_") except ResponseError as e: errors.append((0, e)) # and all the other commands for i, command in enumerate(commands): if EMPTY_RESPONSE in command[1]: errors.append((i, command[1][EMPTY_RESPONSE])) else: try: self.parse_response(connection, "_") except ResponseError as e: self.annotate_exception(e, i + 1, command[0]) errors.append((i, e)) # parse the EXEC. try: response = self.parse_response(connection, "_") except ExecAbortError: if errors: raise errors[0][1] raise # EXEC clears any watched keys self.watching = False if response is None: raise WatchError("Watched variable changed.") # put any parse errors into the response for i, e in errors: response.insert(i, e) if len(response) != len(commands): self.connection.disconnect() raise ResponseError( "Wrong number of response items from " "pipeline execution" ) # find any errors in the response and raise if necessary if raise_on_error: self.raise_first_error(commands, response) # We have to run response callbacks manually data = [] for r, cmd in zip(response, commands): if not isinstance(r, Exception): args, options = cmd command_name = args[0] if command_name in self.response_callbacks: r = self.response_callbacks[command_name](r, options) data.append(r) return data def _execute_pipeline(self, connection, commands, raise_on_error): # build up all commands into a single request to increase network perf all_cmds = connection.pack_commands([args for args, _ in commands]) connection.send_packed_command(all_cmds) response = [] for args, options in commands: try: response.append(self.parse_response(connection, args[0], options)) except ResponseError as e: response.append(e) if raise_on_error: self.raise_first_error(commands, response) return response def raise_first_error(self, commands, response): for i, r in enumerate(response): if isinstance(r, ResponseError): self.annotate_exception(r, i + 1, commands[i][0]) raise r def annotate_exception(self, exception, number, command): cmd = " ".join(map(safe_str, command)) msg = ( f"Command # {number} ({cmd}) of pipeline " f"caused error: {exception.args[0]}" ) exception.args = (msg,) + exception.args[1:] def parse_response(self, connection, command_name, options): result = Redis.parse_response(self, connection, command_name, *options) if command_name in self.UNWATCH_COMMANDS: self.watching = False elif command_name == "WATCH": self.watching = True return result def load_scripts(self): # make sure all scripts that are about to be run on this pipeline exist scripts = list(self.scripts) immediate = self.immediate_execute_command shas = [s.sha for s in scripts] # we can't use the normal script_ methods because they would just # get buffered in the pipeline. exists = immediate("SCRIPT EXISTS", shas) if not all(exists): for s, exist in zip(scripts, exists): if not exist: s.sha = immediate("SCRIPT LOAD", s.script) def _disconnect_raise_reset(self, conn, error): """ Close the connection, raise an exception if we were watching, and raise an exception if retry_on_timeout is not set, or the error is not a TimeoutError """ conn.disconnect() # if we were watching a variable, the watch is no longer valid # since this connection has died. raise a WatchError, which # indicates the user should retry this transaction. if self.watching: raise WatchError( "A ConnectionError occurred on while " "watching one or more keys" ) # if retry_on_timeout is not set, or the error is not # a TimeoutError, raise it if not (conn.retry_on_timeout and isinstance(error, TimeoutError)): self.reset() raise def execute(self, raise_on_error=True): """Execute all the commands in the current pipeline""" stack = self.command_stack if not stack and not self.watching: return [] if self.scripts: self.load_scripts() if self.transaction or self.explicit_transaction: execute = self._execute_transaction else: execute = self._execute_pipeline conn = self.connection if not conn: conn = self.connection_pool.get_connection("MULTI", self.shard_hint) # assign to self.connection so reset() releases the connection # back to the pool after we're done self.connection = conn try: return conn.retry.call_with_retry( lambda: execute(conn, stack, raise_on_error), lambda error: self._disconnect_raise_reset(conn, error), ) finally: self.reset() def discard(self): """ Flushes all previously queued commands See: https://redis.io/commands/DISCARD """ self.execute_command("DISCARD") def watch(self, names): """Watches the values at keys ``names``""" if self.explicit_transaction: raise RedisError("Cannot issue a WATCH after a MULTI") return self.execute_command("WATCH", *names) def unwatch(self): """Unwatches all previously specified keys""" return self.watching and self.execute_command("UNWATCH") or True	71.8 KB	2025-05-25 12:26
📄 cluster.py import copy import logging import random import socket import sys import threading import time from collections import OrderedDict from redis.client import CaseInsensitiveDict, PubSub, Redis from redis.commands import CommandsParser, RedisClusterCommands from redis.connection import ConnectionPool, DefaultParser, Encoder, parse_url from redis.crc import REDIS_CLUSTER_HASH_SLOTS, key_slot from redis.exceptions import ( AskError, BusyLoadingError, ClusterCrossSlotError, ClusterDownError, ClusterError, ConnectionError, DataError, MasterDownError, MovedError, RedisClusterException, RedisError, ResponseError, SlotNotCoveredError, TimeoutError, TryAgainError, ) from redis.utils import ( dict_merge, list_keys_to_dict, merge_result, safe_str, str_if_bytes, ) log = logging.getLogger(__name__) def get_node_name(host, port): return f"{host}:{port}" def get_connection(redis_node, args, options): return redis_node.connection or redis_node.connection_pool.get_connection( args[0], options ) def parse_scan_result(command, res, options): keys_list = [] for primary_res in res.values(): keys_list += primary_res[1] return 0, keys_list def parse_pubsub_numsub(command, res, options): numsub_d = OrderedDict() for numsub_tups in res.values(): for channel, numsubbed in numsub_tups: try: numsub_d[channel] += numsubbed except KeyError: numsub_d[channel] = numsubbed ret_numsub = [(channel, numsub) for channel, numsub in numsub_d.items()] return ret_numsub def parse_cluster_slots(resp, options): current_host = options.get("current_host", "") def fix_server(args): return str_if_bytes(args[0]) or current_host, args[1] slots = {} for slot in resp: start, end, primary = slot[:3] replicas = slot[3:] slots[start, end] = { "primary": fix_server(primary), "replicas": [fix_server(replica) for replica in replicas], } return slots PRIMARY = "primary" REPLICA = "replica" SLOT_ID = "slot-id" REDIS_ALLOWED_KEYS = ( "charset", "connection_class", "connection_pool", "client_name", "db", "decode_responses", "encoding", "encoding_errors", "errors", "host", "max_connections", "nodes_flag", "redis_connect_func", "password", "port", "retry", "retry_on_timeout", "socket_connect_timeout", "socket_keepalive", "socket_keepalive_options", "socket_timeout", "ssl", "ssl_ca_certs", "ssl_certfile", "ssl_cert_reqs", "ssl_keyfile", "ssl_password", "unix_socket_path", "username", ) KWARGS_DISABLED_KEYS = ( "host", "port", ) # Not complete, but covers the major ones # https://redis.io/commands READ_COMMANDS = frozenset( [ "BITCOUNT", "BITPOS", "EXISTS", "GEODIST", "GEOHASH", "GEOPOS", "GEORADIUS", "GEORADIUSBYMEMBER", "GET", "GETBIT", "GETRANGE", "HEXISTS", "HGET", "HGETALL", "HKEYS", "HLEN", "HMGET", "HSTRLEN", "HVALS", "KEYS", "LINDEX", "LLEN", "LRANGE", "MGET", "PTTL", "RANDOMKEY", "SCARD", "SDIFF", "SINTER", "SISMEMBER", "SMEMBERS", "SRANDMEMBER", "STRLEN", "SUNION", "TTL", "ZCARD", "ZCOUNT", "ZRANGE", "ZSCORE", ] ) def cleanup_kwargs(kwargs): """ Remove unsupported or disabled keys from kwargs """ connection_kwargs = { k: v for k, v in kwargs.items() if k in REDIS_ALLOWED_KEYS and k not in KWARGS_DISABLED_KEYS } return connection_kwargs class ClusterParser(DefaultParser): EXCEPTION_CLASSES = dict_merge( DefaultParser.EXCEPTION_CLASSES, { "ASK": AskError, "TRYAGAIN": TryAgainError, "MOVED": MovedError, "CLUSTERDOWN": ClusterDownError, "CROSSSLOT": ClusterCrossSlotError, "MASTERDOWN": MasterDownError, }, ) class RedisCluster(RedisClusterCommands): RedisClusterRequestTTL = 16 PRIMARIES = "primaries" REPLICAS = "replicas" ALL_NODES = "all" RANDOM = "random" DEFAULT_NODE = "default-node" NODE_FLAGS = {PRIMARIES, REPLICAS, ALL_NODES, RANDOM, DEFAULT_NODE} COMMAND_FLAGS = dict_merge( list_keys_to_dict( [ "ACL CAT", "ACL DELUSER", "ACL GENPASS", "ACL GETUSER", "ACL HELP", "ACL LIST", "ACL LOG", "ACL LOAD", "ACL SAVE", "ACL SETUSER", "ACL USERS", "ACL WHOAMI", "CLIENT LIST", "CLIENT SETNAME", "CLIENT GETNAME", "CONFIG SET", "CONFIG REWRITE", "CONFIG RESETSTAT", "TIME", "PUBSUB CHANNELS", "PUBSUB NUMPAT", "PUBSUB NUMSUB", "PING", "INFO", "SHUTDOWN", "KEYS", "SCAN", "DBSIZE", "BGSAVE", "SLOWLOG GET", "SLOWLOG LEN", "SLOWLOG RESET", "WAIT", "SAVE", "MEMORY PURGE", "MEMORY MALLOC-STATS", "MEMORY STATS", "LASTSAVE", "CLIENT TRACKINGINFO", "CLIENT PAUSE", "CLIENT UNPAUSE", "CLIENT UNBLOCK", "CLIENT ID", "CLIENT REPLY", "CLIENT GETREDIR", "CLIENT INFO", "CLIENT KILL", "READONLY", "READWRITE", "CLUSTER INFO", "CLUSTER MEET", "CLUSTER NODES", "CLUSTER REPLICAS", "CLUSTER RESET", "CLUSTER SET-CONFIG-EPOCH", "CLUSTER SLOTS", "CLUSTER COUNT-FAILURE-REPORTS", "CLUSTER KEYSLOT", "COMMAND", "COMMAND COUNT", "COMMAND GETKEYS", "CONFIG GET", "DEBUG", "RANDOMKEY", "READONLY", "READWRITE", "TIME", ], DEFAULT_NODE, ), list_keys_to_dict( [ "FLUSHALL", "FLUSHDB", ], PRIMARIES, ), list_keys_to_dict( [ "CLUSTER COUNTKEYSINSLOT", "CLUSTER DELSLOTS", "CLUSTER GETKEYSINSLOT", "CLUSTER SETSLOT", ], SLOT_ID, ), ) CLUSTER_COMMANDS_RESPONSE_CALLBACKS = { "CLUSTER ADDSLOTS": bool, "CLUSTER COUNT-FAILURE-REPORTS": int, "CLUSTER COUNTKEYSINSLOT": int, "CLUSTER DELSLOTS": bool, "CLUSTER FAILOVER": bool, "CLUSTER FORGET": bool, "CLUSTER GETKEYSINSLOT": list, "CLUSTER KEYSLOT": int, "CLUSTER MEET": bool, "CLUSTER REPLICATE": bool, "CLUSTER RESET": bool, "CLUSTER SAVECONFIG": bool, "CLUSTER SET-CONFIG-EPOCH": bool, "CLUSTER SETSLOT": bool, "CLUSTER SLOTS": parse_cluster_slots, "ASKING": bool, "READONLY": bool, "READWRITE": bool, } RESULT_CALLBACKS = dict_merge( list_keys_to_dict( [ "PUBSUB NUMSUB", ], parse_pubsub_numsub, ), list_keys_to_dict( [ "PUBSUB NUMPAT", ], lambda command, res: sum(list(res.values())), ), list_keys_to_dict( [ "KEYS", "PUBSUB CHANNELS", ], merge_result, ), list_keys_to_dict( [ "PING", "CONFIG SET", "CONFIG REWRITE", "CONFIG RESETSTAT", "CLIENT SETNAME", "BGSAVE", "SLOWLOG RESET", "SAVE", "MEMORY PURGE", "CLIENT PAUSE", "CLIENT UNPAUSE", ], lambda command, res: all(res.values()) if isinstance(res, dict) else res, ), list_keys_to_dict( [ "DBSIZE", "WAIT", ], lambda command, res: sum(res.values()) if isinstance(res, dict) else res, ), list_keys_to_dict( [ "CLIENT UNBLOCK", ], lambda command, res: 1 if sum(res.values()) > 0 else 0, ), list_keys_to_dict( [ "SCAN", ], parse_scan_result, ), ) ERRORS_ALLOW_RETRY = ( ConnectionError, TimeoutError, ClusterDownError, ) def __init__( self, host=None, port=6379, startup_nodes=None, cluster_error_retry_attempts=3, require_full_coverage=False, reinitialize_steps=10, read_from_replicas=False, url=None, kwargs, ): """ Initialize a new RedisCluster client. :startup_nodes: 'list[ClusterNode]' List of nodes from which initial bootstrapping can be done :host: 'str' Can be used to point to a startup node :port: 'int' Can be used to point to a startup node :require_full_coverage: 'bool' When set to False (default value): the client will not require a full coverage of the slots. However, if not all slots are covered, and at least one node has 'cluster-require-full-coverage' set to 'yes,' the server will throw a ClusterDownError for some key-based commands. See - https://redis.io/topics/cluster-tutorial#redis-cluster-configuration-parameters When set to True: all slots must be covered to construct the cluster client. If not all slots are covered, RedisClusterException will be thrown. :read_from_replicas: 'bool' Enable read from replicas in READONLY mode. You can read possibly stale data. When set to true, read commands will be assigned between the primary and its replications in a Round-Robin manner. :cluster_error_retry_attempts: 'int' Retry command execution attempts when encountering ClusterDownError or ConnectionError :reinitialize_steps: 'int' Specifies the number of MOVED errors that need to occur before reinitializing the whole cluster topology. If a MOVED error occurs and the cluster does not need to be reinitialized on this current error handling, only the MOVED slot will be patched with the redirected node. To reinitialize the cluster on every MOVED error, set reinitialize_steps to 1. To avoid reinitializing the cluster on moved errors, set reinitialize_steps to 0. :kwargs: Extra arguments that will be sent into Redis instance when created (See Official redis-py doc for supported kwargs [https://github.com/andymccurdy/redis-py/blob/master/redis/client.py]) Some kwargs are not supported and will raise a RedisClusterException: - db (Redis do not support database SELECT in cluster mode) """ log.info("Creating a new instance of RedisCluster client") if startup_nodes is None: startup_nodes = [] if "db" in kwargs: # Argument 'db' is not possible to use in cluster mode raise RedisClusterException( "Argument 'db' is not possible to use in cluster mode" ) # Get the startup node/s from_url = False if url is not None: from_url = True url_options = parse_url(url) if "path" in url_options: raise RedisClusterException( "RedisCluster does not currently support Unix Domain " "Socket connections" ) if "db" in url_options and url_options["db"] != 0: # Argument 'db' is not possible to use in cluster mode raise RedisClusterException( "A ``db`` querystring option can only be 0 in cluster mode" ) kwargs.update(url_options) host = kwargs.get("host") port = kwargs.get("port", port) startup_nodes.append(ClusterNode(host, port)) elif host is not None and port is not None: startup_nodes.append(ClusterNode(host, port)) elif len(startup_nodes) == 0: # No startup node was provided raise RedisClusterException( "RedisCluster requires at least one node to discover the " "cluster. Please provide one of the followings:\n" "1. host and port, for example:\n" " RedisCluster(host='localhost', port=6379)\n" "2. list of startup nodes, for example:\n" " RedisCluster(startup_nodes=[ClusterNode('localhost', 6379)," " ClusterNode('localhost', 6378)])" ) log.debug(f"startup_nodes : {startup_nodes}") # Update the connection arguments # Whenever a new connection is established, RedisCluster's on_connect # method should be run # If the user passed on_connect function we'll save it and run it # inside the RedisCluster.on_connect() function self.user_on_connect_func = kwargs.pop("redis_connect_func", None) kwargs.update({"redis_connect_func": self.on_connect}) kwargs = cleanup_kwargs(kwargs) self.encoder = Encoder( kwargs.get("encoding", "utf-8"), kwargs.get("encoding_errors", "strict"), kwargs.get("decode_responses", False), ) self.cluster_error_retry_attempts = cluster_error_retry_attempts self.command_flags = self.__class__.COMMAND_FLAGS.copy() self.node_flags = self.__class__.NODE_FLAGS.copy() self.read_from_replicas = read_from_replicas self.reinitialize_counter = 0 self.reinitialize_steps = reinitialize_steps self.nodes_manager = None self.nodes_manager = NodesManager( startup_nodes=startup_nodes, from_url=from_url, require_full_coverage=require_full_coverage, kwargs, ) self.cluster_response_callbacks = CaseInsensitiveDict( self.__class__.CLUSTER_COMMANDS_RESPONSE_CALLBACKS ) self.result_callbacks = CaseInsensitiveDict(self.__class__.RESULT_CALLBACKS) self.commands_parser = CommandsParser(self) self._lock = threading.Lock() def __enter__(self): return self def __exit__(self, exc_type, exc_value, traceback): self.close() def __del__(self): self.close() def disconnect_connection_pools(self): for node in self.get_nodes(): if node.redis_connection: try: node.redis_connection.connection_pool.disconnect() except OSError: # Client was already disconnected. do nothing pass @classmethod def from_url(cls, url, kwargs): """ Return a Redis client object configured from the given URL For example:: redis://[[username]:[password]]@localhost:6379/0 rediss://[[username]:[password]]@localhost:6379/0 unix://[[username]:[password]]@/path/to/socket.sock?db=0 Three URL schemes are supported: - `redis://` creates a TCP socket connection. See more at: <https://www.iana.org/assignments/uri-schemes/prov/redis> - `rediss://` creates a SSL wrapped TCP socket connection. See more at: <https://www.iana.org/assignments/uri-schemes/prov/rediss> - ``unix://``: creates a Unix Domain Socket connection. The username, password, hostname, path and all querystring values are passed through urllib.parse.unquote in order to replace any percent-encoded values with their corresponding characters. There are several ways to specify a database number. The first value found will be used: 1. A ``db`` querystring option, e.g. redis://localhost?db=0 2. If using the redis:// or rediss:// schemes, the path argument of the url, e.g. redis://localhost/0 3. A ``db`` keyword argument to this function. If none of these options are specified, the default db=0 is used. All querystring options are cast to their appropriate Python types. Boolean arguments can be specified with string values "True"/"False" or "Yes"/"No". Values that cannot be properly cast cause a ``ValueError`` to be raised. Once parsed, the querystring arguments and keyword arguments are passed to the ``ConnectionPool``'s class initializer. In the case of conflicting arguments, querystring arguments always win. """ return cls(url=url, kwargs) def on_connect(self, connection): """ Initialize the connection, authenticate and select a database and send READONLY if it is set during object initialization. """ connection.set_parser(ClusterParser) connection.on_connect() if self.read_from_replicas: # Sending READONLY command to server to configure connection as # readonly. Since each cluster node may change its server type due # to a failover, we should establish a READONLY connection # regardless of the server type. If this is a primary connection, # READONLY would not affect executing write commands. connection.send_command("READONLY") if str_if_bytes(connection.read_response()) != "OK": raise ConnectionError("READONLY command failed") if self.user_on_connect_func is not None: self.user_on_connect_func(connection) def get_redis_connection(self, node): if not node.redis_connection: with self._lock: if not node.redis_connection: self.nodes_manager.create_redis_connections([node]) return node.redis_connection def get_node(self, host=None, port=None, node_name=None): return self.nodes_manager.get_node(host, port, node_name) def get_primaries(self): return self.nodes_manager.get_nodes_by_server_type(PRIMARY) def get_replicas(self): return self.nodes_manager.get_nodes_by_server_type(REPLICA) def get_random_node(self): return random.choice(list(self.nodes_manager.nodes_cache.values())) def get_nodes(self): return list(self.nodes_manager.nodes_cache.values()) def get_node_from_key(self, key, replica=False): """ Get the node that holds the key's slot. If replica set to True but the slot doesn't have any replicas, None is returned. """ slot = self.keyslot(key) slot_cache = self.nodes_manager.slots_cache.get(slot) if slot_cache is None or len(slot_cache) == 0: raise SlotNotCoveredError(f'Slot "{slot}" is not covered by the cluster.') if replica and len(self.nodes_manager.slots_cache[slot]) < 2: return None elif replica: node_idx = 1 else: # primary node_idx = 0 return slot_cache[node_idx] def get_default_node(self): """ Get the cluster's default node """ return self.nodes_manager.default_node def set_default_node(self, node): """ Set the default node of the cluster. :param node: 'ClusterNode' :return True if the default node was set, else False """ if node is None or self.get_node(node_name=node.name) is None: log.info( "The requested node does not exist in the cluster, so " "the default node was not changed." ) return False self.nodes_manager.default_node = node log.info(f"Changed the default cluster node to {node}") return True def monitor(self, target_node=None): """ Returns a Monitor object for the specified target node. The default cluster node will be selected if no target node was specified. Monitor is useful for handling the MONITOR command to the redis server. next_command() method returns one command from monitor listen() method yields commands from monitor. """ if target_node is None: target_node = self.get_default_node() if target_node.redis_connection is None: raise RedisClusterException( f"Cluster Node {target_node.name} has no redis_connection" ) return target_node.redis_connection.monitor() def pubsub(self, node=None, host=None, port=None, kwargs): """ Allows passing a ClusterNode, or host&port, to get a pubsub instance connected to the specified node """ return ClusterPubSub(self, node=node, host=host, port=port, *kwargs) def pipeline(self, transaction=None, shard_hint=None): """ Cluster impl: Pipelines do not work in cluster mode the same way they do in normal mode. Create a clone of this object so that simulating pipelines will work correctly. Each command will be called directly when used and when calling execute() will only return the result stack. """ if shard_hint: raise RedisClusterException("shard_hint is deprecated in cluster mode") if transaction: raise RedisClusterException("transaction is deprecated in cluster mode") return ClusterPipeline( nodes_manager=self.nodes_manager, startup_nodes=self.nodes_manager.startup_nodes, result_callbacks=self.result_callbacks, cluster_response_callbacks=self.cluster_response_callbacks, cluster_error_retry_attempts=self.cluster_error_retry_attempts, read_from_replicas=self.read_from_replicas, reinitialize_steps=self.reinitialize_steps, ) def _determine_nodes(self, args, *kwargs): command = args[0] nodes_flag = kwargs.pop("nodes_flag", None) if nodes_flag is not None: # nodes flag passed by the user command_flag = nodes_flag else: # get the nodes group for this command if it was predefined command_flag = self.command_flags.get(command) if command_flag: log.debug(f"Target node/s for {command}: {command_flag}") if command_flag == self.__class__.RANDOM: # return a random node return [self.get_random_node()] elif command_flag == self.__class__.PRIMARIES: # return all primaries return self.get_primaries() elif command_flag == self.__class__.REPLICAS: # return all replicas return self.get_replicas() elif command_flag == self.__class__.ALL_NODES: # return all nodes return self.get_nodes() elif command_flag == self.__class__.DEFAULT_NODE: # return the cluster's default node return [self.nodes_manager.default_node] else: # get the node that holds the key's slot slot = self.determine_slot(args) node = self.nodes_manager.get_node_from_slot( slot, self.read_from_replicas and command in READ_COMMANDS ) log.debug(f"Target for {args}: slot {slot}") return [node] def _should_reinitialized(self): # To reinitialize the cluster on every MOVED error, # set reinitialize_steps to 1. # To avoid reinitializing the cluster on moved errors, set # reinitialize_steps to 0. if self.reinitialize_steps == 0: return False else: return self.reinitialize_counter % self.reinitialize_steps == 0 def keyslot(self, key): """ Calculate keyslot for a given key. See Keys distribution model in https://redis.io/topics/cluster-spec """ k = self.encoder.encode(key) return key_slot(k) def _get_command_keys(self, args): """ Get the keys in the command. If the command has no keys in in, None is returned. """ redis_conn = self.get_default_node().redis_connection return self.commands_parser.get_keys(redis_conn, args) def determine_slot(self, args): """ Figure out what slot based on command and args """ if self.command_flags.get(args[0]) == SLOT_ID: # The command contains the slot ID return args[1] # Get the keys in the command keys = self._get_command_keys(args) if keys is None or len(keys) == 0: raise RedisClusterException( "No way to dispatch this command to Redis Cluster. " "Missing key.\nYou can execute the command by specifying " f"target nodes.\nCommand: {args}" ) if len(keys) > 1: # multi-key command, we need to make sure all keys are mapped to # the same slot slots = {self.keyslot(key) for key in keys} if len(slots) != 1: raise RedisClusterException( f"{args[0]} - all keys must map to the same key slot" ) return slots.pop() else: # single key command return self.keyslot(keys[0]) def reinitialize_caches(self): self.nodes_manager.initialize() def get_encoder(self): """ Get the connections' encoder """ return self.encoder def get_connection_kwargs(self): """ Get the connections' key-word arguments """ return self.nodes_manager.connection_kwargs def _is_nodes_flag(self, target_nodes): return isinstance(target_nodes, str) and target_nodes in self.node_flags def _parse_target_nodes(self, target_nodes): if isinstance(target_nodes, list): nodes = target_nodes elif isinstance(target_nodes, ClusterNode): # Supports passing a single ClusterNode as a variable nodes = [target_nodes] elif isinstance(target_nodes, dict): # Supports dictionaries of the format {node_name: node}. # It enables to execute commands with multi nodes as follows: # rc.cluster_save_config(rc.get_primaries()) nodes = target_nodes.values() else: raise TypeError( "target_nodes type can be one of the following: " "node_flag (PRIMARIES, REPLICAS, RANDOM, ALL_NODES)," "ClusterNode, list<ClusterNode>, or dict<any, ClusterNode>. " f"The passed type is {type(target_nodes)}" ) return nodes def execute_command(self, args, kwargs): """ Wrapper for ERRORS_ALLOW_RETRY error handling. It will try the number of times specified by the config option "self.cluster_error_retry_attempts" which defaults to 3 unless manually configured. If it reaches the number of times, the command will raise the exception Key argument :target_nodes: can be passed with the following types: nodes_flag: PRIMARIES, REPLICAS, ALL_NODES, RANDOM ClusterNode list<ClusterNode> dict<Any, ClusterNode> """ target_nodes_specified = False target_nodes = None passed_targets = kwargs.pop("target_nodes", None) if passed_targets is not None and not self._is_nodes_flag(passed_targets): target_nodes = self._parse_target_nodes(passed_targets) target_nodes_specified = True # If an error that allows retrying was thrown, the nodes and slots # cache were reinitialized. We will retry executing the command with # the updated cluster setup only when the target nodes can be # determined again with the new cache tables. Therefore, when target # nodes were passed to this function, we cannot retry the command # execution since the nodes may not be valid anymore after the tables # were reinitialized. So in case of passed target nodes, # retry_attempts will be set to 1. retry_attempts = ( 1 if target_nodes_specified else self.cluster_error_retry_attempts ) exception = None for _ in range(0, retry_attempts): try: res = {} if not target_nodes_specified: # Determine the nodes to execute the command on target_nodes = self._determine_nodes( args, *kwargs, nodes_flag=passed_targets ) if not target_nodes: raise RedisClusterException( f"No targets were found to execute {args} command on" ) for node in target_nodes: res[node.name] = self._execute_command(node, args, kwargs) # Return the processed result return self._process_result(args[0], res, kwargs) except BaseException as e: if type(e) in RedisCluster.ERRORS_ALLOW_RETRY: # The nodes and slots cache were reinitialized. # Try again with the new cluster setup. exception = e else: # All other errors should be raised. raise e # If it fails the configured number of times then raise exception back # to caller of this method raise exception def _execute_command(self, target_node, args, kwargs): """ Send a command to a node in the cluster """ command = args[0] redis_node = None connection = None redirect_addr = None asking = False moved = False ttl = int(self.RedisClusterRequestTTL) connection_error_retry_counter = 0 while ttl > 0: ttl -= 1 try: if asking: target_node = self.get_node(node_name=redirect_addr) elif moved: # MOVED occurred and the slots cache was updated, # refresh the target node slot = self.determine_slot(args) target_node = self.nodes_manager.get_node_from_slot( slot, self.read_from_replicas and command in READ_COMMANDS ) moved = False log.debug( f"Executing command {command} on target node: " f"{target_node.server_type} {target_node.name}" ) redis_node = self.get_redis_connection(target_node) connection = get_connection(redis_node, args, kwargs) if asking: connection.send_command("ASKING") redis_node.parse_response(connection, "ASKING", kwargs) asking = False connection.send_command(args) response = redis_node.parse_response(connection, command, kwargs) if command in self.cluster_response_callbacks: response = self.cluster_response_callbacks[command]( response, kwargs ) return response except (RedisClusterException, BusyLoadingError) as e: log.exception(type(e)) raise except (ConnectionError, TimeoutError) as e: log.exception(type(e)) # ConnectionError can also be raised if we couldn't get a # connection from the pool before timing out, so check that # this is an actual connection before attempting to disconnect. if connection is not None: connection.disconnect() connection_error_retry_counter += 1 # Give the node 0.25 seconds to get back up and retry again # with same node and configuration. After 5 attempts then try # to reinitialize the cluster and see if the nodes # configuration has changed or not if connection_error_retry_counter < 5: time.sleep(0.25) else: # Hard force of reinitialize of the node/slots setup # and try again with the new setup self.nodes_manager.initialize() raise except MovedError as e: # First, we will try to patch the slots/nodes cache with the # redirected node output and try again. If MovedError exceeds # 'reinitialize_steps' number of times, we will force # reinitializing the tables, and then try again. # 'reinitialize_steps' counter will increase faster when # the same client object is shared between multiple threads. To # reduce the frequency you can set this variable in the # RedisCluster constructor. log.exception("MovedError") self.reinitialize_counter += 1 if self._should_reinitialized(): self.nodes_manager.initialize() # Reset the counter self.reinitialize_counter = 0 else: self.nodes_manager.update_moved_exception(e) moved = True except TryAgainError: log.exception("TryAgainError") if ttl < self.RedisClusterRequestTTL / 2: time.sleep(0.05) except AskError as e: log.exception("AskError") redirect_addr = get_node_name(host=e.host, port=e.port) asking = True except ClusterDownError as e: log.exception("ClusterDownError") # ClusterDownError can occur during a failover and to get # self-healed, we will try to reinitialize the cluster layout # and retry executing the command time.sleep(0.25) self.nodes_manager.initialize() raise e except ResponseError as e: message = e.__str__() log.exception(f"ResponseError: {message}") raise e except BaseException as e: log.exception("BaseException") if connection: connection.disconnect() raise e finally: if connection is not None: redis_node.connection_pool.release(connection) raise ClusterError("TTL exhausted.") def close(self): try: with self._lock: if self.nodes_manager: self.nodes_manager.close() except AttributeError: # RedisCluster's __init__ can fail before nodes_manager is set pass def _process_result(self, command, res, kwargs): """ Process the result of the executed command. The function would return a dict or a single value. :type command: str :type res: dict `res` should be in the following format: Dict<node_name, command_result> """ if command in self.result_callbacks: return self.result_callbacks[command](command, res, kwargs) elif len(res) == 1: # When we execute the command on a single node, we can # remove the dictionary and return a single response return list(res.values())[0] else: return res class ClusterNode: def __init__(self, host, port, server_type=None, redis_connection=None): if host == "localhost": host = socket.gethostbyname(host) self.host = host self.port = port self.name = get_node_name(host, port) self.server_type = server_type self.redis_connection = redis_connection def __repr__(self): return ( f"[host={self.host}," f"port={self.port}," f"name={self.name}," f"server_type={self.server_type}," f"redis_connection={self.redis_connection}]" ) def __eq__(self, obj): return isinstance(obj, ClusterNode) and obj.name == self.name def __del__(self): if self.redis_connection is not None: self.redis_connection.close() class LoadBalancer: """ Round-Robin Load Balancing """ def __init__(self, start_index=0): self.primary_to_idx = {} self.start_index = start_index def get_server_index(self, primary, list_size): server_index = self.primary_to_idx.setdefault(primary, self.start_index) # Update the index self.primary_to_idx[primary] = (server_index + 1) % list_size return server_index def reset(self): self.primary_to_idx.clear() class NodesManager: def __init__( self, startup_nodes, from_url=False, require_full_coverage=False, lock=None, kwargs, ): self.nodes_cache = {} self.slots_cache = {} self.startup_nodes = {} self.default_node = None self.populate_startup_nodes(startup_nodes) self.from_url = from_url self._require_full_coverage = require_full_coverage self._moved_exception = None self.connection_kwargs = kwargs self.read_load_balancer = LoadBalancer() if lock is None: lock = threading.Lock() self._lock = lock self.initialize() def get_node(self, host=None, port=None, node_name=None): """ Get the requested node from the cluster's nodes. nodes. :return: ClusterNode if the node exists, else None """ if host and port: # the user passed host and port if host == "localhost": host = socket.gethostbyname(host) return self.nodes_cache.get(get_node_name(host=host, port=port)) elif node_name: return self.nodes_cache.get(node_name) else: log.error( "get_node requires one of the following: " "1. node name " "2. host and port" ) return None def update_moved_exception(self, exception): self._moved_exception = exception def _update_moved_slots(self): """ Update the slot's node with the redirected one """ e = self._moved_exception redirected_node = self.get_node(host=e.host, port=e.port) if redirected_node is not None: # The node already exists if redirected_node.server_type is not PRIMARY: # Update the node's server type redirected_node.server_type = PRIMARY else: # This is a new node, we will add it to the nodes cache redirected_node = ClusterNode(e.host, e.port, PRIMARY) self.nodes_cache[redirected_node.name] = redirected_node if redirected_node in self.slots_cache[e.slot_id]: # The MOVED error resulted from a failover, and the new slot owner # had previously been a replica. old_primary = self.slots_cache[e.slot_id][0] # Update the old primary to be a replica and add it to the end of # the slot's node list old_primary.server_type = REPLICA self.slots_cache[e.slot_id].append(old_primary) # Remove the old replica, which is now a primary, from the slot's # node list self.slots_cache[e.slot_id].remove(redirected_node) # Override the old primary with the new one self.slots_cache[e.slot_id][0] = redirected_node if self.default_node == old_primary: # Update the default node with the new primary self.default_node = redirected_node else: # The new slot owner is a new server, or a server from a different # shard. We need to remove all current nodes from the slot's list # (including replications) and add just the new node. self.slots_cache[e.slot_id] = [redirected_node] # Reset moved_exception self._moved_exception = None def get_node_from_slot(self, slot, read_from_replicas=False, server_type=None): """ Gets a node that servers this hash slot """ if self._moved_exception: with self._lock: if self._moved_exception: self._update_moved_slots() if self.slots_cache.get(slot) is None or len(self.slots_cache[slot]) == 0: raise SlotNotCoveredError( f'Slot "{slot}" not covered by the cluster. ' f'"require_full_coverage={self._require_full_coverage}"' ) if read_from_replicas is True: # get the server index in a Round-Robin manner primary_name = self.slots_cache[slot][0].name node_idx = self.read_load_balancer.get_server_index( primary_name, len(self.slots_cache[slot]) ) elif ( server_type is None or server_type == PRIMARY or len(self.slots_cache[slot]) == 1 ): # return a primary node_idx = 0 else: # return a replica # randomly choose one of the replicas node_idx = random.randint(1, len(self.slots_cache[slot]) - 1) return self.slots_cache[slot][node_idx] def get_nodes_by_server_type(self, server_type): """ Get all nodes with the specified server type :param server_type: 'primary' or 'replica' :return: list of ClusterNode """ return [ node for node in self.nodes_cache.values() if node.server_type == server_type ] def populate_startup_nodes(self, nodes): """ Populate all startup nodes and filters out any duplicates """ for n in nodes: self.startup_nodes[n.name] = n def check_slots_coverage(self, slots_cache): # Validate if all slots are covered or if we should try next # startup node for i in range(0, REDIS_CLUSTER_HASH_SLOTS): if i not in slots_cache: return False return True def create_redis_connections(self, nodes): """ This function will create a redis connection to all nodes in :nodes: """ for node in nodes: if node.redis_connection is None: node.redis_connection = self.create_redis_node( host=node.host, port=node.port, self.connection_kwargs, ) def create_redis_node(self, host, port, kwargs): if self.from_url: # Create a redis node with a costumed connection pool kwargs.update({"host": host}) kwargs.update({"port": port}) r = Redis(connection_pool=ConnectionPool(kwargs)) else: r = Redis(host=host, port=port, kwargs) return r def initialize(self): """ Initializes the nodes cache, slots cache and redis connections. :startup_nodes: Responsible for discovering other nodes in the cluster """ log.debug("Initializing the nodes' topology of the cluster") self.reset() tmp_nodes_cache = {} tmp_slots = {} disagreements = [] startup_nodes_reachable = False fully_covered = False kwargs = self.connection_kwargs for startup_node in self.startup_nodes.values(): try: if startup_node.redis_connection: r = startup_node.redis_connection else: # Create a new Redis connection and let Redis decode the # responses so we won't need to handle that copy_kwargs = copy.deepcopy(kwargs) copy_kwargs.update({"decode_responses": True, "encoding": "utf-8"}) r = self.create_redis_node( startup_node.host, startup_node.port, copy_kwargs ) self.startup_nodes[startup_node.name].redis_connection = r # Make sure cluster mode is enabled on this node if bool(r.info().get("cluster_enabled")) is False: raise RedisClusterException( "Cluster mode is not enabled on this node" ) cluster_slots = str_if_bytes(r.execute_command("CLUSTER SLOTS")) startup_nodes_reachable = True except (ConnectionError, TimeoutError) as e: msg = e.__str__ log.exception( "An exception occurred while trying to" " initialize the cluster using the seed node" f" {startup_node.name}:\n{msg}" ) continue except ResponseError as e: log.exception('ReseponseError sending "cluster slots" to redis server') # Isn't a cluster connection, so it won't parse these # exceptions automatically message = e.__str__() if "CLUSTERDOWN" in message or "MASTERDOWN" in message: continue else: raise RedisClusterException( 'ERROR sending "cluster slots" command to redis ' f"server: {startup_node}. error: {message}" ) except Exception as e: message = e.__str__() raise RedisClusterException( 'ERROR sending "cluster slots" command to redis ' f"server {startup_node.name}. error: {message}" ) # CLUSTER SLOTS command results in the following output: # [[slot_section[from_slot,to_slot,master,replica1,...,replicaN]]] # where each node contains the following list: [IP, port, node_id] # Therefore, cluster_slots[0][2][0] will be the IP address of the # primary node of the first slot section. # If there's only one server in the cluster, its ``host`` is '' # Fix it to the host in startup_nodes if ( len(cluster_slots) == 1 and len(cluster_slots[0][2][0]) == 0 and len(self.startup_nodes) == 1 ): cluster_slots[0][2][0] = startup_node.host for slot in cluster_slots: primary_node = slot[2] host = primary_node[0] if host == "": host = startup_node.host port = int(primary_node[1]) target_node = tmp_nodes_cache.get(get_node_name(host, port)) if target_node is None: target_node = ClusterNode(host, port, PRIMARY) # add this node to the nodes cache tmp_nodes_cache[target_node.name] = target_node for i in range(int(slot[0]), int(slot[1]) + 1): if i not in tmp_slots: tmp_slots[i] = [] tmp_slots[i].append(target_node) replica_nodes = [slot[j] for j in range(3, len(slot))] for replica_node in replica_nodes: host = replica_node[0] port = replica_node[1] target_replica_node = tmp_nodes_cache.get( get_node_name(host, port) ) if target_replica_node is None: target_replica_node = ClusterNode(host, port, REPLICA) tmp_slots[i].append(target_replica_node) # add this node to the nodes cache tmp_nodes_cache[ target_replica_node.name ] = target_replica_node else: # Validate that 2 nodes want to use the same slot cache # setup tmp_slot = tmp_slots[i][0] if tmp_slot.name != target_node.name: disagreements.append( f"{tmp_slot.name} vs {target_node.name} on slot: {i}" ) if len(disagreements) > 5: raise RedisClusterException( f"startup_nodes could not agree on a valid " f'slots cache: {", ".join(disagreements)}' ) fully_covered = self.check_slots_coverage(tmp_slots) if fully_covered: # Don't need to continue to the next startup node if all # slots are covered break if not startup_nodes_reachable: raise RedisClusterException( "Redis Cluster cannot be connected. Please provide at least " "one reachable node. " ) # Create Redis connections to all nodes self.create_redis_connections(list(tmp_nodes_cache.values())) # Check if the slots are not fully covered if not fully_covered and self._require_full_coverage: # Despite the requirement that the slots be covered, there # isn't a full coverage raise RedisClusterException( f"All slots are not covered after query all startup_nodes. " f"{len(tmp_slots)} of {REDIS_CLUSTER_HASH_SLOTS} " f"covered..." ) # Set the tmp variables to the real variables self.nodes_cache = tmp_nodes_cache self.slots_cache = tmp_slots # Set the default node self.default_node = self.get_nodes_by_server_type(PRIMARY)[0] # Populate the startup nodes with all discovered nodes self.populate_startup_nodes(self.nodes_cache.values()) # If initialize was called after a MovedError, clear it self._moved_exception = None def close(self): self.default_node = None for node in self.nodes_cache.values(): if node.redis_connection: node.redis_connection.close() def reset(self): try: self.read_load_balancer.reset() except TypeError: # The read_load_balancer is None, do nothing pass class ClusterPubSub(PubSub): """ Wrapper for PubSub class. IMPORTANT: before using ClusterPubSub, read about the known limitations with pubsub in Cluster mode and learn how to workaround them: https://redis-py-cluster.readthedocs.io/en/stable/pubsub.html """ def __init__(self, redis_cluster, node=None, host=None, port=None, kwargs): """ When a pubsub instance is created without specifying a node, a single node will be transparently chosen for the pubsub connection on the first command execution. The node will be determined by: 1. Hashing the channel name in the request to find its keyslot 2. Selecting a node that handles the keyslot: If read_from_replicas is set to true, a replica can be selected. :type redis_cluster: RedisCluster :type node: ClusterNode :type host: str :type port: int """ log.info("Creating new instance of ClusterPubSub") self.node = None self.set_pubsub_node(redis_cluster, node, host, port) connection_pool = ( None if self.node is None else redis_cluster.get_redis_connection(self.node).connection_pool ) self.cluster = redis_cluster super().__init__( kwargs, connection_pool=connection_pool, encoder=redis_cluster.encoder ) def set_pubsub_node(self, cluster, node=None, host=None, port=None): """ The pubsub node will be set according to the passed node, host and port When none of the node, host, or port are specified - the node is set to None and will be determined by the keyslot of the channel in the first command to be executed. RedisClusterException will be thrown if the passed node does not exist in the cluster. If host is passed without port, or vice versa, a DataError will be thrown. :type cluster: RedisCluster :type node: ClusterNode :type host: str :type port: int """ if node is not None: # node is passed by the user self._raise_on_invalid_node(cluster, node, node.host, node.port) pubsub_node = node elif host is not None and port is not None: # host and port passed by the user node = cluster.get_node(host=host, port=port) self._raise_on_invalid_node(cluster, node, host, port) pubsub_node = node elif any([host, port]) is True: # only 'host' or 'port' passed raise DataError("Passing a host requires passing a port, " "and vice versa") else: # nothing passed by the user. set node to None pubsub_node = None self.node = pubsub_node def get_pubsub_node(self): """ Get the node that is being used as the pubsub connection """ return self.node def _raise_on_invalid_node(self, redis_cluster, node, host, port): """ Raise a RedisClusterException if the node is None or doesn't exist in the cluster. """ if node is None or redis_cluster.get_node(node_name=node.name) is None: raise RedisClusterException( f"Node {host}:{port} doesn't exist in the cluster" ) def execute_command(self, args, kwargs): """ Execute a publish/subscribe command. Taken code from redis-py and tweak to make it work within a cluster. """ # NOTE: don't parse the response in this function -- it could pull a # legitimate message off the stack if the connection is already # subscribed to one or more channels if self.connection is None: if self.connection_pool is None: if len(args) > 1: # Hash the first channel and get one of the nodes holding # this slot channel = args[1] slot = self.cluster.keyslot(channel) node = self.cluster.nodes_manager.get_node_from_slot( slot, self.cluster.read_from_replicas ) else: # Get a random node node = self.cluster.get_random_node() self.node = node redis_connection = self.cluster.get_redis_connection(node) self.connection_pool = redis_connection.connection_pool self.connection = self.connection_pool.get_connection( "pubsub", self.shard_hint ) # register a callback that re-subscribes to any channels we # were listening to when we were disconnected self.connection.register_connect_callback(self.on_connect) connection = self.connection self._execute(connection, connection.send_command, args) def get_redis_connection(self): """ Get the Redis connection of the pubsub connected node. """ if self.node is not None: return self.node.redis_connection class ClusterPipeline(RedisCluster): """ Support for Redis pipeline in cluster mode """ ERRORS_ALLOW_RETRY = ( ConnectionError, TimeoutError, MovedError, AskError, TryAgainError, ) def __init__( self, nodes_manager, result_callbacks=None, cluster_response_callbacks=None, startup_nodes=None, read_from_replicas=False, cluster_error_retry_attempts=5, reinitialize_steps=10, *kwargs, ): """ """ log.info("Creating new instance of ClusterPipeline") self.command_stack = [] self.nodes_manager = nodes_manager self.refresh_table_asap = False self.result_callbacks = ( result_callbacks or self.__class__.RESULT_CALLBACKS.copy() ) self.startup_nodes = startup_nodes if startup_nodes else [] self.read_from_replicas = read_from_replicas self.command_flags = self.__class__.COMMAND_FLAGS.copy() self.cluster_response_callbacks = cluster_response_callbacks self.cluster_error_retry_attempts = cluster_error_retry_attempts self.reinitialize_counter = 0 self.reinitialize_steps = reinitialize_steps self.encoder = Encoder( kwargs.get("encoding", "utf-8"), kwargs.get("encoding_errors", "strict"), kwargs.get("decode_responses", False), ) # The commands parser refers to the parent # so that we don't push the COMMAND command # onto the stack self.commands_parser = CommandsParser(super()) def __repr__(self): """ """ return f"{type(self).__name__}" def __enter__(self): """ """ return self def __exit__(self, exc_type, exc_value, traceback): """ """ self.reset() def __del__(self): try: self.reset() except Exception: pass def __len__(self): """ """ return len(self.command_stack) def __nonzero__(self): "Pipeline instances should always evaluate to True on Python 2.7" return True def __bool__(self): "Pipeline instances should always evaluate to True on Python 3+" return True def execute_command(self, args, *kwargs): """ Wrapper function for pipeline_execute_command """ return self.pipeline_execute_command(args, *kwargs) def pipeline_execute_command(self, args, *options): """ Appends the executed command to the pipeline's command stack """ self.command_stack.append( PipelineCommand(args, options, len(self.command_stack)) ) return self def raise_first_error(self, stack): """ Raise the first exception on the stack """ for c in stack: r = c.result if isinstance(r, Exception): self.annotate_exception(r, c.position + 1, c.args) raise r def annotate_exception(self, exception, number, command): """ Provides extra context to the exception prior to it being handled """ cmd = " ".join(map(safe_str, command)) msg = ( f"Command # {number} ({cmd}) of pipeline " f"caused error: {exception.args[0]}" ) exception.args = (msg,) + exception.args[1:] def execute(self, raise_on_error=True): """ Execute all the commands in the current pipeline """ stack = self.command_stack try: return self.send_cluster_commands(stack, raise_on_error) finally: self.reset() def reset(self): """ Reset back to empty pipeline. """ self.command_stack = [] self.scripts = set() # TODO: Implement # make sure to reset the connection state in the event that we were # watching something # if self.watching and self.connection: # try: # # call this manually since our unwatch or # # immediate_execute_command methods can call reset() # self.connection.send_command('UNWATCH') # self.connection.read_response() # except ConnectionError: # # disconnect will also remove any previous WATCHes # self.connection.disconnect() # clean up the other instance attributes self.watching = False self.explicit_transaction = False # TODO: Implement # we can safely return the connection to the pool here since we're # sure we're no longer WATCHing anything # if self.connection: # self.connection_pool.release(self.connection) # self.connection = None def send_cluster_commands( self, stack, raise_on_error=True, allow_redirections=True ): """ Wrapper for CLUSTERDOWN error handling. If the cluster reports it is down it is assumed that: - connection_pool was disconnected - connection_pool was reseted - refereh_table_asap set to True It will try the number of times specified by the config option "self.cluster_error_retry_attempts" which defaults to 3 unless manually configured. If it reaches the number of times, the command will raises ClusterDownException. """ if not stack: return [] for _ in range(0, self.cluster_error_retry_attempts): try: return self._send_cluster_commands( stack, raise_on_error=raise_on_error, allow_redirections=allow_redirections, ) except ClusterDownError: # Try again with the new cluster setup. All other errors # should be raised. pass # If it fails the configured number of times then raise # exception back to caller of this method raise ClusterDownError("CLUSTERDOWN error. Unable to rebuild the cluster") def _send_cluster_commands( self, stack, raise_on_error=True, allow_redirections=True ): """ Send a bunch of cluster commands to the redis cluster. `allow_redirections` If the pipeline should follow `ASK` & `MOVED` responses automatically. If set to false it will raise RedisClusterException. """ # the first time sending the commands we send all of # the commands that were queued up. # if we have to run through it again, we only retry # the commands that failed. attempt = sorted(stack, key=lambda x: x.position) # build a list of node objects based on node names we need to nodes = {} # as we move through each command that still needs to be processed, # we figure out the slot number that command maps to, then from # the slot determine the node. for c in attempt: # refer to our internal node -> slot table that # tells us where a given # command should route to. slot = self.determine_slot(c.args) node = self.nodes_manager.get_node_from_slot( slot, self.read_from_replicas and c.args[0] in READ_COMMANDS ) # now that we know the name of the node # ( it's just a string in the form of host:port ) # we can build a list of commands for each node. node_name = node.name if node_name not in nodes: redis_node = self.get_redis_connection(node) connection = get_connection(redis_node, c.args) nodes[node_name] = NodeCommands( redis_node.parse_response, redis_node.connection_pool, connection ) nodes[node_name].append(c) # send the commands in sequence. # we write to all the open sockets for each node first, # before reading anything # this allows us to flush all the requests out across the # network essentially in parallel # so that we can read them all in parallel as they come back. # we dont' multiplex on the sockets as they come available, # but that shouldn't make too much difference. node_commands = nodes.values() for n in node_commands: n.write() for n in node_commands: n.read() # release all of the redis connections we allocated earlier # back into the connection pool. # we used to do this step as part of a try/finally block, # but it is really dangerous to # release connections back into the pool if for some # reason the socket has data still left in it # from a previous operation. The write and # read operations already have try/catch around them for # all known types of errors including connection # and socket level errors. # So if we hit an exception, something really bad # happened and putting any oF # these connections back into the pool is a very bad idea. # the socket might have unread buffer still sitting in it, # and then the next time we read from it we pass the # buffered result back from a previous command and # every single request after to that connection will always get # a mismatched result. for n in nodes.values(): n.connection_pool.release(n.connection) # if the response isn't an exception it is a # valid response from the node # we're all done with that command, YAY! # if we have more commands to attempt, we've run into problems. # collect all the commands we are allowed to retry. # (MOVED, ASK, or connection errors or timeout errors) attempt = sorted( ( c for c in attempt if isinstance(c.result, ClusterPipeline.ERRORS_ALLOW_RETRY) ), key=lambda x: x.position, ) if attempt and allow_redirections: # RETRY MAGIC HAPPENS HERE! # send these remaing comamnds one at a time using `execute_command` # in the main client. This keeps our retry logic # in one place mostly, # and allows us to be more confident in correctness of behavior. # at this point any speed gains from pipelining have been lost # anyway, so we might as well make the best # attempt to get the correct behavior. # # The client command will handle retries for each # individual command sequentially as we pass each # one into `execute_command`. Any exceptions # that bubble out should only appear once all # retries have been exhausted. # # If a lot of commands have failed, we'll be setting the # flag to rebuild the slots table from scratch. # So MOVED errors should correct themselves fairly quickly. log.exception( f"An exception occurred during pipeline execution. " f"args: {attempt[-1].args}, " f"error: {type(attempt[-1].result).__name__} " f"{str(attempt[-1].result)}" ) self.reinitialize_counter += 1 if self._should_reinitialized(): self.nodes_manager.initialize() for c in attempt: try: # send each command individually like we # do in the main client. c.result = super().execute_command(c.args, c.options) except RedisError as e: c.result = e # turn the response back into a simple flat array that corresponds # to the sequence of commands issued in the stack in pipeline.execute() response = [c.result for c in sorted(stack, key=lambda x: x.position)] if raise_on_error: self.raise_first_error(stack) return response def _fail_on_redirect(self, allow_redirections): """ """ if not allow_redirections: raise RedisClusterException( "ASK & MOVED redirection not allowed in this pipeline" ) def eval(self): """ """ raise RedisClusterException("method eval() is not implemented") def multi(self): """ """ raise RedisClusterException("method multi() is not implemented") def immediate_execute_command(self, args, *options): """ """ raise RedisClusterException( "method immediate_execute_command() is not implemented" ) def _execute_transaction(self, args, *kwargs): """ """ raise RedisClusterException("method _execute_transaction() is not implemented") def load_scripts(self): """ """ raise RedisClusterException("method load_scripts() is not implemented") def watch(self, names): """ """ raise RedisClusterException("method watch() is not implemented") def unwatch(self): """ """ raise RedisClusterException("method unwatch() is not implemented") def script_load_for_pipeline(self, args, kwargs): """ """ raise RedisClusterException( "method script_load_for_pipeline() is not implemented" ) def delete(self, names): """ "Delete a key specified by ``names``" """ if len(names) != 1: raise RedisClusterException( "deleting multiple keys is not " "implemented in pipeline command" ) return self.execute_command("DEL", names[0]) def block_pipeline_command(func): """ Prints error because some pipelined commands should be blocked when running in cluster-mode """ def inner(args, kwargs): raise RedisClusterException( f"ERROR: Calling pipelined function {func.__name__} is blocked " f"when running redis in cluster mode..." ) return inner # Blocked pipeline commands ClusterPipeline.bitop = block_pipeline_command(RedisCluster.bitop) ClusterPipeline.brpoplpush = block_pipeline_command(RedisCluster.brpoplpush) ClusterPipeline.client_getname = block_pipeline_command(RedisCluster.client_getname) ClusterPipeline.client_list = block_pipeline_command(RedisCluster.client_list) ClusterPipeline.client_setname = block_pipeline_command(RedisCluster.client_setname) ClusterPipeline.config_set = block_pipeline_command(RedisCluster.config_set) ClusterPipeline.dbsize = block_pipeline_command(RedisCluster.dbsize) ClusterPipeline.flushall = block_pipeline_command(RedisCluster.flushall) ClusterPipeline.flushdb = block_pipeline_command(RedisCluster.flushdb) ClusterPipeline.keys = block_pipeline_command(RedisCluster.keys) ClusterPipeline.mget = block_pipeline_command(RedisCluster.mget) ClusterPipeline.move = block_pipeline_command(RedisCluster.move) ClusterPipeline.mset = block_pipeline_command(RedisCluster.mset) ClusterPipeline.msetnx = block_pipeline_command(RedisCluster.msetnx) ClusterPipeline.pfmerge = block_pipeline_command(RedisCluster.pfmerge) ClusterPipeline.pfcount = block_pipeline_command(RedisCluster.pfcount) ClusterPipeline.ping = block_pipeline_command(RedisCluster.ping) ClusterPipeline.publish = block_pipeline_command(RedisCluster.publish) ClusterPipeline.randomkey = block_pipeline_command(RedisCluster.randomkey) ClusterPipeline.rename = block_pipeline_command(RedisCluster.rename) ClusterPipeline.renamenx = block_pipeline_command(RedisCluster.renamenx) ClusterPipeline.rpoplpush = block_pipeline_command(RedisCluster.rpoplpush) ClusterPipeline.scan = block_pipeline_command(RedisCluster.scan) ClusterPipeline.sdiff = block_pipeline_command(RedisCluster.sdiff) ClusterPipeline.sdiffstore = block_pipeline_command(RedisCluster.sdiffstore) ClusterPipeline.sinter = block_pipeline_command(RedisCluster.sinter) ClusterPipeline.sinterstore = block_pipeline_command(RedisCluster.sinterstore) ClusterPipeline.smove = block_pipeline_command(RedisCluster.smove) ClusterPipeline.sort = block_pipeline_command(RedisCluster.sort) ClusterPipeline.sunion = block_pipeline_command(RedisCluster.sunion) ClusterPipeline.sunionstore = block_pipeline_command(RedisCluster.sunionstore) ClusterPipeline.readwrite = block_pipeline_command(RedisCluster.readwrite) ClusterPipeline.readonly = block_pipeline_command(RedisCluster.readonly) class PipelineCommand: """ """ def __init__(self, args, options=None, position=None): self.args = args if options is None: options = {} self.options = options self.position = position self.result = None self.node = None self.asking = False class NodeCommands: """ """ def __init__(self, parse_response, connection_pool, connection): """ """ self.parse_response = parse_response self.connection_pool = connection_pool self.connection = connection self.commands = [] def append(self, c): """ """ self.commands.append(c) def write(self): """ Code borrowed from Redis so it can be fixed """ connection = self.connection commands = self.commands # We are going to clobber the commands with the write, so go ahead # and ensure that nothing is sitting there from a previous run. for c in commands: c.result = None # build up all commands into a single request to increase network perf # send all the commands and catch connection and timeout errors. try: connection.send_packed_command( connection.pack_commands([c.args for c in commands]) ) except (ConnectionError, TimeoutError) as e: for c in commands: c.result = e def read(self): """ """ connection = self.connection for c in self.commands: # if there is a result on this command, # it means we ran into an exception # like a connection error. Trying to parse # a response on a connection that # is no longer open will result in a # connection error raised by redis-py. # but redis-py doesn't check in parse_response # that the sock object is # still set and if you try to # read from a closed connection, it will # result in an AttributeError because # it will do a readline() call on None. # This can have all kinds of nasty side-effects. # Treating this case as a connection error # is fine because it will dump # the connection object back into the # pool and on the next write, it will # explicitly open the connection and all will be well. if c.result is None: try: c.result = self.parse_response(connection, c.args[0], *c.options) except (ConnectionError, TimeoutError) as e: for c in self.commands: c.result = e return except RedisError: c.result = sys.exc_info()[1]	75.87 KB	2025-05-25 12:26
📁 commands	-	2025-05-25 12:26
📄 connection.py import copy import errno import io import os import socket import threading import weakref from itertools import chain from queue import Empty, Full, LifoQueue from time import time from urllib.parse import parse_qs, unquote, urlparse from packaging.version import Version from redis.backoff import NoBackoff from redis.exceptions import ( AuthenticationError, AuthenticationWrongNumberOfArgsError, BusyLoadingError, ChildDeadlockedError, ConnectionError, DataError, ExecAbortError, InvalidResponse, ModuleError, NoPermissionError, NoScriptError, ReadOnlyError, RedisError, ResponseError, TimeoutError, ) from redis.retry import Retry from redis.utils import CRYPTOGRAPHY_AVAILABLE, HIREDIS_AVAILABLE, str_if_bytes try: import ssl ssl_available = True except ImportError: ssl_available = False NONBLOCKING_EXCEPTION_ERROR_NUMBERS = { BlockingIOError: errno.EWOULDBLOCK, } if ssl_available: if hasattr(ssl, "SSLWantReadError"): NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLWantReadError] = 2 NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLWantWriteError] = 2 else: NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLError] = 2 NONBLOCKING_EXCEPTIONS = tuple(NONBLOCKING_EXCEPTION_ERROR_NUMBERS.keys()) if HIREDIS_AVAILABLE: import hiredis hiredis_version = Version(hiredis.__version__) HIREDIS_SUPPORTS_CALLABLE_ERRORS = hiredis_version >= Version("0.1.3") HIREDIS_SUPPORTS_BYTE_BUFFER = hiredis_version >= Version("0.1.4") HIREDIS_SUPPORTS_ENCODING_ERRORS = hiredis_version >= Version("1.0.0") HIREDIS_USE_BYTE_BUFFER = True # only use byte buffer if hiredis supports it if not HIREDIS_SUPPORTS_BYTE_BUFFER: HIREDIS_USE_BYTE_BUFFER = False SYM_STAR = b"" SYM_DOLLAR = b"$" SYM_CRLF = b"\r\n" SYM_EMPTY = b"" SERVER_CLOSED_CONNECTION_ERROR = "Connection closed by server." SENTINEL = object() MODULE_LOAD_ERROR = "Error loading the extension. " "Please check the server logs." NO_SUCH_MODULE_ERROR = "Error unloading module: no such module with that name" MODULE_UNLOAD_NOT_POSSIBLE_ERROR = "Error unloading module: operation not " "possible." MODULE_EXPORTS_DATA_TYPES_ERROR = ( "Error unloading module: the module " "exports one or more module-side data " "types, can't unload" ) class Encoder: "Encode strings to bytes-like and decode bytes-like to strings" def __init__(self, encoding, encoding_errors, decode_responses): self.encoding = encoding self.encoding_errors = encoding_errors self.decode_responses = decode_responses def encode(self, value): "Return a bytestring or bytes-like representation of the value" if isinstance(value, (bytes, memoryview)): return value elif isinstance(value, bool): # special case bool since it is a subclass of int raise DataError( "Invalid input of type: 'bool'. Convert to a " "bytes, string, int or float first." ) elif isinstance(value, (int, float)): value = repr(value).encode() elif not isinstance(value, str): # a value we don't know how to deal with. throw an error typename = type(value).__name__ raise DataError( f"Invalid input of type: '{typename}'. " f"Convert to a bytes, string, int or float first." ) if isinstance(value, str): value = value.encode(self.encoding, self.encoding_errors) return value def decode(self, value, force=False): "Return a unicode string from the bytes-like representation" if self.decode_responses or force: if isinstance(value, memoryview): value = value.tobytes() if isinstance(value, bytes): value = value.decode(self.encoding, self.encoding_errors) return value class BaseParser: EXCEPTION_CLASSES = { "ERR": { "max number of clients reached": ConnectionError, "Client sent AUTH, but no password is set": AuthenticationError, "invalid password": AuthenticationError, # some Redis server versions report invalid command syntax # in lowercase "wrong number of arguments " "for 'auth' command": AuthenticationWrongNumberOfArgsError, # some Redis server versions report invalid command syntax # in uppercase "wrong number of arguments " "for 'AUTH' command": AuthenticationWrongNumberOfArgsError, MODULE_LOAD_ERROR: ModuleError, MODULE_EXPORTS_DATA_TYPES_ERROR: ModuleError, NO_SUCH_MODULE_ERROR: ModuleError, MODULE_UNLOAD_NOT_POSSIBLE_ERROR: ModuleError, }, "EXECABORT": ExecAbortError, "LOADING": BusyLoadingError, "NOSCRIPT": NoScriptError, "READONLY": ReadOnlyError, "NOAUTH": AuthenticationError, "NOPERM": NoPermissionError, } def parse_error(self, response): "Parse an error response" error_code = response.split(" ")[0] if error_code in self.EXCEPTION_CLASSES: response = response[len(error_code) + 1 :] exception_class = self.EXCEPTION_CLASSES[error_code] if isinstance(exception_class, dict): exception_class = exception_class.get(response, ResponseError) return exception_class(response) return ResponseError(response) class SocketBuffer: def __init__(self, socket, socket_read_size, socket_timeout): self._sock = socket self.socket_read_size = socket_read_size self.socket_timeout = socket_timeout self._buffer = io.BytesIO() # number of bytes written to the buffer from the socket self.bytes_written = 0 # number of bytes read from the buffer self.bytes_read = 0 @property def length(self): return self.bytes_written - self.bytes_read def _read_from_socket(self, length=None, timeout=SENTINEL, raise_on_timeout=True): sock = self._sock socket_read_size = self.socket_read_size buf = self._buffer buf.seek(self.bytes_written) marker = 0 custom_timeout = timeout is not SENTINEL try: if custom_timeout: sock.settimeout(timeout) while True: data = self._sock.recv(socket_read_size) # an empty string indicates the server shutdown the socket if isinstance(data, bytes) and len(data) == 0: raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) buf.write(data) data_length = len(data) self.bytes_written += data_length marker += data_length if length is not None and length > marker: continue return True except socket.timeout: if raise_on_timeout: raise TimeoutError("Timeout reading from socket") return False except NONBLOCKING_EXCEPTIONS as ex: # if we're in nonblocking mode and the recv raises a # blocking error, simply return False indicating that # there's no data to be read. otherwise raise the # original exception. allowed = NONBLOCKING_EXCEPTION_ERROR_NUMBERS.get(ex.__class__, -1) if not raise_on_timeout and ex.errno == allowed: return False raise ConnectionError(f"Error while reading from socket: {ex.args}") finally: if custom_timeout: sock.settimeout(self.socket_timeout) def can_read(self, timeout): return bool(self.length) or self._read_from_socket( timeout=timeout, raise_on_timeout=False ) def read(self, length): length = length + 2 # make sure to read the \r\n terminator # make sure we've read enough data from the socket if length > self.length: self._read_from_socket(length - self.length) self._buffer.seek(self.bytes_read) data = self._buffer.read(length) self.bytes_read += len(data) # purge the buffer when we've consumed it all so it doesn't # grow forever if self.bytes_read == self.bytes_written: self.purge() return data[:-2] def readline(self): buf = self._buffer buf.seek(self.bytes_read) data = buf.readline() while not data.endswith(SYM_CRLF): # there's more data in the socket that we need self._read_from_socket() buf.seek(self.bytes_read) data = buf.readline() self.bytes_read += len(data) # purge the buffer when we've consumed it all so it doesn't # grow forever if self.bytes_read == self.bytes_written: self.purge() return data[:-2] def purge(self): self._buffer.seek(0) self._buffer.truncate() self.bytes_written = 0 self.bytes_read = 0 def close(self): try: self.purge() self._buffer.close() except Exception: # issue #633 suggests the purge/close somehow raised a # BadFileDescriptor error. Perhaps the client ran out of # memory or something else? It's probably OK to ignore # any error being raised from purge/close since we're # removing the reference to the instance below. pass self._buffer = None self._sock = None class PythonParser(BaseParser): "Plain Python parsing class" def __init__(self, socket_read_size): self.socket_read_size = socket_read_size self.encoder = None self._sock = None self._buffer = None def __del__(self): try: self.on_disconnect() except Exception: pass def on_connect(self, connection): "Called when the socket connects" self._sock = connection._sock self._buffer = SocketBuffer( self._sock, self.socket_read_size, connection.socket_timeout ) self.encoder = connection.encoder def on_disconnect(self): "Called when the socket disconnects" self._sock = None if self._buffer is not None: self._buffer.close() self._buffer = None self.encoder = None def can_read(self, timeout): return self._buffer and self._buffer.can_read(timeout) def read_response(self, disable_decoding=False): raw = self._buffer.readline() if not raw: raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) byte, response = raw[:1], raw[1:] if byte not in (b"-", b"+", b":", b"$", b""): raise InvalidResponse(f"Protocol Error: {raw!r}") # server returned an error if byte == b"-": response = response.decode("utf-8", errors="replace") error = self.parse_error(response) # if the error is a ConnectionError, raise immediately so the user # is notified if isinstance(error, ConnectionError): raise error # otherwise, we're dealing with a ResponseError that might belong # inside a pipeline response. the connection's read_response() # and/or the pipeline's execute() will raise this error if # necessary, so just return the exception instance here. return error # single value elif byte == b"+": pass # int value elif byte == b":": response = int(response) # bulk response elif byte == b"$": length = int(response) if length == -1: return None response = self._buffer.read(length) # multi-bulk response elif byte == b"": length = int(response) if length == -1: return None response = [ self.read_response(disable_decoding=disable_decoding) for i in range(length) ] if isinstance(response, bytes) and disable_decoding is False: response = self.encoder.decode(response) return response class HiredisParser(BaseParser): "Parser class for connections using Hiredis" def __init__(self, socket_read_size): if not HIREDIS_AVAILABLE: raise RedisError("Hiredis is not installed") self.socket_read_size = socket_read_size if HIREDIS_USE_BYTE_BUFFER: self._buffer = bytearray(socket_read_size) def __del__(self): try: self.on_disconnect() except Exception: pass def on_connect(self, connection, kwargs): self._sock = connection._sock self._socket_timeout = connection.socket_timeout kwargs = { "protocolError": InvalidResponse, "replyError": self.parse_error, } # hiredis < 0.1.3 doesn't support functions that create exceptions if not HIREDIS_SUPPORTS_CALLABLE_ERRORS: kwargs["replyError"] = ResponseError if connection.encoder.decode_responses: kwargs["encoding"] = connection.encoder.encoding if HIREDIS_SUPPORTS_ENCODING_ERRORS: kwargs["errors"] = connection.encoder.encoding_errors self._reader = hiredis.Reader(kwargs) self._next_response = False def on_disconnect(self): self._sock = None self._reader = None self._next_response = False def can_read(self, timeout): if not self._reader: raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) if self._next_response is False: self._next_response = self._reader.gets() if self._next_response is False: return self.read_from_socket(timeout=timeout, raise_on_timeout=False) return True def read_from_socket(self, timeout=SENTINEL, raise_on_timeout=True): sock = self._sock custom_timeout = timeout is not SENTINEL try: if custom_timeout: sock.settimeout(timeout) if HIREDIS_USE_BYTE_BUFFER: bufflen = self._sock.recv_into(self._buffer) if bufflen == 0: raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) self._reader.feed(self._buffer, 0, bufflen) else: buffer = self._sock.recv(self.socket_read_size) # an empty string indicates the server shutdown the socket if not isinstance(buffer, bytes) or len(buffer) == 0: raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) self._reader.feed(buffer) # data was read from the socket and added to the buffer. # return True to indicate that data was read. return True except socket.timeout: if raise_on_timeout: raise TimeoutError("Timeout reading from socket") return False except NONBLOCKING_EXCEPTIONS as ex: # if we're in nonblocking mode and the recv raises a # blocking error, simply return False indicating that # there's no data to be read. otherwise raise the # original exception. allowed = NONBLOCKING_EXCEPTION_ERROR_NUMBERS.get(ex.__class__, -1) if not raise_on_timeout and ex.errno == allowed: return False raise ConnectionError(f"Error while reading from socket: {ex.args}") finally: if custom_timeout: sock.settimeout(self._socket_timeout) def read_response(self, disable_decoding=False): if not self._reader: raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR) # _next_response might be cached from a can_read() call if self._next_response is not False: response = self._next_response self._next_response = False return response response = self._reader.gets() while response is False: self.read_from_socket() response = self._reader.gets() # if an older version of hiredis is installed, we need to attempt # to convert ResponseErrors to their appropriate types. if not HIREDIS_SUPPORTS_CALLABLE_ERRORS: if isinstance(response, ResponseError): response = self.parse_error(response.args[0]) elif ( isinstance(response, list) and response and isinstance(response[0], ResponseError) ): response[0] = self.parse_error(response[0].args[0]) # if the response is a ConnectionError or the response is a list and # the first item is a ConnectionError, raise it as something bad # happened if isinstance(response, ConnectionError): raise response elif ( isinstance(response, list) and response and isinstance(response[0], ConnectionError) ): raise response[0] return response if HIREDIS_AVAILABLE: DefaultParser = HiredisParser else: DefaultParser = PythonParser class Connection: "Manages TCP communication to and from a Redis server" def __init__( self, host="localhost", port=6379, db=0, password=None, socket_timeout=None, socket_connect_timeout=None, socket_keepalive=False, socket_keepalive_options=None, socket_type=0, retry_on_timeout=False, retry_on_error=[], encoding="utf-8", encoding_errors="strict", decode_responses=False, parser_class=DefaultParser, socket_read_size=65536, health_check_interval=0, client_name=None, username=None, retry=None, redis_connect_func=None, ): """ Initialize a new Connection. To specify a retry policy for specific errors, first set `retry_on_error` to a list of the error/s to retry on, then set `retry` to a valid `Retry` object. To retry on TimeoutError, `retry_on_timeout` can also be set to `True`. """ self.pid = os.getpid() self.host = host self.port = int(port) self.db = db self.username = username self.client_name = client_name self.password = password self.socket_timeout = socket_timeout self.socket_connect_timeout = socket_connect_timeout or socket_timeout self.socket_keepalive = socket_keepalive self.socket_keepalive_options = socket_keepalive_options or {} self.socket_type = socket_type self.retry_on_timeout = retry_on_timeout if retry_on_timeout: # Add TimeoutError to the errors list to retry on retry_on_error.append(TimeoutError) self.retry_on_error = retry_on_error if retry_on_error: if retry is None: self.retry = Retry(NoBackoff(), 1) else: # deep-copy the Retry object as it is mutable self.retry = copy.deepcopy(retry) # Update the retry's supported errors with the specified errors self.retry.update_supported_erros(retry_on_error) else: self.retry = Retry(NoBackoff(), 0) self.health_check_interval = health_check_interval self.next_health_check = 0 self.redis_connect_func = redis_connect_func self.encoder = Encoder(encoding, encoding_errors, decode_responses) self._sock = None self._socket_read_size = socket_read_size self.set_parser(parser_class) self._connect_callbacks = [] self._buffer_cutoff = 6000 def __repr__(self): repr_args = ",".join([f"{k}={v}" for k, v in self.repr_pieces()]) return f"{self.__class__.__name__}<{repr_args}>" def repr_pieces(self): pieces = [("host", self.host), ("port", self.port), ("db", self.db)] if self.client_name: pieces.append(("client_name", self.client_name)) return pieces def __del__(self): try: self.disconnect() except Exception: pass def register_connect_callback(self, callback): self._connect_callbacks.append(weakref.WeakMethod(callback)) def clear_connect_callbacks(self): self._connect_callbacks = [] def set_parser(self, parser_class): """ Creates a new instance of parser_class with socket size: _socket_read_size and assigns it to the parser for the connection :param parser_class: The required parser class """ self._parser = parser_class(socket_read_size=self._socket_read_size) def connect(self): "Connects to the Redis server if not already connected" if self._sock: return try: sock = self.retry.call_with_retry( lambda: self._connect(), lambda error: self.disconnect(error) ) except socket.timeout: raise TimeoutError("Timeout connecting to server") except OSError as e: raise ConnectionError(self._error_message(e)) self._sock = sock try: if self.redis_connect_func is None: # Use the default on_connect function self.on_connect() else: # Use the passed function redis_connect_func self.redis_connect_func(self) except RedisError: # clean up after any error in on_connect self.disconnect() raise # run any user callbacks. right now the only internal callback # is for pubsub channel/pattern resubscription for ref in self._connect_callbacks: callback = ref() if callback: callback(self) def _connect(self): "Create a TCP socket connection" # we want to mimic what socket.create_connection does to support # ipv4/ipv6, but we want to set options prior to calling # socket.connect() err = None for res in socket.getaddrinfo( self.host, self.port, self.socket_type, socket.SOCK_STREAM ): family, socktype, proto, canonname, socket_address = res sock = None try: sock = socket.socket(family, socktype, proto) # TCP_NODELAY sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) # TCP_KEEPALIVE if self.socket_keepalive: sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) for k, v in self.socket_keepalive_options.items(): sock.setsockopt(socket.IPPROTO_TCP, k, v) # set the socket_connect_timeout before we connect sock.settimeout(self.socket_connect_timeout) # connect sock.connect(socket_address) # set the socket_timeout now that we're connected sock.settimeout(self.socket_timeout) return sock except OSError as _: err = _ if sock is not None: sock.close() if err is not None: raise err raise OSError("socket.getaddrinfo returned an empty list") def _error_message(self, exception): # args for socket.error can either be (errno, "message") # or just "message" if len(exception.args) == 1: try: return f"Error connecting to {self.host}:{self.port}. \ {exception.args[0]}." except AttributeError: return f"Connection Error: {exception.args[0]}" else: try: return ( f"Error {exception.args[0]} connecting to " f"{self.host}:{self.port}. {exception.args[1]}." ) except AttributeError: return f"Connection Error: {exception.args[0]}" def on_connect(self): "Initialize the connection, authenticate and select a database" self._parser.on_connect(self) # if username and/or password are set, authenticate if self.username or self.password: if self.username: auth_args = (self.username, self.password or "") else: auth_args = (self.password,) # avoid checking health here -- PING will fail if we try # to check the health prior to the AUTH self.send_command("AUTH", auth_args, check_health=False) try: auth_response = self.read_response() except AuthenticationWrongNumberOfArgsError: # a username and password were specified but the Redis # server seems to be < 6.0.0 which expects a single password # arg. retry auth with just the password. # https://github.com/andymccurdy/redis-py/issues/1274 self.send_command("AUTH", self.password, check_health=False) auth_response = self.read_response() if str_if_bytes(auth_response) != "OK": raise AuthenticationError("Invalid Username or Password") # if a client_name is given, set it if self.client_name: self.send_command("CLIENT", "SETNAME", self.client_name) if str_if_bytes(self.read_response()) != "OK": raise ConnectionError("Error setting client name") # if a database is specified, switch to it if self.db: self.send_command("SELECT", self.db) if str_if_bytes(self.read_response()) != "OK": raise ConnectionError("Invalid Database") def disconnect(self, args): "Disconnects from the Redis server" self._parser.on_disconnect() if self._sock is None: return if os.getpid() == self.pid: try: self._sock.shutdown(socket.SHUT_RDWR) except OSError: pass try: self._sock.close() except OSError: pass self._sock = None def _send_ping(self): """Send PING, expect PONG in return""" self.send_command("PING", check_health=False) if str_if_bytes(self.read_response()) != "PONG": raise ConnectionError("Bad response from PING health check") def _ping_failed(self, error): """Function to call when PING fails""" self.disconnect() def check_health(self): """Check the health of the connection with a PING/PONG""" if self.health_check_interval and time() > self.next_health_check: self.retry.call_with_retry(self._send_ping, self._ping_failed) def send_packed_command(self, command, check_health=True): """Send an already packed command to the Redis server""" if not self._sock: self.connect() # guard against health check recursion if check_health: self.check_health() try: if isinstance(command, str): command = [command] for item in command: self._sock.sendall(item) except socket.timeout: self.disconnect() raise TimeoutError("Timeout writing to socket") except OSError as e: self.disconnect() if len(e.args) == 1: errno, errmsg = "UNKNOWN", e.args[0] else: errno = e.args[0] errmsg = e.args[1] raise ConnectionError(f"Error {errno} while writing to socket. {errmsg}.") except BaseException: self.disconnect() raise def send_command(self, args, *kwargs): """Pack and send a command to the Redis server""" self.send_packed_command( self.pack_command(args), check_health=kwargs.get("check_health", True) ) def can_read(self, timeout=0): """Poll the socket to see if there's data that can be read.""" sock = self._sock if not sock: self.connect() return self._parser.can_read(timeout) def read_response(self, disable_decoding=False): """Read the response from a previously sent command""" try: hosterr = f"{self.host}:{self.port}" except AttributeError: hosterr = "connection" try: response = self._parser.read_response(disable_decoding=disable_decoding) except socket.timeout: self.disconnect() raise TimeoutError(f"Timeout reading from {hosterr}") except OSError as e: self.disconnect() raise ConnectionError(f"Error while reading from {hosterr}" f" : {e.args}") except BaseException: self.disconnect() raise if self.health_check_interval: self.next_health_check = time() + self.health_check_interval if isinstance(response, ResponseError): raise response return response def pack_command(self, args): """Pack a series of arguments into the Redis protocol""" output = [] # the client might have included 1 or more literal arguments in # the command name, e.g., 'CONFIG GET'. The Redis server expects these # arguments to be sent separately, so split the first argument # manually. These arguments should be bytestrings so that they are # not encoded. if isinstance(args[0], str): args = tuple(args[0].encode().split()) + args[1:] elif b" " in args[0]: args = tuple(args[0].split()) + args[1:] buff = SYM_EMPTY.join((SYM_STAR, str(len(args)).encode(), SYM_CRLF)) buffer_cutoff = self._buffer_cutoff for arg in map(self.encoder.encode, args): # to avoid large string mallocs, chunk the command into the # output list if we're sending large values or memoryviews arg_length = len(arg) if ( len(buff) > buffer_cutoff or arg_length > buffer_cutoff or isinstance(arg, memoryview) ): buff = SYM_EMPTY.join( (buff, SYM_DOLLAR, str(arg_length).encode(), SYM_CRLF) ) output.append(buff) output.append(arg) buff = SYM_CRLF else: buff = SYM_EMPTY.join( ( buff, SYM_DOLLAR, str(arg_length).encode(), SYM_CRLF, arg, SYM_CRLF, ) ) output.append(buff) return output def pack_commands(self, commands): """Pack multiple commands into the Redis protocol""" output = [] pieces = [] buffer_length = 0 buffer_cutoff = self._buffer_cutoff for cmd in commands: for chunk in self.pack_command(cmd): chunklen = len(chunk) if ( buffer_length > buffer_cutoff or chunklen > buffer_cutoff or isinstance(chunk, memoryview) ): output.append(SYM_EMPTY.join(pieces)) buffer_length = 0 pieces = [] if chunklen > buffer_cutoff or isinstance(chunk, memoryview): output.append(chunk) else: pieces.append(chunk) buffer_length += chunklen if pieces: output.append(SYM_EMPTY.join(pieces)) return output class SSLConnection(Connection): """Manages SSL connections to and from the Redis server(s). This class extends the Connection class, adding SSL functionality, and making use of ssl.SSLContext (https://docs.python.org/3/library/ssl.html#ssl.SSLContext) """ # noqa def __init__( self, ssl_keyfile=None, ssl_certfile=None, ssl_cert_reqs="required", ssl_ca_certs=None, ssl_check_hostname=False, ssl_ca_path=None, ssl_password=None, ssl_validate_ocsp=False, ssl_validate_ocsp_stapled=False, ssl_ocsp_context=None, ssl_ocsp_expected_cert=None, kwargs, ): """Constructor Args: ssl_keyfile: Path to an ssl private key. Defaults to None. ssl_certfile: Path to an ssl certificate. Defaults to None. ssl_cert_reqs: The string value for the SSLContext.verify_mode (none, optional, required). Defaults to "required". ssl_ca_certs: The path to a file of concatenated CA certificates in PEM format. Defaults to None. ssl_check_hostname: If set, match the hostname during the SSL handshake. Defaults to False. ssl_ca_path: The path to a directory containing several CA certificates in PEM format. Defaults to None. ssl_password: Password for unlocking an encrypted private key. Defaults to None. ssl_validate_ocsp: If set, perform a full ocsp validation (i.e not a stapled verification) ssl_validate_ocsp_stapled: If set, perform a validation on a stapled ocsp response ssl_ocsp_context: A fully initialized OpenSSL.SSL.Context object to be used in verifying the ssl_ocsp_expected_cert ssl_ocsp_expected_cert: A PEM armoured string containing the expected certificate to be returned from the ocsp verification service. Raises: RedisError """ # noqa if not ssl_available: raise RedisError("Python wasn't built with SSL support") super().__init__(kwargs) self.keyfile = ssl_keyfile self.certfile = ssl_certfile if ssl_cert_reqs is None: ssl_cert_reqs = ssl.CERT_NONE elif isinstance(ssl_cert_reqs, str): CERT_REQS = { "none": ssl.CERT_NONE, "optional": ssl.CERT_OPTIONAL, "required": ssl.CERT_REQUIRED, } if ssl_cert_reqs not in CERT_REQS: raise RedisError( f"Invalid SSL Certificate Requirements Flag: {ssl_cert_reqs}" ) ssl_cert_reqs = CERT_REQS[ssl_cert_reqs] self.cert_reqs = ssl_cert_reqs self.ca_certs = ssl_ca_certs self.ca_path = ssl_ca_path self.check_hostname = ssl_check_hostname self.certificate_password = ssl_password self.ssl_validate_ocsp = ssl_validate_ocsp self.ssl_validate_ocsp_stapled = ssl_validate_ocsp_stapled self.ssl_ocsp_context = ssl_ocsp_context self.ssl_ocsp_expected_cert = ssl_ocsp_expected_cert def _connect(self): "Wrap the socket with SSL support" sock = super()._connect() context = ssl.create_default_context() context.check_hostname = self.check_hostname context.verify_mode = self.cert_reqs if self.certfile or self.keyfile: context.load_cert_chain( certfile=self.certfile, keyfile=self.keyfile, password=self.certificate_password, ) if self.ca_certs is not None or self.ca_path is not None: context.load_verify_locations(cafile=self.ca_certs, capath=self.ca_path) sslsock = context.wrap_socket(sock, server_hostname=self.host) if self.ssl_validate_ocsp is True and CRYPTOGRAPHY_AVAILABLE is False: raise RedisError("cryptography is not installed.") if self.ssl_validate_ocsp_stapled and self.ssl_validate_ocsp: raise RedisError( "Either an OCSP staple or pure OCSP connection must be validated " "- not both." ) # validation for the stapled case if self.ssl_validate_ocsp_stapled: import OpenSSL from .ocsp import ocsp_staple_verifier # if a context is provided use it - otherwise, a basic context if self.ssl_ocsp_context is None: staple_ctx = OpenSSL.SSL.Context(OpenSSL.SSL.SSLv23_METHOD) staple_ctx.use_certificate_file(self.certfile) staple_ctx.use_privatekey_file(self.keyfile) else: staple_ctx = self.ssl_ocsp_context staple_ctx.set_ocsp_client_callback( ocsp_staple_verifier, self.ssl_ocsp_expected_cert, ) # need another socket con = OpenSSL.SSL.Connection(staple_ctx, socket.socket()) con.request_ocsp() con.connect((self.host, self.port)) con.do_handshake() con.shutdown() return sslsock # pure ocsp validation if self.ssl_validate_ocsp is True and CRYPTOGRAPHY_AVAILABLE: from .ocsp import OCSPVerifier o = OCSPVerifier(sslsock, self.host, self.port, self.ca_certs) if o.is_valid(): return sslsock else: raise ConnectionError("ocsp validation error") return sslsock class UnixDomainSocketConnection(Connection): def __init__( self, path="", db=0, username=None, password=None, socket_timeout=None, encoding="utf-8", encoding_errors="strict", decode_responses=False, retry_on_timeout=False, retry_on_error=[], parser_class=DefaultParser, socket_read_size=65536, health_check_interval=0, client_name=None, retry=None, redis_connect_func=None, ): """ Initialize a new UnixDomainSocketConnection. To specify a retry policy for specific errors, first set `retry_on_error` to a list of the error/s to retry on, then set `retry` to a valid `Retry` object. To retry on TimeoutError, `retry_on_timeout` can also be set to `True`. """ self.pid = os.getpid() self.path = path self.db = db self.username = username self.client_name = client_name self.password = password self.socket_timeout = socket_timeout self.retry_on_timeout = retry_on_timeout if retry_on_timeout: # Add TimeoutError to the errors list to retry on retry_on_error.append(TimeoutError) self.retry_on_error = retry_on_error if self.retry_on_error: if retry is None: self.retry = Retry(NoBackoff(), 1) else: # deep-copy the Retry object as it is mutable self.retry = copy.deepcopy(retry) # Update the retry's supported errors with the specified errors self.retry.update_supported_erros(retry_on_error) else: self.retry = Retry(NoBackoff(), 0) self.health_check_interval = health_check_interval self.next_health_check = 0 self.redis_connect_func = redis_connect_func self.encoder = Encoder(encoding, encoding_errors, decode_responses) self._sock = None self._socket_read_size = socket_read_size self.set_parser(parser_class) self._connect_callbacks = [] self._buffer_cutoff = 6000 def repr_pieces(self): pieces = [ ("path", self.path), ("db", self.db), ] if self.client_name: pieces.append(("client_name", self.client_name)) return pieces def _connect(self): "Create a Unix domain socket connection" sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.settimeout(self.socket_timeout) sock.connect(self.path) return sock def _error_message(self, exception): # args for socket.error can either be (errno, "message") # or just "message" if len(exception.args) == 1: return f"Error connecting to unix socket: {self.path}. {exception.args[0]}." else: return ( f"Error {exception.args[0]} connecting to unix socket: " f"{self.path}. {exception.args[1]}." ) FALSE_STRINGS = ("0", "F", "FALSE", "N", "NO") def to_bool(value): if value is None or value == "": return None if isinstance(value, str) and value.upper() in FALSE_STRINGS: return False return bool(value) URL_QUERY_ARGUMENT_PARSERS = { "db": int, "socket_timeout": float, "socket_connect_timeout": float, "socket_keepalive": to_bool, "retry_on_timeout": to_bool, "retry_on_error": list, "max_connections": int, "health_check_interval": int, "ssl_check_hostname": to_bool, } def parse_url(url): url = urlparse(url) kwargs = {} for name, value in parse_qs(url.query).items(): if value and len(value) > 0: value = unquote(value[0]) parser = URL_QUERY_ARGUMENT_PARSERS.get(name) if parser: try: kwargs[name] = parser(value) except (TypeError, ValueError): raise ValueError(f"Invalid value for `{name}` in connection URL.") else: kwargs[name] = value if url.username: kwargs["username"] = unquote(url.username) if url.password: kwargs["password"] = unquote(url.password) # We only support redis://, rediss:// and unix:// schemes. if url.scheme == "unix": if url.path: kwargs["path"] = unquote(url.path) kwargs["connection_class"] = UnixDomainSocketConnection elif url.scheme in ("redis", "rediss"): if url.hostname: kwargs["host"] = unquote(url.hostname) if url.port: kwargs["port"] = int(url.port) # If there's a path argument, use it as the db argument if a # querystring value wasn't specified if url.path and "db" not in kwargs: try: kwargs["db"] = int(unquote(url.path).replace("/", "")) except (AttributeError, ValueError): pass if url.scheme == "rediss": kwargs["connection_class"] = SSLConnection else: raise ValueError( "Redis URL must specify one of the following " "schemes (redis://, rediss://, unix://)" ) return kwargs class ConnectionPool: """ Create a connection pool. ``If max_connections`` is set, then this object raises :py:class:`~redis.exceptions.ConnectionError` when the pool's limit is reached. By default, TCP connections are created unless ``connection_class`` is specified. Use class:`.UnixDomainSocketConnection` for unix sockets. Any additional keyword arguments are passed to the constructor of ``connection_class``. """ @classmethod def from_url(cls, url, kwargs): """ Return a connection pool configured from the given URL. For example:: redis://[[username]:[password]]@localhost:6379/0 rediss://[[username]:[password]]@localhost:6379/0 unix://[[username]:[password]]@/path/to/socket.sock?db=0 Three URL schemes are supported: - `redis://` creates a TCP socket connection. See more at: <https://www.iana.org/assignments/uri-schemes/prov/redis> - `rediss://` creates a SSL wrapped TCP socket connection. See more at: <https://www.iana.org/assignments/uri-schemes/prov/rediss> - ``unix://``: creates a Unix Domain Socket connection. The username, password, hostname, path and all querystring values are passed through urllib.parse.unquote in order to replace any percent-encoded values with their corresponding characters. There are several ways to specify a database number. The first value found will be used: 1. A ``db`` querystring option, e.g. redis://localhost?db=0 2. If using the redis:// or rediss:// schemes, the path argument of the url, e.g. redis://localhost/0 3. A ``db`` keyword argument to this function. If none of these options are specified, the default db=0 is used. All querystring options are cast to their appropriate Python types. Boolean arguments can be specified with string values "True"/"False" or "Yes"/"No". Values that cannot be properly cast cause a ``ValueError`` to be raised. Once parsed, the querystring arguments and keyword arguments are passed to the ``ConnectionPool``'s class initializer. In the case of conflicting arguments, querystring arguments always win. """ url_options = parse_url(url) if "connection_class" in kwargs: url_options["connection_class"] = kwargs["connection_class"] kwargs.update(url_options) return cls(kwargs) def __init__( self, connection_class=Connection, max_connections=None, connection_kwargs ): max_connections = max_connections or 2 31 if not isinstance(max_connections, int) or max_connections < 0: raise ValueError('"max_connections" must be a positive integer') self.connection_class = connection_class self.connection_kwargs = connection_kwargs self.max_connections = max_connections # a lock to protect the critical section in _checkpid(). # this lock is acquired when the process id changes, such as # after a fork. during this time, multiple threads in the child # process could attempt to acquire this lock. the first thread # to acquire the lock will reset the data structures and lock # object of this pool. subsequent threads acquiring this lock # will notice the first thread already did the work and simply # release the lock. self._fork_lock = threading.Lock() self.reset() def __repr__(self): return ( f"{type(self).__name__}" f"<{repr(self.connection_class(*self.connection_kwargs))}>" ) def reset(self): self._lock = threading.Lock() self._created_connections = 0 self._available_connections = [] self._in_use_connections = set() # this must be the last operation in this method. while reset() is # called when holding _fork_lock, other threads in this process # can call _checkpid() which compares self.pid and os.getpid() without # holding any lock (for performance reasons). keeping this assignment # as the last operation ensures that those other threads will also # notice a pid difference and block waiting for the first thread to # release _fork_lock. when each of these threads eventually acquire # _fork_lock, they will notice that another thread already called # reset() and they will immediately release _fork_lock and continue on. self.pid = os.getpid() def _checkpid(self): # _checkpid() attempts to keep ConnectionPool fork-safe on modern # systems. this is called by all ConnectionPool methods that # manipulate the pool's state such as get_connection() and release(). # # _checkpid() determines whether the process has forked by comparing # the current process id to the process id saved on the ConnectionPool # instance. if these values are the same, _checkpid() simply returns. # # when the process ids differ, _checkpid() assumes that the process # has forked and that we're now running in the child process. the child # process cannot use the parent's file descriptors (e.g., sockets). # therefore, when _checkpid() sees the process id change, it calls # reset() in order to reinitialize the child's ConnectionPool. this # will cause the child to make all new connection objects. # # _checkpid() is protected by self._fork_lock to ensure that multiple # threads in the child process do not call reset() multiple times. # # there is an extremely small chance this could fail in the following # scenario: # 1. process A calls _checkpid() for the first time and acquires # self._fork_lock. # 2. while holding self._fork_lock, process A forks (the fork() # could happen in a different thread owned by process A) # 3. process B (the forked child process) inherits the # ConnectionPool's state from the parent. that state includes # a locked _fork_lock. process B will not be notified when # process A releases the _fork_lock and will thus never be # able to acquire the _fork_lock. # # to mitigate this possible deadlock, _checkpid() will only wait 5 # seconds to acquire _fork_lock. if _fork_lock cannot be acquired in # that time it is assumed that the child is deadlocked and a # redis.ChildDeadlockedError error is raised. if self.pid != os.getpid(): acquired = self._fork_lock.acquire(timeout=5) if not acquired: raise ChildDeadlockedError # reset() the instance for the new process if another thread # hasn't already done so try: if self.pid != os.getpid(): self.reset() finally: self._fork_lock.release() def get_connection(self, command_name, keys, options): "Get a connection from the pool" self._checkpid() with self._lock: try: connection = self._available_connections.pop() except IndexError: connection = self.make_connection() self._in_use_connections.add(connection) try: # ensure this connection is connected to Redis connection.connect() # connections that the pool provides should be ready to send # a command. if not, the connection was either returned to the # pool before all data has been read or the socket has been # closed. either way, reconnect and verify everything is good. try: if connection.can_read(): raise ConnectionError("Connection has data") except (ConnectionError, OSError): connection.disconnect() connection.connect() if connection.can_read(): raise ConnectionError("Connection not ready") except BaseException: # release the connection back to the pool so that we don't # leak it self.release(connection) raise return connection def get_encoder(self): "Return an encoder based on encoding settings" kwargs = self.connection_kwargs return Encoder( encoding=kwargs.get("encoding", "utf-8"), encoding_errors=kwargs.get("encoding_errors", "strict"), decode_responses=kwargs.get("decode_responses", False), ) def make_connection(self): "Create a new connection" if self._created_connections >= self.max_connections: raise ConnectionError("Too many connections") self._created_connections += 1 return self.connection_class(self.connection_kwargs) def release(self, connection): "Releases the connection back to the pool" self._checkpid() with self._lock: try: self._in_use_connections.remove(connection) except KeyError: # Gracefully fail when a connection is returned to this pool # that the pool doesn't actually own pass if self.owns_connection(connection): self._available_connections.append(connection) else: # pool doesn't own this connection. do not add it back # to the pool and decrement the count so that another # connection can take its place if needed self._created_connections -= 1 connection.disconnect() return def owns_connection(self, connection): return connection.pid == self.pid def disconnect(self, inuse_connections=True): """ Disconnects connections in the pool If ``inuse_connections`` is True, disconnect connections that are current in use, potentially by other threads. Otherwise only disconnect connections that are idle in the pool. """ self._checkpid() with self._lock: if inuse_connections: connections = chain( self._available_connections, self._in_use_connections ) else: connections = self._available_connections for connection in connections: connection.disconnect() class BlockingConnectionPool(ConnectionPool): """ Thread-safe blocking connection pool:: >>> from redis.client import Redis >>> client = Redis(connection_pool=BlockingConnectionPool()) It performs the same function as the default :py:class:`~redis.ConnectionPool` implementation, in that, it maintains a pool of reusable connections that can be shared by multiple redis clients (safely across threads if required). The difference is that, in the event that a client tries to get a connection from the pool when all of connections are in use, rather than raising a :py:class:`~redis.ConnectionError` (as the default :py:class:`~redis.ConnectionPool` implementation does), it makes the client wait ("blocks") for a specified number of seconds until a connection becomes available. Use ``max_connections`` to increase / decrease the pool size:: >>> pool = BlockingConnectionPool(max_connections=10) Use ``timeout`` to tell it either how many seconds to wait for a connection to become available, or to block forever: >>> # Block forever. >>> pool = BlockingConnectionPool(timeout=None) >>> # Raise a ``ConnectionError`` after five seconds if a connection is >>> # not available. >>> pool = BlockingConnectionPool(timeout=5) """ def __init__( self, max_connections=50, timeout=20, connection_class=Connection, queue_class=LifoQueue, connection_kwargs, ): self.queue_class = queue_class self.timeout = timeout super().__init__( connection_class=connection_class, max_connections=max_connections, connection_kwargs, ) def reset(self): # Create and fill up a thread safe queue with ``None`` values. self.pool = self.queue_class(self.max_connections) while True: try: self.pool.put_nowait(None) except Full: break # Keep a list of actual connection instances so that we can # disconnect them later. self._connections = [] # this must be the last operation in this method. while reset() is # called when holding _fork_lock, other threads in this process # can call _checkpid() which compares self.pid and os.getpid() without # holding any lock (for performance reasons). keeping this assignment # as the last operation ensures that those other threads will also # notice a pid difference and block waiting for the first thread to # release _fork_lock. when each of these threads eventually acquire # _fork_lock, they will notice that another thread already called # reset() and they will immediately release _fork_lock and continue on. self.pid = os.getpid() def make_connection(self): "Make a fresh connection." connection = self.connection_class(*self.connection_kwargs) self._connections.append(connection) return connection def get_connection(self, command_name, keys, **options): """ Get a connection, blocking for ``self.timeout`` until a connection is available from the pool. If the connection returned is ``None`` then creates a new connection. Because we use a last-in first-out queue, the existing connections (having been returned to the pool after the initial ``None`` values were added) will be returned before ``None`` values. This means we only create new connections when we need to, i.e.: the actual number of connections will only increase in response to demand. """ # Make sure we haven't changed process. self._checkpid() # Try and get a connection from the pool. If one isn't available within # self.timeout then raise a ``ConnectionError``. connection = None try: connection = self.pool.get(block=True, timeout=self.timeout) except Empty: # Note that this is not caught by the redis client and will be # raised unless handled by application code. If you want never to raise ConnectionError("No connection available.") # If the ``connection`` is actually ``None`` then that's a cue to make # a new connection to add to the pool. if connection is None: connection = self.make_connection() try: # ensure this connection is connected to Redis connection.connect() # connections that the pool provides should be ready to send # a command. if not, the connection was either returned to the # pool before all data has been read or the socket has been # closed. either way, reconnect and verify everything is good. try: if connection.can_read(): raise ConnectionError("Connection has data") except ConnectionError: connection.disconnect() connection.connect() if connection.can_read(): raise ConnectionError("Connection not ready") except BaseException: # release the connection back to the pool so that we don't leak it self.release(connection) raise return connection def release(self, connection): "Releases the connection back to the pool." # Make sure we haven't changed process. self._checkpid() if not self.owns_connection(connection): # pool doesn't own this connection. do not add it back # to the pool. instead add a None value which is a placeholder # that will cause the pool to recreate the connection if # its needed. connection.disconnect() self.pool.put_nowait(None) return # Put the connection back into the pool. try: self.pool.put_nowait(connection) except Full: # perhaps the pool has been reset() after a fork? regardless, # we don't want this connection pass def disconnect(self): "Disconnects all connections in the pool." self._checkpid() for connection in self._connections: connection.disconnect()	59.14 KB	2025-05-25 12:26
📄 crc.py from binascii import crc_hqx # Redis Cluster's key space is divided into 16384 slots. # For more information see: https://github.com/redis/redis/issues/2576 REDIS_CLUSTER_HASH_SLOTS = 16384 __all__ = ["key_slot", "REDIS_CLUSTER_HASH_SLOTS"] def key_slot(key, bucket=REDIS_CLUSTER_HASH_SLOTS): """Calculate key slot for a given key. See Keys distribution model in https://redis.io/topics/cluster-spec :param key - bytes :param bucket - int """ start = key.find(b"{") if start > -1: end = key.find(b"}", start + 1) if end > -1 and end != start + 1: key = key[start + 1 : end] return crc_hqx(key, 0) % bucket	670 B	2025-05-25 12:26
📄 exceptions.py "Core exceptions raised by the Redis client" class RedisError(Exception): pass class ConnectionError(RedisError): pass class TimeoutError(RedisError): pass class AuthenticationError(ConnectionError): pass class AuthorizationError(ConnectionError): pass class BusyLoadingError(ConnectionError): pass class InvalidResponse(RedisError): pass class ResponseError(RedisError): pass class DataError(RedisError): pass class PubSubError(RedisError): pass class WatchError(RedisError): pass class NoScriptError(ResponseError): pass class ExecAbortError(ResponseError): pass class ReadOnlyError(ResponseError): pass class NoPermissionError(ResponseError): pass class ModuleError(ResponseError): pass class LockError(RedisError, ValueError): "Errors acquiring or releasing a lock" # NOTE: For backwards compatibility, this class derives from ValueError. # This was originally chosen to behave like threading.Lock. pass class LockNotOwnedError(LockError): "Error trying to extend or release a lock that is (no longer) owned" pass class ChildDeadlockedError(Exception): "Error indicating that a child process is deadlocked after a fork()" pass class AuthenticationWrongNumberOfArgsError(ResponseError): """ An error to indicate that the wrong number of args were sent to the AUTH command """ pass class RedisClusterException(Exception): """ Base exception for the RedisCluster client """ pass class ClusterError(RedisError): """ Cluster errors occurred multiple times, resulting in an exhaustion of the command execution TTL """ pass class ClusterDownError(ClusterError, ResponseError): """ Error indicated CLUSTERDOWN error received from cluster. By default Redis Cluster nodes stop accepting queries if they detect there is at least a hash slot uncovered (no available node is serving it). This way if the cluster is partially down (for example a range of hash slots are no longer covered) the entire cluster eventually becomes unavailable. It automatically returns available as soon as all the slots are covered again. """ def __init__(self, resp): self.args = (resp,) self.message = resp class AskError(ResponseError): """ Error indicated ASK error received from cluster. When a slot is set as MIGRATING, the node will accept all queries that pertain to this hash slot, but only if the key in question exists, otherwise the query is forwarded using a -ASK redirection to the node that is target of the migration. src node: MIGRATING to dst node get > ASK error ask dst node > ASKING command dst node: IMPORTING from src node asking command only affects next command any op will be allowed after asking command """ def __init__(self, resp): """should only redirect to master node""" self.args = (resp,) self.message = resp slot_id, new_node = resp.split(" ") host, port = new_node.rsplit(":", 1) self.slot_id = int(slot_id) self.node_addr = self.host, self.port = host, int(port) class TryAgainError(ResponseError): """ Error indicated TRYAGAIN error received from cluster. Operations on keys that don't exist or are - during resharding - split between the source and destination nodes, will generate a -TRYAGAIN error. """ def __init__(self, args, *kwargs): pass class ClusterCrossSlotError(ResponseError): """ Error indicated CROSSSLOT error received from cluster. A CROSSSLOT error is generated when keys in a request don't hash to the same slot. """ message = "Keys in request don't hash to the same slot" class MovedError(AskError): """ Error indicated MOVED error received from cluster. A request sent to a node that doesn't serve this key will be replayed with a MOVED error that points to the correct node. """ pass class MasterDownError(ClusterDownError): """ Error indicated MASTERDOWN error received from cluster. Link with MASTER is down and replica-serve-stale-data is set to 'no'. """ pass class SlotNotCoveredError(RedisClusterException): """ This error only happens in the case where the connection pool will try to fetch what node that is covered by a given slot. If this error is raised the client should drop the current node layout and attempt to reconnect and refresh the node layout again """ pass	4.54 KB	2025-05-25 12:26
📄 lock.py import threading import time as mod_time import uuid from types import SimpleNamespace from redis.exceptions import LockError, LockNotOwnedError class Lock: """ A shared, distributed Lock. Using Redis for locking allows the Lock to be shared across processes and/or machines. It's left to the user to resolve deadlock issues and make sure multiple clients play nicely together. """ lua_release = None lua_extend = None lua_reacquire = None # KEYS[1] - lock name # ARGV[1] - token # return 1 if the lock was released, otherwise 0 LUA_RELEASE_SCRIPT = """ local token = redis.call('get', KEYS[1]) if not token or token ~= ARGV[1] then return 0 end redis.call('del', KEYS[1]) return 1 """ # KEYS[1] - lock name # ARGV[1] - token # ARGV[2] - additional milliseconds # ARGV[3] - "0" if the additional time should be added to the lock's # existing ttl or "1" if the existing ttl should be replaced # return 1 if the locks time was extended, otherwise 0 LUA_EXTEND_SCRIPT = """ local token = redis.call('get', KEYS[1]) if not token or token ~= ARGV[1] then return 0 end local expiration = redis.call('pttl', KEYS[1]) if not expiration then expiration = 0 end if expiration < 0 then return 0 end local newttl = ARGV[2] if ARGV[3] == "0" then newttl = ARGV[2] + expiration end redis.call('pexpire', KEYS[1], newttl) return 1 """ # KEYS[1] - lock name # ARGV[1] - token # ARGV[2] - milliseconds # return 1 if the locks time was reacquired, otherwise 0 LUA_REACQUIRE_SCRIPT = """ local token = redis.call('get', KEYS[1]) if not token or token ~= ARGV[1] then return 0 end redis.call('pexpire', KEYS[1], ARGV[2]) return 1 """ def __init__( self, redis, name, timeout=None, sleep=0.1, blocking=True, blocking_timeout=None, thread_local=True, ): """ Create a new Lock instance named ``name`` using the Redis client supplied by ``redis``. ``timeout`` indicates a maximum life for the lock in seconds. By default, it will remain locked until release() is called. ``timeout`` can be specified as a float or integer, both representing the number of seconds to wait. ``sleep`` indicates the amount of time to sleep in seconds per loop iteration when the lock is in blocking mode and another client is currently holding the lock. ``blocking`` indicates whether calling ``acquire`` should block until the lock has been acquired or to fail immediately, causing ``acquire`` to return False and the lock not being acquired. Defaults to True. Note this value can be overridden by passing a ``blocking`` argument to ``acquire``. ``blocking_timeout`` indicates the maximum amount of time in seconds to spend trying to acquire the lock. A value of ``None`` indicates continue trying forever. ``blocking_timeout`` can be specified as a float or integer, both representing the number of seconds to wait. ``thread_local`` indicates whether the lock token is placed in thread-local storage. By default, the token is placed in thread local storage so that a thread only sees its token, not a token set by another thread. Consider the following timeline: time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds. thread-1 sets the token to "abc" time: 1, thread-2 blocks trying to acquire `my-lock` using the Lock instance. time: 5, thread-1 has not yet completed. redis expires the lock key. time: 5, thread-2 acquired `my-lock` now that it's available. thread-2 sets the token to "xyz" time: 6, thread-1 finishes its work and calls release(). if the token is not stored in thread local storage, then thread-1 would see the token value as "xyz" and would be able to successfully release the thread-2's lock. In some use cases it's necessary to disable thread local storage. For example, if you have code where one thread acquires a lock and passes that lock instance to a worker thread to release later. If thread local storage isn't disabled in this case, the worker thread won't see the token set by the thread that acquired the lock. Our assumption is that these cases aren't common and as such default to using thread local storage. """ self.redis = redis self.name = name self.timeout = timeout self.sleep = sleep self.blocking = blocking self.blocking_timeout = blocking_timeout self.thread_local = bool(thread_local) self.local = threading.local() if self.thread_local else SimpleNamespace() self.local.token = None self.register_scripts() def register_scripts(self): cls = self.__class__ client = self.redis if cls.lua_release is None: cls.lua_release = client.register_script(cls.LUA_RELEASE_SCRIPT) if cls.lua_extend is None: cls.lua_extend = client.register_script(cls.LUA_EXTEND_SCRIPT) if cls.lua_reacquire is None: cls.lua_reacquire = client.register_script(cls.LUA_REACQUIRE_SCRIPT) def __enter__(self): if self.acquire(): return self raise LockError("Unable to acquire lock within the time specified") def __exit__(self, exc_type, exc_value, traceback): self.release() def acquire(self, blocking=None, blocking_timeout=None, token=None): """ Use Redis to hold a shared, distributed lock named ``name``. Returns True once the lock is acquired. If ``blocking`` is False, always return immediately. If the lock was acquired, return True, otherwise return False. ``blocking_timeout`` specifies the maximum number of seconds to wait trying to acquire the lock. ``token`` specifies the token value to be used. If provided, token must be a bytes object or a string that can be encoded to a bytes object with the default encoding. If a token isn't specified, a UUID will be generated. """ sleep = self.sleep if token is None: token = uuid.uuid1().hex.encode() else: encoder = self.redis.connection_pool.get_encoder() token = encoder.encode(token) if blocking is None: blocking = self.blocking if blocking_timeout is None: blocking_timeout = self.blocking_timeout stop_trying_at = None if blocking_timeout is not None: stop_trying_at = mod_time.monotonic() + blocking_timeout while True: if self.do_acquire(token): self.local.token = token return True if not blocking: return False next_try_at = mod_time.monotonic() + sleep if stop_trying_at is not None and next_try_at > stop_trying_at: return False mod_time.sleep(sleep) def do_acquire(self, token): if self.timeout: # convert to milliseconds timeout = int(self.timeout * 1000) else: timeout = None if self.redis.set(self.name, token, nx=True, px=timeout): return True return False def locked(self): """ Returns True if this key is locked by any process, otherwise False. """ return self.redis.get(self.name) is not None def owned(self): """ Returns True if this key is locked by this lock, otherwise False. """ stored_token = self.redis.get(self.name) # need to always compare bytes to bytes # TODO: this can be simplified when the context manager is finished if stored_token and not isinstance(stored_token, bytes): encoder = self.redis.connection_pool.get_encoder() stored_token = encoder.encode(stored_token) return self.local.token is not None and stored_token == self.local.token def release(self): "Releases the already acquired lock" expected_token = self.local.token if expected_token is None: raise LockError("Cannot release an unlocked lock") self.local.token = None self.do_release(expected_token) def do_release(self, expected_token): if not bool( self.lua_release(keys=[self.name], args=[expected_token], client=self.redis) ): raise LockNotOwnedError("Cannot release a lock" " that's no longer owned") def extend(self, additional_time, replace_ttl=False): """ Adds more time to an already acquired lock. ``additional_time`` can be specified as an integer or a float, both representing the number of seconds to add. ``replace_ttl`` if False (the default), add `additional_time` to the lock's existing ttl. If True, replace the lock's ttl with `additional_time`. """ if self.local.token is None: raise LockError("Cannot extend an unlocked lock") if self.timeout is None: raise LockError("Cannot extend a lock with no timeout") return self.do_extend(additional_time, replace_ttl) def do_extend(self, additional_time, replace_ttl): additional_time = int(additional_time * 1000) if not bool( self.lua_extend( keys=[self.name], args=[self.local.token, additional_time, replace_ttl and "1" or "0"], client=self.redis, ) ): raise LockNotOwnedError("Cannot extend a lock that's" " no longer owned") return True def reacquire(self): """ Resets a TTL of an already acquired lock back to a timeout value. """ if self.local.token is None: raise LockError("Cannot reacquire an unlocked lock") if self.timeout is None: raise LockError("Cannot reacquire a lock with no timeout") return self.do_reacquire() def do_reacquire(self): timeout = int(self.timeout * 1000) if not bool( self.lua_reacquire( keys=[self.name], args=[self.local.token, timeout], client=self.redis ) ): raise LockNotOwnedError("Cannot reacquire a lock that's" " no longer owned") return True	10.76 KB	2025-05-25 12:26
📄 ocsp.py import base64 import datetime import ssl from urllib.parse import urljoin, urlparse import cryptography.hazmat.primitives.hashes import requests from cryptography import hazmat, x509 from cryptography.exceptions import InvalidSignature from cryptography.hazmat import backends from cryptography.hazmat.primitives.asymmetric.dsa import DSAPublicKey from cryptography.hazmat.primitives.asymmetric.ec import ECDSA, EllipticCurvePublicKey from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15 from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey from cryptography.hazmat.primitives.hashes import SHA1, Hash from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat from cryptography.x509 import ocsp from redis.exceptions import AuthorizationError, ConnectionError def _verify_response(issuer_cert, ocsp_response): pubkey = issuer_cert.public_key() try: if isinstance(pubkey, RSAPublicKey): pubkey.verify( ocsp_response.signature, ocsp_response.tbs_response_bytes, PKCS1v15(), ocsp_response.signature_hash_algorithm, ) elif isinstance(pubkey, DSAPublicKey): pubkey.verify( ocsp_response.signature, ocsp_response.tbs_response_bytes, ocsp_response.signature_hash_algorithm, ) elif isinstance(pubkey, EllipticCurvePublicKey): pubkey.verify( ocsp_response.signature, ocsp_response.tbs_response_bytes, ECDSA(ocsp_response.signature_hash_algorithm), ) else: pubkey.verify(ocsp_response.signature, ocsp_response.tbs_response_bytes) except InvalidSignature: raise ConnectionError("failed to valid ocsp response") def _check_certificate(issuer_cert, ocsp_bytes, validate=True): """A wrapper the return the validity of a known ocsp certificate""" ocsp_response = ocsp.load_der_ocsp_response(ocsp_bytes) if ocsp_response.response_status == ocsp.OCSPResponseStatus.UNAUTHORIZED: raise AuthorizationError("you are not authorized to view this ocsp certificate") if ocsp_response.response_status == ocsp.OCSPResponseStatus.SUCCESSFUL: if ocsp_response.certificate_status != ocsp.OCSPCertStatus.GOOD: raise ConnectionError( f'Received an {str(ocsp_response.certificate_status).split(".")[1]} ' "ocsp certificate status" ) else: raise ConnectionError( "failed to retrieve a sucessful response from the ocsp responder" ) if ocsp_response.this_update >= datetime.datetime.now(): raise ConnectionError("ocsp certificate was issued in the future") if ( ocsp_response.next_update and ocsp_response.next_update < datetime.datetime.now() ): raise ConnectionError("ocsp certificate has invalid update - in the past") responder_name = ocsp_response.responder_name issuer_hash = ocsp_response.issuer_key_hash responder_hash = ocsp_response.responder_key_hash cert_to_validate = issuer_cert if ( responder_name is not None and responder_name == issuer_cert.subject or responder_hash == issuer_hash ): cert_to_validate = issuer_cert else: certs = ocsp_response.certificates responder_certs = _get_certificates( certs, issuer_cert, responder_name, responder_hash ) try: responder_cert = responder_certs[0] except IndexError: raise ConnectionError("no certificates found for the responder") ext = responder_cert.extensions.get_extension_for_class(x509.ExtendedKeyUsage) if ext is None or x509.oid.ExtendedKeyUsageOID.OCSP_SIGNING not in ext.value: raise ConnectionError("delegate not autorized for ocsp signing") cert_to_validate = responder_cert if validate: _verify_response(cert_to_validate, ocsp_response) return True def _get_certificates(certs, issuer_cert, responder_name, responder_hash): if responder_name is None: certificates = [ c for c in certs if _get_pubkey_hash(c) == responder_hash and c.issuer == issuer_cert.subject ] else: certificates = [ c for c in certs if c.subject == responder_name and c.issuer == issuer_cert.subject ] return certificates def _get_pubkey_hash(certificate): pubkey = certificate.public_key() # https://stackoverflow.com/a/46309453/600498 if isinstance(pubkey, RSAPublicKey): h = pubkey.public_bytes(Encoding.DER, PublicFormat.PKCS1) elif isinstance(pubkey, EllipticCurvePublicKey): h = pubkey.public_bytes(Encoding.X962, PublicFormat.UncompressedPoint) else: h = pubkey.public_bytes(Encoding.DER, PublicFormat.SubjectPublicKeyInfo) sha1 = Hash(SHA1(), backend=backends.default_backend()) sha1.update(h) return sha1.finalize() def ocsp_staple_verifier(con, ocsp_bytes, expected=None): """An implemention of a function for set_ocsp_client_callback in PyOpenSSL. This function validates that the provide ocsp_bytes response is valid, and matches the expected, stapled responses. """ if ocsp_bytes in [b"", None]: raise ConnectionError("no ocsp response present") issuer_cert = None peer_cert = con.get_peer_certificate().to_cryptography() for c in con.get_peer_cert_chain(): cert = c.to_cryptography() if cert.subject == peer_cert.issuer: issuer_cert = cert break if issuer_cert is None: raise ConnectionError("no matching issuer cert found in certificate chain") if expected is not None: e = x509.load_pem_x509_certificate(expected) if peer_cert != e: raise ConnectionError("received and expected certificates do not match") return _check_certificate(issuer_cert, ocsp_bytes) class OCSPVerifier: """A class to verify ssl sockets for RFC6960/RFC6961. This can be used when using direct validation of OCSP responses and certificate revocations. @see https://datatracker.ietf.org/doc/html/rfc6960 @see https://datatracker.ietf.org/doc/html/rfc6961 """ def __init__(self, sock, host, port, ca_certs=None): self.SOCK = sock self.HOST = host self.PORT = port self.CA_CERTS = ca_certs def _bin2ascii(self, der): """Convert SSL certificates in a binary (DER) format to ASCII PEM.""" pem = ssl.DER_cert_to_PEM_cert(der) cert = x509.load_pem_x509_certificate(pem.encode(), backends.default_backend()) return cert def components_from_socket(self): """This function returns the certificate, primary issuer, and primary ocsp server in the chain for a socket already wrapped with ssl. """ # convert the binary certifcate to text der = self.SOCK.getpeercert(True) if der is False: raise ConnectionError("no certificate found for ssl peer") cert = self._bin2ascii(der) return self._certificate_components(cert) def _certificate_components(self, cert): """Given an SSL certificate, retract the useful components for validating the certificate status with an OCSP server. Args: cert ([bytes]): A PEM encoded ssl certificate """ try: aia = cert.extensions.get_extension_for_oid( x509.oid.ExtensionOID.AUTHORITY_INFORMATION_ACCESS ).value except cryptography.x509.extensions.ExtensionNotFound: raise ConnectionError("No AIA information present in ssl certificate") # fetch certificate issuers issuers = [ i for i in aia if i.access_method == x509.oid.AuthorityInformationAccessOID.CA_ISSUERS ] try: issuer = issuers[0].access_location.value except IndexError: issuer = None # now, the series of ocsp server entries ocsps = [ i for i in aia if i.access_method == x509.oid.AuthorityInformationAccessOID.OCSP ] try: ocsp = ocsps[0].access_location.value except IndexError: raise ConnectionError("no ocsp servers in certificate") return cert, issuer, ocsp def components_from_direct_connection(self): """Return the certificate, primary issuer, and primary ocsp server from the host defined by the socket. This is useful in cases where different certificates are occasionally presented. """ pem = ssl.get_server_certificate((self.HOST, self.PORT), ca_certs=self.CA_CERTS) cert = x509.load_pem_x509_certificate(pem.encode(), backends.default_backend()) return self._certificate_components(cert) def build_certificate_url(self, server, cert, issuer_cert): """Return the complete url to the ocsp""" orb = ocsp.OCSPRequestBuilder() # add_certificate returns an initialized OCSPRequestBuilder orb = orb.add_certificate( cert, issuer_cert, cryptography.hazmat.primitives.hashes.SHA256() ) request = orb.build() path = base64.b64encode( request.public_bytes(hazmat.primitives.serialization.Encoding.DER) ) url = urljoin(server, path.decode("ascii")) return url def check_certificate(self, server, cert, issuer_url): """Checks the validitity of an ocsp server for an issuer""" r = requests.get(issuer_url) if not r.ok: raise ConnectionError("failed to fetch issuer certificate") der = r.content issuer_cert = self._bin2ascii(der) ocsp_url = self.build_certificate_url(server, cert, issuer_cert) # HTTP 1.1 mandates the addition of the Host header in ocsp responses header = { "Host": urlparse(ocsp_url).netloc, "Content-Type": "application/ocsp-request", } r = requests.get(ocsp_url, headers=header) if not r.ok: raise ConnectionError("failed to fetch ocsp certificate") return _check_certificate(issuer_cert, r.content, True) def is_valid(self): """Returns the validity of the certificate wrapping our socket. This first retrieves for validate the certificate, issuer_url, and ocsp_server for certificate validate. Then retrieves the issuer certificate from the issuer_url, and finally checks the valididy of OCSP revocation status. """ # validate the certificate try: cert, issuer_url, ocsp_server = self.components_from_socket() if issuer_url is None: raise ConnectionError("no issuers found in certificate chain") return self.check_certificate(ocsp_server, cert, issuer_url) except AuthorizationError: cert, issuer_url, ocsp_server = self.components_from_direct_connection() if issuer_url is None: raise ConnectionError("no issuers found in certificate chain") return self.check_certificate(ocsp_server, cert, issuer_url)	11.18 KB	2025-05-25 12:26
📄 retry.py import socket from time import sleep from redis.exceptions import ConnectionError, TimeoutError class Retry: """Retry a specific number of times after a failure""" def __init__( self, backoff, retries, supported_errors=(ConnectionError, TimeoutError, socket.timeout), ): """ Initialize a `Retry` object with a `Backoff` object that retries a maximum of `retries` times. You can specify the types of supported errors which trigger a retry with the `supported_errors` parameter. """ self._backoff = backoff self._retries = retries self._supported_errors = supported_errors def update_supported_erros(self, specified_errors: list): """ Updates the supported errors with the specified error types """ self._supported_errors = tuple( set(self._supported_errors + tuple(specified_errors)) ) def call_with_retry(self, do, fail): """ Execute an operation that might fail and returns its result, or raise the exception that was thrown depending on the `Backoff` object. `do`: the operation to call. Expects no argument. `fail`: the failure handler, expects the last error that was thrown """ self._backoff.reset() failures = 0 while True: try: return do() except self._supported_errors as error: failures += 1 fail(error) if failures > self._retries: raise error backoff = self._backoff.compute(failures) if backoff > 0: sleep(backoff)	1.7 KB	2025-05-25 12:26
📄 sentinel.py import random import weakref from redis.client import Redis from redis.commands import SentinelCommands from redis.connection import Connection, ConnectionPool, SSLConnection from redis.exceptions import ConnectionError, ReadOnlyError, ResponseError, TimeoutError from redis.utils import str_if_bytes class MasterNotFoundError(ConnectionError): pass class SlaveNotFoundError(ConnectionError): pass class SentinelManagedConnection(Connection): def __init__(self, kwargs): self.connection_pool = kwargs.pop("connection_pool") super().__init__(kwargs) def __repr__(self): pool = self.connection_pool s = f"{type(self).__name__}<service={pool.service_name}%s>" if self.host: host_info = f",host={self.host},port={self.port}" s = s % host_info return s def connect_to(self, address): self.host, self.port = address super().connect() if self.connection_pool.check_connection: self.send_command("PING") if str_if_bytes(self.read_response()) != "PONG": raise ConnectionError("PING failed") def _connect_retry(self): if self._sock: return # already connected if self.connection_pool.is_master: self.connect_to(self.connection_pool.get_master_address()) else: for slave in self.connection_pool.rotate_slaves(): try: return self.connect_to(slave) except ConnectionError: continue raise SlaveNotFoundError # Never be here def connect(self): return self.retry.call_with_retry( self._connect_retry, lambda error: None, ) def read_response(self, disable_decoding=False): try: return super().read_response(disable_decoding=disable_decoding) except ReadOnlyError: if self.connection_pool.is_master: # When talking to a master, a ReadOnlyError when likely # indicates that the previous master that we're still connected # to has been demoted to a slave and there's a new master. # calling disconnect will force the connection to re-query # sentinel during the next connect() attempt. self.disconnect() raise ConnectionError("The previous master is now a slave") raise class SentinelManagedSSLConnection(SentinelManagedConnection, SSLConnection): pass class SentinelConnectionPool(ConnectionPool): """ Sentinel backed connection pool. If ``check_connection`` flag is set to True, SentinelManagedConnection sends a PING command right after establishing the connection. """ def __init__(self, service_name, sentinel_manager, kwargs): kwargs["connection_class"] = kwargs.get( "connection_class", SentinelManagedSSLConnection if kwargs.pop("ssl", False) else SentinelManagedConnection, ) self.is_master = kwargs.pop("is_master", True) self.check_connection = kwargs.pop("check_connection", False) super().__init__(kwargs) self.connection_kwargs["connection_pool"] = weakref.proxy(self) self.service_name = service_name self.sentinel_manager = sentinel_manager def __repr__(self): role = "master" if self.is_master else "slave" return f"{type(self).__name__}<service={self.service_name}({role})" def reset(self): super().reset() self.master_address = None self.slave_rr_counter = None def owns_connection(self, connection): check = not self.is_master or ( self.is_master and self.master_address == (connection.host, connection.port) ) parent = super() return check and parent.owns_connection(connection) def get_master_address(self): master_address = self.sentinel_manager.discover_master(self.service_name) if self.is_master: if self.master_address != master_address: self.master_address = master_address # disconnect any idle connections so that they reconnect # to the new master the next time that they are used. self.disconnect(inuse_connections=False) return master_address def rotate_slaves(self): "Round-robin slave balancer" slaves = self.sentinel_manager.discover_slaves(self.service_name) if slaves: if self.slave_rr_counter is None: self.slave_rr_counter = random.randint(0, len(slaves) - 1) for _ in range(len(slaves)): self.slave_rr_counter = (self.slave_rr_counter + 1) % len(slaves) slave = slaves[self.slave_rr_counter] yield slave # Fallback to the master connection try: yield self.get_master_address() except MasterNotFoundError: pass raise SlaveNotFoundError(f"No slave found for {self.service_name!r}") class Sentinel(SentinelCommands): """ Redis Sentinel cluster client >>> from redis.sentinel import Sentinel >>> sentinel = Sentinel([('localhost', 26379)], socket_timeout=0.1) >>> master = sentinel.master_for('mymaster', socket_timeout=0.1) >>> master.set('foo', 'bar') >>> slave = sentinel.slave_for('mymaster', socket_timeout=0.1) >>> slave.get('foo') b'bar' ``sentinels`` is a list of sentinel nodes. Each node is represented by a pair (hostname, port). ``min_other_sentinels`` defined a minimum number of peers for a sentinel. When querying a sentinel, if it doesn't meet this threshold, responses from that sentinel won't be considered valid. ``sentinel_kwargs`` is a dictionary of connection arguments used when connecting to sentinel instances. Any argument that can be passed to a normal Redis connection can be specified here. If ``sentinel_kwargs`` is not specified, any socket_timeout and socket_keepalive options specified in ``connection_kwargs`` will be used. ``connection_kwargs`` are keyword arguments that will be used when establishing a connection to a Redis server. """ def __init__( self, sentinels, min_other_sentinels=0, sentinel_kwargs=None, *connection_kwargs, ): # if sentinel_kwargs isn't defined, use the socket_ options from # connection_kwargs if sentinel_kwargs is None: sentinel_kwargs = { k: v for k, v in connection_kwargs.items() if k.startswith("socket_") } self.sentinel_kwargs = sentinel_kwargs self.sentinels = [ Redis(hostname, port, *self.sentinel_kwargs) for hostname, port in sentinels ] self.min_other_sentinels = min_other_sentinels self.connection_kwargs = connection_kwargs def execute_command(self, args, *kwargs): """ Execute Sentinel command in sentinel nodes. once - If set to True, then execute the resulting command on a single node at random, rather than across the entire sentinel cluster. """ once = bool(kwargs.get("once", False)) if "once" in kwargs.keys(): kwargs.pop("once") if once: for sentinel in self.sentinels: sentinel.execute_command(args, *kwargs) else: random.choice(self.sentinels).execute_command(args, kwargs) return True def __repr__(self): sentinel_addresses = [] for sentinel in self.sentinels: sentinel_addresses.append( "{host}:{port}".format_map( sentinel.connection_pool.connection_kwargs, ) ) return f'{type(self).__name__}<sentinels=[{",".join(sentinel_addresses)}]>' def check_master_state(self, state, service_name): if not state["is_master"] or state["is_sdown"] or state["is_odown"]: return False # Check if our sentinel doesn't see other nodes if state["num-other-sentinels"] < self.min_other_sentinels: return False return True def discover_master(self, service_name): """ Asks sentinel servers for the Redis master's address corresponding to the service labeled ``service_name``. Returns a pair (address, port) or raises MasterNotFoundError if no master is found. """ for sentinel_no, sentinel in enumerate(self.sentinels): try: masters = sentinel.sentinel_masters() except (ConnectionError, TimeoutError): continue state = masters.get(service_name) if state and self.check_master_state(state, service_name): # Put this sentinel at the top of the list self.sentinels[0], self.sentinels[sentinel_no] = ( sentinel, self.sentinels[0], ) return state["ip"], state["port"] raise MasterNotFoundError(f"No master found for {service_name!r}") def filter_slaves(self, slaves): "Remove slaves that are in an ODOWN or SDOWN state" slaves_alive = [] for slave in slaves: if slave["is_odown"] or slave["is_sdown"]: continue slaves_alive.append((slave["ip"], slave["port"])) return slaves_alive def discover_slaves(self, service_name): "Returns a list of alive slaves for service ``service_name``" for sentinel in self.sentinels: try: slaves = sentinel.sentinel_slaves(service_name) except (ConnectionError, ResponseError, TimeoutError): continue slaves = self.filter_slaves(slaves) if slaves: return slaves return [] def master_for( self, service_name, redis_class=Redis, connection_pool_class=SentinelConnectionPool, kwargs, ): """ Returns a redis client instance for the ``service_name`` master. A :py:class:`~redis.sentinel.SentinelConnectionPool` class is used to retrieve the master's address before establishing a new connection. NOTE: If the master's address has changed, any cached connections to the old master are closed. By default clients will be a :py:class:`~redis.Redis` instance. Specify a different class to the ``redis_class`` argument if you desire something different. The ``connection_pool_class`` specifies the connection pool to use. The :py:class:`~redis.sentinel.SentinelConnectionPool` will be used by default. All other keyword arguments are merged with any connection_kwargs passed to this class and passed to the connection pool as keyword arguments to be used to initialize Redis connections. """ kwargs["is_master"] = True connection_kwargs = dict(self.connection_kwargs) connection_kwargs.update(kwargs) return redis_class( connection_pool=connection_pool_class( service_name, self, connection_kwargs ) ) def slave_for( self, service_name, redis_class=Redis, connection_pool_class=SentinelConnectionPool, kwargs, ): """ Returns redis client instance for the ``service_name`` slave(s). A SentinelConnectionPool class is used to retrieve the slave's address before establishing a new connection. By default clients will be a :py:class:`~redis.Redis` instance. Specify a different class to the ``redis_class`` argument if you desire something different. The ``connection_pool_class`` specifies the connection pool to use. The SentinelConnectionPool will be used by default. All other keyword arguments are merged with any connection_kwargs passed to this class and passed to the connection pool as keyword arguments to be used to initialize Redis connections. """ kwargs["is_master"] = False connection_kwargs = dict(self.connection_kwargs) connection_kwargs.update(kwargs) return redis_class( connection_pool=connection_pool_class( service_name, self, **connection_kwargs ) )	12.4 KB	2025-05-25 12:26
📄 utils.py from contextlib import contextmanager try: import hiredis # noqa HIREDIS_AVAILABLE = True except ImportError: HIREDIS_AVAILABLE = False try: import cryptography # noqa CRYPTOGRAPHY_AVAILABLE = True except ImportError: CRYPTOGRAPHY_AVAILABLE = False def from_url(url, kwargs): """ Returns an active Redis client generated from the given database URL. Will attempt to extract the database id from the path url fragment, if none is provided. """ from redis.client import Redis return Redis.from_url(url, kwargs) @contextmanager def pipeline(redis_obj): p = redis_obj.pipeline() yield p p.execute() def str_if_bytes(value): return ( value.decode("utf-8", errors="replace") if isinstance(value, bytes) else value ) def safe_str(value): return str(str_if_bytes(value)) def dict_merge(dicts): """ Merge all provided dicts into 1 dict. dicts : `dict` dictionaries to merge """ merged = {} for d in dicts: merged.update(d) return merged def list_keys_to_dict(key_list, callback): return dict.fromkeys(key_list, callback) def merge_result(command, res): """ Merge all items in `res` into a list. This command is used when sending a command to multiple nodes and they result from each node should be merged into a single list. res : 'dict' """ result = set() for v in res.values(): for value in v: result.add(value) return list(result)	1.51 KB	2025-05-25 12:26

Terminal