Coverage for src / lstautorta / utils / queue.py: 100%

1from collections.abc import Callable 

2from queue import Empty, Queue 

3from typing import Any 

4 

5 

6def get_all_non_blocking(q): 

7 """Get all items present in the queue, non-blocking. 

8 

9 Parameters 

10 ---------- 

11 q : Queue 

12 The queue to query 

13 

14 Returns 

15 ------- 

16 List[Any] 

17 Items retrieved from the queue. 

18 """ 

19 items = [] 

20 while True: 

21 try: 

22 items.append(q.get(block=False)) 

23 except Empty: 

24 break 

25 return items 

26 

27 

28def process_all_items( 

29 q: Queue[Any], 

30 processed_items: set[Any], 

31 process_item_fct: Callable[..., None], 

32 process_item_fct_extra_kwargs: dict[str, Any], 

33) -> int: 

34 """Empty `q` and apply `process_item_fct` on all items not in `processed_items` with kwargs from `process_item_fct_extra_kwargs` 

35 

36 Warnings 

37 -------- 

38 If the queue is filled faster than this function empties it, this will create an infinite list! 

39 

40 Parameters 

41 ---------- 

42 q : Queue[Any] 

43 queue which items will be processed 

44 processed_items : set[Any] 

45 Set of items that should not be processed. 

46 process_item_fct : Callable[..., None] 

47 Function to apply to the queue items 

48 process_item_fct_extra_kwargs : Dict[str, Any] 

49 Extra kwargs to pass to `process_item_fct` when proccessing items. 

50 

51 Returns 

52 ------- 

53 int 

54 The number of processed items. 

55 """ 

56 queue_items = get_all_non_blocking(q) 

57 for item in queue_items: 

58 if item not in processed_items: 

59 processed_items.add(item) 

60 process_item_fct(item, **process_item_fct_extra_kwargs) 

61 

62 return len(queue_items)