Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/redis/utils.py: 65%

1import logging 

2import sys 

3from contextlib import contextmanager 

4from functools import wraps 

5from typing import Any, Dict, Mapping, Union 

6 

7try: 

8 import hiredis # noqa 

9 

10 # Only support Hiredis >= 1.0: 

11 HIREDIS_AVAILABLE = not hiredis.__version__.startswith("0.") 

12 HIREDIS_PACK_AVAILABLE = hasattr(hiredis, "pack_command") 

13except ImportError: 

14 HIREDIS_AVAILABLE = False 

15 HIREDIS_PACK_AVAILABLE = False 

16 

17try: 

18 import ssl # noqa 

19 

20 SSL_AVAILABLE = True 

21except ImportError: 

22 SSL_AVAILABLE = False 

23 

24try: 

25 import cryptography # noqa 

26 

27 CRYPTOGRAPHY_AVAILABLE = True 

28except ImportError: 

29 CRYPTOGRAPHY_AVAILABLE = False 

30 

31if sys.version_info >= (3, 8): 

32 from importlib import metadata 

33else: 

34 import importlib_metadata as metadata 

35 

36 

37def from_url(url, **kwargs): 

38 """ 

39 Returns an active Redis client generated from the given database URL. 

40 

41 Will attempt to extract the database id from the path url fragment, if 

42 none is provided. 

43 """ 

44 from redis.client import Redis 

45 

46 return Redis.from_url(url, **kwargs) 

47 

48 

49@contextmanager 

50def pipeline(redis_obj): 

51 p = redis_obj.pipeline() 

52 yield p 

53 p.execute() 

54 

55 

56def str_if_bytes(value: Union[str, bytes]) -> str: 

57 return ( 

58 value.decode("utf-8", errors="replace") if isinstance(value, bytes) else value 

59 ) 

60 

61 

62def safe_str(value): 

63 return str(str_if_bytes(value)) 

64 

65 

66def dict_merge(*dicts: Mapping[str, Any]) -> Dict[str, Any]: 

67 """ 

68 Merge all provided dicts into 1 dict. 

69 *dicts : `dict` 

70 dictionaries to merge 

71 """ 

72 merged = {} 

73 

74 for d in dicts: 

75 merged.update(d) 

76 

77 return merged 

78 

79 

80def list_keys_to_dict(key_list, callback): 

81 return dict.fromkeys(key_list, callback) 

82 

83 

84def merge_result(command, res): 

85 """ 

86 Merge all items in `res` into a list. 

87 

88 This command is used when sending a command to multiple nodes 

89 and the result from each node should be merged into a single list. 

90 

91 res : 'dict' 

92 """ 

93 result = set() 

94 

95 for v in res.values(): 

96 for value in v: 

97 result.add(value) 

98 

99 return list(result) 

100 

101 

102def warn_deprecated(name, reason="", version="", stacklevel=2): 

103 import warnings 

104 

105 msg = f"Call to deprecated {name}." 

106 if reason: 

107 msg += f" ({reason})" 

108 if version: 

109 msg += f" -- Deprecated since version {version}." 

110 warnings.warn(msg, category=DeprecationWarning, stacklevel=stacklevel) 

111 

112 

113def deprecated_function(reason="", version="", name=None): 

114 """ 

115 Decorator to mark a function as deprecated. 

116 """ 

117 

118 def decorator(func): 

119 @wraps(func) 

120 def wrapper(*args, **kwargs): 

121 warn_deprecated(name or func.__name__, reason, version, stacklevel=3) 

122 return func(*args, **kwargs) 

123 

124 return wrapper 

125 

126 return decorator 

127 

128 

129def _set_info_logger(): 

130 """ 

131 Set up a logger that log info logs to stdout. 

132 (This is used by the default push response handler) 

133 """ 

134 if "push_response" not in logging.root.manager.loggerDict.keys(): 

135 logger = logging.getLogger("push_response") 

136 logger.setLevel(logging.INFO) 

137 handler = logging.StreamHandler() 

138 handler.setLevel(logging.INFO) 

139 logger.addHandler(handler) 

140 

141 

142def get_lib_version(): 

143 try: 

144 libver = metadata.version("redis") 

145 except metadata.PackageNotFoundError: 

146 libver = "99.99.99" 

147 return libver