Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

RESP3 response callbacks #2798

Merged
merged 11 commits into from
Jun 15, 2023
Merged

RESP3 response callbacks #2798

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
fc9fbc5
start cleaning
dvora-h Apr 30, 2023
0faf16b
clean sone callbacks
dvora-h Jun 1, 2023
5210e26
Merge branch '5.0' into resp3-response-callbacks
dvora-h Jun 1, 2023
a24af93
Merge branch '5.0' into resp3-response-callbacks
dvora-h Jun 4, 2023
c08d0ac
response callbacks
dvora-h Jun 14, 2023
1c7d3c0
Merge branch '5.0' into resp3-response-callbacks
chayim Jun 15, 2023
32a6d3c
revert redismod-url change
dvora-h Jun 15, 2023
21c9e30
Merge branch 'resp3-response-callbacks' of https://github.com/dvora-h…
dvora-h Jun 15, 2023
85b47f7
fix async tests
dvora-h Jun 15, 2023
715110e
linters
dvora-h Jun 15, 2023
7c41c21
async cluster
dvora-h Jun 15, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion redis/asyncio/connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -379,7 +379,7 @@ async def on_connect(self) -> None:
raise AuthenticationError("Invalid Username or Password")

# if resp version is specified, switch to it
elif self.protocol != 2:
elif self.protocol not in [2, "2"]:
if isinstance(self._parser, _AsyncRESP2Parser):
self.set_parser(_AsyncRESP3Parser)
# update cluster exception classes
Expand Down
243 changes: 121 additions & 122 deletions redis/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -726,162 +726,159 @@ def parse_set_result(response, **options):

class AbstractRedis:
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("EXPIRE EXPIREAT PEXPIRE PEXPIREAT AUTH", bool),
**string_keys_to_dict("EXISTS", 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 ASKING 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,
**string_keys_to_dict("READONLY", bool_ok),
"CLUSTER DELSLOTS": bool_ok,
"CLUSTER ADDSLOTS": bool_ok,
"COMMAND": parse_command,
"INFO": parse_info,
"SET": parse_set_result,
"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 ADDSLOTSRANGE": bool_ok,
"LASTSAVE": timestamp_to_datetime,
"RESET": str_if_bytes,
"SLOWLOG GET": parse_slowlog_get,
"TIME": lambda x: (int(x[0]), int(x[1])),
**string_keys_to_dict("BLPOP BRPOP", lambda r: r and tuple(r) or None),
"SCAN": parse_scan,
"CLIENT GETNAME": str_if_bytes,
"SSCAN": parse_scan,
"ACL LOG": parse_acl_log,
"ACL WHOAMI": str_if_bytes,
"ACL GENPASS": str_if_bytes,
"ACL CAT": lambda r: list(map(str_if_bytes, r)),
"HSCAN": parse_hscan,
"ZSCAN": parse_zscan,
**string_keys_to_dict(
"BZPOPMIN BZPOPMAX", lambda r: r and (r[0], r[1], float(r[2])) or None
),
"CLUSTER COUNT-FAILURE-REPORTS": lambda x: int(x),
"CLUSTER COUNTKEYSINSLOT": lambda x: int(x),
"CLUSTER DELSLOTS": bool_ok,
"CLUSTER DELSLOTSRANGE": bool_ok,
"CLUSTER FAILOVER": bool_ok,
"CLUSTER FORGET": bool_ok,
"CLUSTER GETKEYSINSLOT": lambda r: list(map(str_if_bytes, r)),
"CLUSTER INFO": parse_cluster_info,
"CLUSTER KEYSLOT": lambda x: int(x),
"CLUSTER MEET": bool_ok,
"CLUSTER NODES": parse_cluster_nodes,
"CLUSTER REPLICAS": 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,
"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,
"FUNCTION DELETE": bool_ok,
"FUNCTION FLUSH": bool_ok,
"FUNCTION RESTORE": bool_ok,
**string_keys_to_dict("GEODIST", float_or_none),
"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,
"XAUTOCLAIM": parse_xautoclaim,
"XINFO STREAM": parse_xinfo_stream,
"XPENDING": parse_xpending,
**string_keys_to_dict("XREAD XREADGROUP", parse_xread),
"COMMAND GETKEYS": lambda r: list(map(str_if_bytes, r)),
**string_keys_to_dict("SORT", sort_return_tuples),
"PING": lambda r: str_if_bytes(r) == "PONG",
"QUIT": bool_ok,
"STRALGO": parse_stralgo,
"ACL SETUSER": bool_ok,
"PUBSUB NUMSUB": parse_pubsub_numsub,
"PUBSUB SHARDNUMSUB": 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])),
"ACL GETUSER": parse_acl_getuser,
"CONFIG SET": bool_ok,
**string_keys_to_dict("XREVRANGE XRANGE", parse_stream_list),
"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,

}

RESP2_RESPONSE_CALLBACKS = {
"CONFIG GET": parse_config_get,
**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("ZSCORE ZINCRBY", float_or_none),
"ZADD": parse_zadd,
"ZSCAN": parse_zscan,
"ZMSCORE": parse_zmscore,
"HGETALL": lambda r: r and pairs_to_dict(r) or {},
"MEMORY STATS": parse_memory_stats,
"MODULE LIST": lambda r: [pairs_to_dict(m) for m in r],

# **string_keys_to_dict(
# "COPY "
# "HEXISTS HMSET MOVE MSETNX PERSIST "
# "PSETEX RENAMENX SISMEMBER SMOVE SETEX SETNX",
# bool,
# ),
# **string_keys_to_dict(
# "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(
# "FLUSHALL FLUSHDB LSET LTRIM MSET PFMERGE ASKING READWRITE "
# "RENAME SAVE SELECT SHUTDOWN SLAVEOF SWAPDB WATCH UNWATCH ",
# bool_ok,
# ),
# **string_keys_to_dict("ZRANK ZREVRANK", int_or_none),
# **string_keys_to_dict("BGREWRITEAOF BGSAVE", lambda r: True),
# "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 SAVE": bool_ok,
# "ACL USERS": lambda r: list(map(str_if_bytes, r)),
# "CLIENT UNBLOCK": lambda r: r and int(r) == 1 or False,
# "CLIENT PAUSE": bool_ok,
# "CLUSTER ADDSLOTSRANGE": bool_ok,
# "CLUSTER DELSLOTSRANGE": bool_ok,
# "CLUSTER GETKEYSINSLOT": lambda r: list(map(str_if_bytes, r)),
# "CLUSTER REPLICAS": parse_cluster_nodes,
# "CLUSTER SET-CONFIG-EPOCH": bool_ok,
# "CONFIG RESETSTAT": bool_ok,
# "DEBUG OBJECT": parse_debug_object,
# "FUNCTION DELETE": bool_ok,
# "FUNCTION FLUSH": bool_ok,
# "FUNCTION RESTORE": bool_ok,
# "MEMORY PURGE": bool_ok,
# "MEMORY USAGE": int_or_none,
# "MODULE LOAD": parse_module_result,
# "MODULE UNLOAD": parse_module_result,
# "OBJECT": parse_object,
# "QUIT": bool_ok,
# "STRALGO": parse_stralgo,
# "RANDOMKEY": lambda r: r and r or None,
# "SCRIPT EXISTS": lambda r: list(map(bool, r)),
# "SCRIPT KILL": bool_ok,
# "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,
# "SLOWLOG RESET": bool_ok,
# "XGROUP CREATE": bool_ok,
# "XGROUP DESTROY": bool,
# "XGROUP SETID": bool_ok,
# "XINFO CONSUMERS": parse_list_of_dicts,
"XINFO GROUPS": parse_list_of_dicts,
}

RESP3_RESPONSE_CALLBACKS = {
Expand Down Expand Up @@ -1122,6 +1119,8 @@ def __init__(

if self.connection_pool.connection_kwargs.get("protocol") in ["3", 3]:
self.response_callbacks.update(self.__class__.RESP3_RESPONSE_CALLBACKS)
else:
self.response_callbacks.update(self.__class__.RESP2_RESPONSE_CALLBACKS)

def __repr__(self):
return f"{type(self).__name__}<{repr(self.connection_pool)}>"
Expand Down
4 changes: 2 additions & 2 deletions redis/connection.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -288,7 +288,7 @@ def on_connect(self):
auth_args = cred_provider.get_credentials()
# if resp version is specified and we have auth args,
# we need to send them via HELLO
if auth_args and self.protocol != 2:
if auth_args and self.protocol not in [2, "2"]:
if isinstance(self._parser, _RESP2Parser):
self.set_parser(_RESP3Parser)
# update cluster exception classes
Expand Down Expand Up @@ -321,7 +321,7 @@ def on_connect(self):
raise AuthenticationError("Invalid Username or Password")

# if resp version is specified, switch to it
elif self.protocol != 2:
elif self.protocol not in [2, "2"]:
if isinstance(self._parser, _RESP2Parser):
self.set_parser(_RESP3Parser)
# update cluster exception classes
Expand Down
2 changes: 1 addition & 1 deletion tests/conftest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@

REDIS_INFO = {}
default_redis_url = "redis://localhost:6379/0"
default_redismod_url = "redis://localhost:36379"
default_redismod_url = "redis://localhost:6379"
default_redis_unstable_url = "redis://localhost:6378"

# default ssl client ignores verification for the purpose of testing
Expand Down
Loading