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

1import copy 

2import random 

3import string 

4from typing import List, Tuple 

5 

6import redis 

7from redis.typing import KeysT, KeyT 

8 

9 

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

11 # returns a single new list combining keys and args 

12 try: 

13 iter(keys) 

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

15 # keys wasn't passed as a list 

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

17 keys = [keys] 

18 else: 

19 keys = list(keys) 

20 except TypeError: 

21 keys = [keys] 

22 if args: 

23 keys.extend(args) 

24 return keys 

25 

26 

27def nativestr(x): 

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

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

30 if r == "null": 

31 return 

32 return r 

33 

34 

35def delist(x): 

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

37 if x is None: 

38 return x 

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

40 

41 

42def parse_to_list(response): 

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

44 res = [] 

45 

46 special_values = {"infinity", "nan", "-infinity"} 

47 

48 if response is None: 

49 return res 

50 

51 for item in response: 

52 if item is None: 

53 res.append(None) 

54 continue 

55 try: 

56 item_str = nativestr(item) 

57 except TypeError: 

58 res.append(None) 

59 continue 

60 

61 if isinstance(item_str, str) and item_str.lower() in special_values: 

62 res.append(item_str) # Keep as string 

63 else: 

64 try: 

65 res.append(int(item)) 

66 except ValueError: 

67 try: 

68 res.append(float(item)) 

69 except ValueError: 

70 res.append(item_str) 

71 

72 return res 

73 

74 

75def random_string(length=10): 

76 """ 

77 Returns a random N character long string. 

78 """ 

79 return "".join( # nosec 

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

81 ) 

82 

83 

84def decode_dict_keys(obj): 

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

86 newobj = copy.copy(obj) 

87 for k in obj.keys(): 

88 if isinstance(k, bytes): 

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

90 newobj.pop(k) 

91 return newobj 

92 

93 

94def get_protocol_version(client): 

95 if isinstance(client, redis.Redis) or isinstance(client, redis.asyncio.Redis): 

96 return client.connection_pool.connection_kwargs.get("protocol") 

97 elif isinstance(client, redis.cluster.AbstractRedisCluster): 

98 return client.nodes_manager.connection_kwargs.get("protocol")