Coverage for src/lexigram/admin/data/subscriptions.py: 0%

1"""Real-time data subscriptions for Lexigram Admin.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from dataclasses import dataclass, field 

7from datetime import UTC, datetime 

8from typing import TYPE_CHECKING, Any, Literal 

9 

10if TYPE_CHECKING: 

11 from collections.abc import AsyncIterator 

12 

13 from lexigram.admin.data.query import QuerySpec as Query 

14 

15 

16@dataclass 

17class DataChange: 

18 """Represents a single data change event. 

19 

20 Attributes: 

21 type: The type of change ("created", "updated", "deleted"). 

22 resource: The name of the resource that changed. 

23 id: The unique identifier of the affected entity. 

24 data: The new data (for created/updated) or None. 

25 timestamp: When the change occurred. 

26 """ 

27 

28 type: Literal["created", "updated", "deleted"] 

29 resource: str 

30 id: Any 

31 data: dict[str, Any] | None 

32 timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) 

33 

34 

35class DataSubscription: 

36 """Subscription to real-time data changes. 

37 

38 This class manages the lifecycle of a data subscription, typically 

39 backed by Server-Sent Events (SSE) or a message broker. 

40 """ 

41 

42 def __init__( 

43 self, 

44 resource: str, 

45 query: Query | None = None, 

46 *, 

47 buffer_size: int = 100, 

48 ) -> None: 

49 """Initialize a subscription. 

50 

51 Args: 

52 resource: The resource name to watch. 

53 query: Optional query to filter which changes to receive. 

54 buffer_size: Maximum number of events to buffer in the queue. 

55 """ 

56 self.resource = resource 

57 self.query = query 

58 self._queue: asyncio.Queue[DataChange] = asyncio.Queue(maxsize=buffer_size) 

59 self._active = False 

60 

61 async def subscribe(self) -> AsyncIterator[DataChange]: 

62 """Start receiving data change events. 

63 

64 Yields: 

65 DataChange objects as they occur. 

66 """ 

67 self._active = True 

68 try: 

69 while self._active: 

70 change = await self._queue.get() 

71 yield change 

72 finally: 

73 self._active = False 

74 

75 async def unsubscribe(self) -> None: 

76 """Stop the subscription and clear the queue.""" 

77 self._active = False 

78 # Clear the queue by pulling all items 

79 while not self._queue.empty(): 

80 try: 

81 self._queue.get_nowait() 

82 except asyncio.QueueEmpty: 

83 break 

84 

85 async def emit(self, change: DataChange) -> None: 

86 """Manually emit a change into this subscription (infrastructure use).""" 

87 if not self._active: 

88 return 

89 

90 # If the subscription has a query, we should ideally check if the 

91 # changed data matches the query. This is complex for general queries. 

92 # For now, we just check resource name. 

93 if change.resource != self.resource: 

94 return 

95 

96 try: 

97 # We don't want to block if the queue is full, 

98 # we'd rather drop old events if necessary or let the user handle it. 

99 if self._queue.full(): 

100 self._queue.get_nowait() # Drop oldest 

101 self._queue.put_nowait(change) 

102 except (asyncio.QueueFull, asyncio.QueueEmpty): 

103 pass