Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/factory/utils.py: 66%

1# Copyright: See the LICENSE file. 

2 

3 

4import collections 

5import importlib 

6 

7 

8def import_object(module_name, attribute_name): 

9 """Import an object from its absolute path. 

10 

11 Example: 

12 >>> import_object('datetime', 'datetime') 

13 <type 'datetime.datetime'> 

14 """ 

15 module = importlib.import_module(module_name) 

16 return getattr(module, attribute_name) 

17 

18 

19class log_pprint: 

20 """Helper for properly printing args / kwargs passed to an object. 

21 

22 Since it is only used with factory.debug(), the computation is 

23 performed lazily. 

24 """ 

25 __slots__ = ['args', 'kwargs'] 

26 

27 def __init__(self, args=(), kwargs=None): 

28 self.args = args 

29 self.kwargs = kwargs or {} 

30 

31 def __repr__(self): 

32 return repr(str(self)) 

33 

34 def __str__(self): 

35 return ', '.join( 

36 [ 

37 repr(arg) for arg in self.args 

38 ] + [ 

39 '%s=%s' % (key, repr(value)) 

40 for key, value in self.kwargs.items() 

41 ] 

42 ) 

43 

44 

45class ResetableIterator: 

46 """An iterator wrapper that can be 'reset()' to its start.""" 

47 def __init__(self, iterator, **kwargs): 

48 super().__init__(**kwargs) 

49 self.iterator = iter(iterator) 

50 self.past_elements = collections.deque() 

51 self.next_elements = collections.deque() 

52 

53 def __iter__(self): 

54 while True: 

55 if self.next_elements: 

56 yield self.next_elements.popleft() 

57 else: 

58 try: 

59 value = next(self.iterator) 

60 except StopIteration: 

61 break 

62 else: 

63 self.past_elements.append(value) 

64 yield value 

65 

66 def reset(self): 

67 self.next_elements.clear() 

68 self.next_elements.extend(self.past_elements) 

69 

70 

71class OrderedBase: 

72 """Marks a class as being ordered. 

73 

74 Each instance (even from subclasses) will share a global creation counter. 

75 """ 

76 

77 CREATION_COUNTER_FIELD = '_creation_counter' 

78 

79 def __init__(self, **kwargs): 

80 super().__init__(**kwargs) 

81 if type(self) is not OrderedBase: 81 ↛ exitline 81 didn't return from function '__init__', because the condition on line 81 was never false

82 self.touch_creation_counter() 

83 

84 def touch_creation_counter(self): 

85 bases = type(self).__mro__ 

86 root = bases[bases.index(OrderedBase) - 1] 

87 if not hasattr(root, self.CREATION_COUNTER_FIELD): 

88 setattr(root, self.CREATION_COUNTER_FIELD, 0) 

89 next_counter = getattr(root, self.CREATION_COUNTER_FIELD) 

90 setattr(self, self.CREATION_COUNTER_FIELD, next_counter) 

91 setattr(root, self.CREATION_COUNTER_FIELD, next_counter + 1) 

92 

93 

94def sort_ordered_objects(items, getter=lambda x: x): 94 ↛ exitline 94 didn't run the lambda on line 94

95 """Sort an iterable of OrderedBase instances. 

96 

97 Args: 

98 items (iterable): the objects to sort 

99 getter (callable or None): a function to extract the OrderedBase instance from an object. 

100 

101 Examples: 

102 >>> sort_ordered_objects([x, y, z]) 

103 >>> sort_ordered_objects(v.items(), getter=lambda e: e[1]) 

104 """ 

105 return sorted(items, key=lambda x: getattr(getter(x), OrderedBase.CREATION_COUNTER_FIELD, -1))