Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/serde/utils.py: 44%

1""" 

2This module defines some utility functions. 

3""" 

4 

5import importlib 

6from collections import OrderedDict 

7from itertools import zip_longest 

8 

9from serde.exceptions import MissingDependencyError 

10 

11 

12def dict_partition(d, keyfunc, dict=OrderedDict): 

13 """ 

14 Partition a dictionary. 

15 

16 Args: 

17 d (dict): the dictionary to operate on. 

18 keyfunc (function): the function to partition with. It must accept the 

19 dictionary key and value as arguments, and should return a boolean. 

20 dict (type): the type of dictionaries to return. 

21 

22 Return: 

23 (dict, dict): all of the elements for which the key function returned 

24 True, and all of the elements for which it returned False. 

25 """ 

26 left = dict() 

27 right = dict() 

28 for key, value in d.items(): 

29 if keyfunc(key, value): 

30 left[key] = value 

31 else: 

32 right[key] = value 

33 return left, right 

34 

35 

36def is_subclass(cls, class_or_tuple): 

37 """ 

38 Return whether 'cls' is a derived from another class or is the same class. 

39 

40 Does not raise `TypeError` if the given `cls` is not a class. 

41 """ 

42 try: 

43 return issubclass(cls, class_or_tuple) 

44 except TypeError: 

45 return False 

46 

47 

48def subclasses(cls): 

49 """ 

50 Returns the recursed subclasses. 

51 

52 Args: 

53 cls (class): the class whose subclasses we should recurse. 

54 

55 Returns: 

56 list: the subclasses. 

57 """ 

58 subs = cls.__subclasses__() 

59 variants = [] 

60 for sub in subs: 

61 variants.extend(subclasses(sub)) 

62 return subs + variants 

63 

64 

65def try_lookup(name): 

66 """ 

67 Try lookup a fully qualified Python path, importing the module if necessary. 

68 

69 Args: 

70 name (str): the fully qualifed Python path. Example: 'validators.email'. 

71 

72 Returns: 

73 the object at the path. 

74 

75 Raises: 

76 serde.exceptions.MissingDependencyError: if the path could not be imported. 

77 """ 

78 module, path = name.split('.', 1) 

79 try: 

80 obj = importlib.import_module(module) 

81 except ImportError: 

82 raise MissingDependencyError( 

83 f"{module!r} is missing, did you forget to install the serde 'ext' feature?" 

84 ) 

85 for attr in path.split('.'): 

86 obj = getattr(obj, attr) 

87 return obj 

88 

89 

90def zip_equal(*iterables): 

91 """ 

92 A zip function that validates that all the iterables have the same length. 

93 

94 Args: 

95 *iterables: the iterables to pass to `zip_longest`. 

96 

97 Yields: 

98 each zipped element. 

99 

100 Raises: 

101 ValueError: if one of the iterables is the wrong length. 

102 """ 

103 sentinel = object() 

104 for element in zip_longest(*iterables, fillvalue=sentinel): 

105 if sentinel in element: 

106 raise ValueError('iterables have different lengths') 

107 yield element 

108 

109 

110def zip_until_right(*iterables): 

111 """ 

112 A zip function that validates that the right iterable is consumed. 

113 

114 Args: 

115 *iterables: the iterables to pass to `zip`. 

116 

117 Yields: 

118 each zipped element. 

119 

120 Raises: 

121 ValueError: if the left iterable is consumed before the right. 

122 """ 

123 lefts = iterables[:-1] 

124 right = iter(iterables[-1]) 

125 iterables = lefts + (right,) 

126 for item in zip(*iterables): 

127 yield item 

128 try: 

129 next(right) 

130 raise ValueError('the right-most iterable was not consumed') 

131 except StopIteration: 

132 pass