Coverage for lynceus/core/exchange/lynceus_exchange.py: 88%

1from asyncio import Queue 

2from uuid import UUID, uuid4 

3 

4 

5class LynceusExchange(Queue): 

6 """ 

7 An asynchronous queue-based exchange mechanism for Lynceus framework. 

8 

9 This class extends asyncio.Queue to provide a communication channel 

10 between different components of the Lynceus system. Each exchange 

11 instance is uniquely identified by a UUID for tracking and debugging 

12 purposes. 

13 

14 The exchange facilitates asynchronous message passing and can be used 

15 for inter-component communication, task coordination, and data flow 

16 management within the Lynceus ecosystem. 

17 

18 Inherits all queue functionality from asyncio.Queue including: 

19 - put() and get() operations 

20 - qsize() for queue monitoring 

21 - empty() and full() status checks 

22 - task tracking with task_done() and join() 

23 

24 Attributes: 

25 __uid (UUID): Unique identifier for this exchange instance 

26 """ 

27 

28 def __init__(self, *args, **kwargs): 

29 """ 

30 Initialize a new LynceusExchange instance. 

31 

32 Parameters 

33 ---------- 

34 *args 

35 Variable positional arguments passed to asyncio.Queue 

36 **kwargs 

37 Variable keyword arguments passed to asyncio.Queue 

38 Common kwargs include: 

39 - maxsize: Maximum number of items in the queue (0 = unlimited) 

40 

41 Notes 

42 ----- 

43 Automatically generates a unique UUID for this exchange instance 

44 that can be used for identification and logging purposes. 

45 """ 

46 super().__init__(*args, **kwargs) 

47 self.__uid: UUID = uuid4() 

48 

49 def __str__(self): 

50 """ 

51 Return a string representation of the exchange. 

52 

53 Returns 

54 ------- 

55 str 

56 Human-readable string containing the class name and unique ID 

57 """ 

58 return f'LynceusExchange-{self.__uid}'