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

1from contextlib import contextmanager 

2from functools import wraps 

3from typing import Any, Dict, Mapping, Union 

4 

5try: 

6 import hiredis # noqa 

7 

8 # Only support Hiredis >= 1.0: 

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

10 HIREDIS_PACK_AVAILABLE = hasattr(hiredis, "pack_command") 

11except ImportError: 

12 HIREDIS_AVAILABLE = False 

13 HIREDIS_PACK_AVAILABLE = False 

14 

15try: 

16 import cryptography # noqa 

17 

18 CRYPTOGRAPHY_AVAILABLE = True 

19except ImportError: 

20 CRYPTOGRAPHY_AVAILABLE = False 

21 

22 

23def from_url(url, **kwargs): 

24 """ 

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

26 

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

28 none is provided. 

29 """ 

30 from redis.client import Redis 

31 

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

33 

34 

35@contextmanager 

36def pipeline(redis_obj): 

37 p = redis_obj.pipeline() 

38 yield p 

39 p.execute() 

40 

41 

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

43 return ( 

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

45 ) 

46 

47 

48def safe_str(value): 

49 return str(str_if_bytes(value)) 

50 

51 

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

53 """ 

54 Merge all provided dicts into 1 dict. 

55 *dicts : `dict` 

56 dictionaries to merge 

57 """ 

58 merged = {} 

59 

60 for d in dicts: 

61 merged.update(d) 

62 

63 return merged 

64 

65 

66def list_keys_to_dict(key_list, callback): 

67 return dict.fromkeys(key_list, callback) 

68 

69 

70def merge_result(command, res): 

71 """ 

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

73 

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

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

76 

77 res : 'dict' 

78 """ 

79 result = set() 

80 

81 for v in res.values(): 

82 for value in v: 

83 result.add(value) 

84 

85 return list(result) 

86 

87 

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

89 import warnings 

90 

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

92 if reason: 

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

94 if version: 

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

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

97 

98 

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

100 """ 

101 Decorator to mark a function as deprecated. 

102 """ 

103 

104 def decorator(func): 

105 @wraps(func) 

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

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

108 return func(*args, **kwargs) 

109 

110 return wrapper 

111 

112 return decorator