1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
import logging
from contextlib import contextmanager
from functools import wraps
from typing import Any, Dict, Mapping, Union
try:
import hiredis # noqa
# Only support Hiredis >= 1.0:
HIREDIS_AVAILABLE = not hiredis.__version__.startswith("0.")
HIREDIS_PACK_AVAILABLE = hasattr(hiredis, "pack_command")
except ImportError:
HIREDIS_AVAILABLE = False
HIREDIS_PACK_AVAILABLE = False
try:
import ssl # noqa
SSL_AVAILABLE = True
except ImportError:
SSL_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: Union[str, bytes]) -> str:
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: Mapping[str, Any]) -> Dict[str, Any]:
"""
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 the 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)
def warn_deprecated(name, reason="", version="", stacklevel=2):
import warnings
msg = f"Call to deprecated {name}."
if reason:
msg += f" ({reason})"
if version:
msg += f" -- Deprecated since version {version}."
warnings.warn(msg, category=DeprecationWarning, stacklevel=stacklevel)
def deprecated_function(reason="", version="", name=None):
"""
Decorator to mark a function as deprecated.
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
warn_deprecated(name or func.__name__, reason, version, stacklevel=3)
return func(*args, **kwargs)
return wrapper
return decorator
def _set_info_logger():
"""
Set up a logger that log info logs to stdout.
(This is used by the default push response handler)
"""
if "push_response" not in logging.root.manager.loggerDict.keys():
logger = logging.getLogger("push_response")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setLevel(logging.INFO)
logger.addHandler(handler)
|
