Coverage for src/lexigram/admin/realtime/sse.py: 63%

1"""Server-Sent Events (SSE) integration for lexigram-admin. 

2 

3This module provides SSE handlers for real-time updates in admin. 

4 

5FWK-10: SSE for real-time updates using @sse_endpoint. 

6""" 

7 

8from __future__ import annotations 

9 

10import asyncio 

11from dataclasses import dataclass, field 

12from datetime import UTC, datetime 

13from enum import StrEnum 

14from typing import TYPE_CHECKING, Any 

15 

16if TYPE_CHECKING: 

17 from collections.abc import AsyncGenerator 

18 

19# ============================================================================ 

20# Protocols for optional web integration 

21# ============================================================================ 

22 

23# Placeholder types for when lexigram-web is not available 

24# These are replaced via container registration when lexigram-web is present 

25 

26 

27class SSEHandler: 

28 """Base SSE handler class. 

29 

30 This is a local implementation that does not depend on lexigram-web. 

31 When lexigram-web is available, its SSEHandler is registered in the container 

32 and resolved through IoC. 

33 """ 

34 

35 async def stream(self) -> AsyncGenerator[dict[str, Any], None]: 

36 """Yield SSE events as dictionaries.""" 

37 if False: # pragma: no cover 

38 yield {} 

39 return 

40 

41 

42# ============================================================================ 

43# Event Types 

44# ============================================================================ 

45 

46 

47class AdminEventType(StrEnum): 

48 """Standard admin event types.""" 

49 

50 # Resource events 

51 RESOURCE_CREATED = "resource.created" 

52 RESOURCE_UPDATED = "resource.updated" 

53 RESOURCE_DELETED = "resource.deleted" 

54 

55 # Bulk operation events 

56 BULK_PROGRESS = "bulk.progress" 

57 BULK_COMPLETED = "bulk.completed" 

58 BULK_FAILED = "bulk.failed" 

59 

60 # Notification events 

61 NOTIFICATION = "notification" 

62 TOAST = "toast" 

63 

64 # System events 

65 HEARTBEAT = "heartbeat" 

66 RECONNECT = "reconnect" 

67 

68 

69@dataclass 

70class AdminEvent: 

71 """Admin SSE event.""" 

72 

73 event_type: AdminEventType | str 

74 data: dict[str, Any] 

75 id: str | None = None 

76 resource_type: str | None = None 

77 resource_id: Any = None 

78 tenant_id: str | None = None 

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

80 

81 def to_dict(self) -> dict[str, Any]: 

82 """Convert to dictionary for SSE.""" 

83 return { 

84 "event": str( 

85 self.event_type.value 

86 if isinstance(self.event_type, AdminEventType) 

87 else self.event_type, 

88 ), 

89 "data": { 

90 **self.data, 

91 "timestamp": self.timestamp.isoformat(), 

92 "resource_type": self.resource_type, 

93 "resource_id": self.resource_id, 

94 }, 

95 "id": self.id, 

96 } 

97 

98 

99HAS_SSE = True # local placeholder is always available 

100 

101# ============================================================================ 

102# Bulk Operation Progress Handler 

103# ============================================================================ 

104 

105 

106class BulkOperationProgressHandler(SSEHandler if HAS_SSE else object): # type: ignore[misc] 

107 """SSE handler for bulk operation progress. 

108 

109 Streams progress updates for long-running bulk operations. 

110 

111 Usage: 

112 >>> @sse_endpoint("/admin/bulk/{operation_id}/progress") 

113 ... class BulkProgressEndpoint(BulkOperationProgressHandler): 

114 ... pass 

115 """ 

116 

117 heartbeat_interval: int = 5 

118 retry: int = 1000 

119 

120 # In-memory progress tracking (should be Redis-backed in production) 

121 _progress: dict[str, dict[str, Any]] = {} 

122 

123 @classmethod 

124 def start_operation( 

125 cls, 

126 operation_id: str, 

127 total: int, 

128 description: str = "", 

129 ) -> None: 

130 """Start tracking a new operation.""" 

131 cls._progress[operation_id] = { 

132 "total": total, 

133 "processed": 0, 

134 "failed": 0, 

135 "description": description, 

136 "status": "running", 

137 "started_at": datetime.now(UTC).isoformat(), 

138 } 

139 

140 @classmethod 

141 def update_progress( 

142 cls, 

143 operation_id: str, 

144 processed: int, 

145 failed: int = 0, 

146 ) -> None: 

147 """Update operation progress.""" 

148 if operation_id in cls._progress: 

149 cls._progress[operation_id]["processed"] = processed 

150 cls._progress[operation_id]["failed"] = failed 

151 

152 @classmethod 

153 def complete_operation( 

154 cls, 

155 operation_id: str, 

156 success: bool = True, 

157 message: str = "", 

158 ) -> None: 

159 """Mark operation as complete.""" 

160 if operation_id in cls._progress: 

161 cls._progress[operation_id]["status"] = "completed" if success else "failed" 

162 cls._progress[operation_id]["message"] = message 

163 cls._progress[operation_id]["completed_at"] = datetime.now( 

164 UTC, 

165 ).isoformat() 

166 

167 async def stream(self, request: Any) -> AsyncGenerator[dict[str, Any], None]: 

168 """Stream progress for an operation.""" 

169 operation_id = "" 

170 if hasattr(request, "path_params"): 

171 operation_id = request.path_params.get("operation_id", "") 

172 

173 if not operation_id or operation_id not in self._progress: 

174 yield { 

175 "event": "error", 

176 "data": {"message": "Operation not found"}, 

177 } 

178 return 

179 

180 while True: 

181 progress = self._progress.get(operation_id) 

182 if not progress: 

183 break 

184 

185 yield { 

186 "event": "progress", 

187 "data": progress, 

188 } 

189 

190 if progress["status"] in ("completed", "failed"): 

191 # Final event 

192 yield { 

193 "event": progress["status"], 

194 "data": progress, 

195 } 

196 # Cleanup 

197 self._progress.pop(operation_id, None) 

198 break 

199 

200 await asyncio.sleep(0.5) 

201 

202 

203__all__ = [ 

204 # Flags 

205 "HAS_SSE", 

206 "AdminEvent", 

207 # Event types 

208 "AdminEventType", 

209 # Handlers 

210 "BulkOperationProgressHandler", 

211]