Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/redis/commands/helpers.py: 85%

1import copy 

2import random 

3import string 

4from typing import List, Tuple 

5 

6from redis.typing import KeysT, KeyT 

7 

8 

9def list_or_args(keys: KeysT, args: Tuple[KeyT, ...]) -> List[KeyT]: 

10 # returns a single new list combining keys and args 

11 try: 

12 iter(keys) 

13 # a string or bytes instance can be iterated, but indicates 

14 # keys wasn't passed as a list 

15 if isinstance(keys, (bytes, str)): 

16 keys = [keys] 

17 else: 

18 keys = list(keys) 

19 except TypeError: 

20 keys = [keys] 

21 if args: 

22 keys.extend(args) 

23 return keys 

24 

25 

26def nativestr(x): 

27 """Return the decoded binary string, or a string, depending on type.""" 

28 r = x.decode("utf-8", "replace") if isinstance(x, bytes) else x 

29 if r == "null": 

30 return 

31 return r 

32 

33 

34def delist(x): 

35 """Given a list of binaries, return the stringified version.""" 

36 if x is None: 

37 return x 

38 return [nativestr(obj) for obj in x] 

39 

40 

41def parse_to_list(response): 

42 """Optimistically parse the response to a list.""" 

43 res = [] 

44 

45 if response is None: 

46 return res 

47 

48 for item in response: 

49 try: 

50 res.append(int(item)) 

51 except ValueError: 

52 try: 

53 res.append(float(item)) 

54 except ValueError: 

55 res.append(nativestr(item)) 

56 except TypeError: 

57 res.append(None) 

58 return res 

59 

60 

61def parse_list_to_dict(response): 

62 res = {} 

63 for i in range(0, len(response), 2): 

64 if isinstance(response[i], list): 

65 res["Child iterators"].append(parse_list_to_dict(response[i])) 

66 elif isinstance(response[i + 1], list): 

67 res["Child iterators"] = [parse_list_to_dict(response[i + 1])] 

68 else: 

69 try: 

70 res[response[i]] = float(response[i + 1]) 

71 except (TypeError, ValueError): 

72 res[response[i]] = response[i + 1] 

73 return res 

74 

75 

76def parse_to_dict(response): 

77 if response is None: 

78 return {} 

79 

80 res = {} 

81 for det in response: 

82 if isinstance(det[1], list): 

83 res[det[0]] = parse_list_to_dict(det[1]) 

84 else: 

85 try: # try to set the attribute. may be provided without value 

86 try: # try to convert the value to float 

87 res[det[0]] = float(det[1]) 

88 except (TypeError, ValueError): 

89 res[det[0]] = det[1] 

90 except IndexError: 

91 pass 

92 return res 

93 

94 

95def random_string(length=10): 

96 """ 

97 Returns a random N character long string. 

98 """ 

99 return "".join( # nosec 

100 random.choice(string.ascii_lowercase) for x in range(length) 

101 ) 

102 

103 

104def quote_string(v): 

105 """ 

106 RedisGraph strings must be quoted, 

107 quote_string wraps given v with quotes incase 

108 v is a string. 

109 """ 

110 

111 if isinstance(v, bytes): 

112 v = v.decode() 

113 elif not isinstance(v, str): 

114 return v 

115 if len(v) == 0: 

116 return '""' 

117 

118 v = v.replace("\\", "\\\\") 

119 v = v.replace('"', '\\"') 

120 

121 return f'"{v}"' 

122 

123 

124def decode_dict_keys(obj): 

125 """Decode the keys of the given dictionary with utf-8.""" 

126 newobj = copy.copy(obj) 

127 for k in obj.keys(): 

128 if isinstance(k, bytes): 

129 newobj[k.decode("utf-8")] = newobj[k] 

130 newobj.pop(k) 

131 return newobj 

132 

133 

134def stringify_param_value(value): 

135 """ 

136 Turn a parameter value into a string suitable for the params header of 

137 a Cypher command. 

138 You may pass any value that would be accepted by `json.dumps()`. 

139 

140 Ways in which output differs from that of `str()`: 

141 * Strings are quoted. 

142 * None --> "null". 

143 * In dictionaries, keys are _not_ quoted. 

144 

145 :param value: The parameter value to be turned into a string. 

146 :return: string 

147 """ 

148 

149 if isinstance(value, str): 

150 return quote_string(value) 

151 elif value is None: 

152 return "null" 

153 elif isinstance(value, (list, tuple)): 

154 return f'[{",".join(map(stringify_param_value, value))}]' 

155 elif isinstance(value, dict): 

156 return f'{{{",".join(f"{k}:{stringify_param_value(v)}" for k, v in value.items())}}}' # noqa 

157 else: 

158 return str(value)