Coverage for src / lexigram / admin / realtime / sse.py: 40%
156 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-08-13 22:14 +0800
1"""Server-Sent Events (SSE) integration for lexigram-admin.
3This module provides SSE handlers for real-time updates in admin.
5FWK-10: SSE for real-time updates using @sse_endpoint.
6"""
8from __future__ import annotations
10import asyncio
11from dataclasses import dataclass, field
12from datetime import UTC, datetime
13from enum import StrEnum
14from typing import TYPE_CHECKING, Any
16from lexigram.contracts.web.sse import ServerSentEvent, SseResponseFactoryProtocol
18if TYPE_CHECKING:
19 from collections.abc import AsyncGenerator
21# ============================================================================
22# Protocols for optional web integration
23# ============================================================================
25# Placeholder types for when lexigram-web is not available
26# These are replaced via container registration when lexigram-web is present
29class SSEHandler:
30 """Base SSE handler class.
32 This is a local implementation that does not depend on lexigram-web.
33 When lexigram-web is available, its SSEHandler is registered in the container
34 and resolved through IoC.
35 """
37 async def stream(self) -> AsyncGenerator[dict[str, Any], None]:
38 """Yield SSE events as dictionaries."""
39 if False: # pragma: no cover
40 yield {}
41 return
44# ============================================================================
45# Event Types
46# ============================================================================
49class AdminEventType(StrEnum):
50 """Standard admin event types."""
52 # Resource events
53 RESOURCE_CREATED = "resource.created"
54 RESOURCE_UPDATED = "resource.updated"
55 RESOURCE_DELETED = "resource.deleted"
57 # Bulk operation events
58 BULK_PROGRESS = "bulk.progress"
59 BULK_COMPLETED = "bulk.completed"
60 BULK_FAILED = "bulk.failed"
62 # Notification events
63 NOTIFICATION = "notification"
64 TOAST = "toast"
66 # System events
67 HEARTBEAT = "heartbeat"
68 RECONNECT = "reconnect"
71@dataclass
72class AdminEvent:
73 """Admin SSE event."""
75 event_type: AdminEventType | str
76 data: dict[str, Any]
77 id: str | None = None
78 resource_type: str | None = None
79 resource_id: Any = None
80 timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
82 def to_dict(self) -> dict[str, Any]:
83 """Convert to dictionary for SSE."""
84 return {
85 "event": str(
86 self.event_type.value
87 if isinstance(self.event_type, AdminEventType)
88 else self.event_type,
89 ),
90 "data": {
91 **self.data,
92 "timestamp": self.timestamp.isoformat(),
93 "resource_type": self.resource_type,
94 "resource_id": self.resource_id,
95 },
96 "id": self.id,
97 }
100# ============================================================================
101# Event Hub
102# ============================================================================
105class AdminEventHub:
106 """Central hub for admin SSE events.
108 Manages subscriptions and broadcasts events to connected clients.
110 Example:
111 >>> hub = AdminEventHub()
112 >>>
113 >>> # Subscribe to events
114 >>> async for event in hub.subscribe(user_id=1, resources=["users"]):
115 ... print(event)
116 >>>
117 >>> # Publish an event
118 >>> await hub.publish(AdminEvent(
119 ... event_type=AdminEventType.RESOURCE_UPDATED,
120 ... data={"changes": {...}},
121 ... resource_type="users",
122 ... resource_id=42,
123 ... ))
124 """
126 def __init__(self) -> None:
127 """Initialize the hub."""
128 self._subscribers: dict[str, asyncio.Queue[AdminEvent]] = {}
129 self._resource_subscriptions: dict[
130 str,
131 set[str],
132 ] = {} # resource -> subscriber_ids
133 self._user_subscriptions: dict[Any, set[str]] = {} # user_id -> subscriber_ids
135 def _generate_subscriber_id(self, user_id: Any | None = None) -> str:
136 """Generate unique subscriber ID."""
137 import uuid
139 base = str(uuid.uuid4())[:8]
140 if user_id:
141 return f"{user_id}:{base}"
142 return base
144 async def subscribe(
145 self,
146 user_id: Any | None = None,
147 resources: list[str] | None = None,
148 event_types: list[AdminEventType] | None = None,
149 ) -> AsyncGenerator[AdminEvent, None]:
150 """Subscribe to admin events.
152 Args:
153 user_id: Optional user ID for user-specific events
154 resources: Optional list of resource types to watch
155 event_types: Optional list of event types to receive
157 Yields:
158 AdminEvent objects as they arrive
159 """
160 subscriber_id = self._generate_subscriber_id(user_id)
161 queue: asyncio.Queue[AdminEvent] = asyncio.Queue()
163 # Register subscriber
164 self._subscribers[subscriber_id] = queue
166 if resources:
167 for resource in resources:
168 if resource not in self._resource_subscriptions:
169 self._resource_subscriptions[resource] = set()
170 self._resource_subscriptions[resource].add(subscriber_id)
172 if user_id:
173 if user_id not in self._user_subscriptions:
174 self._user_subscriptions[user_id] = set()
175 self._user_subscriptions[user_id].add(subscriber_id)
177 try:
178 while True:
179 event = await queue.get()
181 # Filter by event type if specified
182 if event_types and event.event_type not in event_types:
183 continue
185 yield event
186 finally:
187 # Cleanup on disconnect
188 self._subscribers.pop(subscriber_id, None)
190 if resources:
191 for resource in resources:
192 if resource in self._resource_subscriptions:
193 self._resource_subscriptions[resource].discard(subscriber_id)
195 if user_id and user_id in self._user_subscriptions:
196 self._user_subscriptions[user_id].discard(subscriber_id)
198 async def publish(
199 self,
200 event: AdminEvent,
201 target_users: list[Any] | None = None,
202 ) -> int:
203 """Publish an event to subscribers.
205 Args:
206 event: Event to publish
207 target_users: Optional specific users to target
209 Returns:
210 Number of subscribers that received the event
211 """
212 delivered = 0
213 target_subscriber_ids: set[str] = set()
215 # Determine target subscribers
216 if target_users:
217 for user_id in target_users:
218 if user_id in self._user_subscriptions:
219 target_subscriber_ids.update(self._user_subscriptions[user_id])
220 elif event.resource_type:
221 # Broadcast to resource subscribers
222 if event.resource_type in self._resource_subscriptions:
223 target_subscriber_ids.update(
224 self._resource_subscriptions[event.resource_type],
225 )
226 else:
227 # Broadcast to all
228 target_subscriber_ids.update(self._subscribers.keys())
230 # Deliver event
231 for subscriber_id in target_subscriber_ids:
232 if subscriber_id in self._subscribers:
233 try:
234 self._subscribers[subscriber_id].put_nowait(event)
235 delivered += 1
236 except asyncio.QueueFull:
237 pass # Skip if queue is full
239 return delivered
241 async def publish_resource_event(
242 self,
243 event_type: AdminEventType,
244 resource_type: str,
245 resource_id: Any,
246 data: dict[str, Any] | None = None,
247 ) -> int:
248 """Convenience method to publish a resource event."""
249 event = AdminEvent(
250 event_type=event_type,
251 data=data or {},
252 resource_type=resource_type,
253 resource_id=resource_id,
254 )
255 return await self.publish(event)
257 async def publish_notification(
258 self,
259 title: str,
260 message: str,
261 level: str = "info",
262 target_users: list[Any] | None = None,
263 ) -> int:
264 """Publish a notification event."""
265 event = AdminEvent(
266 event_type=AdminEventType.NOTIFICATION,
267 data={
268 "title": title,
269 "message": message,
270 "level": level,
271 },
272 )
273 return await self.publish(event, target_users=target_users)
275 async def publish_toast(
276 self,
277 message: str,
278 variant: str = "success",
279 duration: int = 5000,
280 target_users: list[Any] | None = None,
281 ) -> int:
282 """Publish a toast notification event."""
283 event = AdminEvent(
284 event_type=AdminEventType.TOAST,
285 data={
286 "message": message,
287 "variant": variant,
288 "duration": duration,
289 },
290 )
291 return await self.publish(event, target_users=target_users)
294HAS_SSE = True # local placeholder is always available
296# ============================================================================
297# Admin SSE Handler
298# ============================================================================
301class AdminEventsHandler(SSEHandler if HAS_SSE else object): # type: ignore[misc]
302 """SSE handler for admin events.
304 Streams events from AdminEventHub to connected clients.
306 Usage with lexigram.web:
307 >>> @sse_endpoint("/admin/events")
308 ... class AdminEventsEndpoint(AdminEventsHandler):
309 ... pass
310 """
312 heartbeat_interval: int = 30
313 retry: int = 3000
314 event_types: list[str] = [e.value for e in AdminEventType]
316 def __init__(self, hub: AdminEventHub) -> None:
317 self._hub = hub
319 async def stream(self, request: Any) -> AsyncGenerator[dict[str, Any], None]:
320 """Stream admin events to client."""
321 # Get user from request
322 user = getattr(request, "user", None)
323 user_id = getattr(user, "id", None) if user else None
325 # Get resource filter from query params
326 resources = None
327 if hasattr(request, "query_params"):
328 resources_param = request.query_params.get("resources")
329 if resources_param:
330 resources = resources_param.split(",")
332 # Subscribe and yield events
333 async for event in self._hub.subscribe(
334 user_id=user_id,
335 resources=resources,
336 ):
337 yield event.to_dict()
339 async def on_connect(self, request: Any) -> None:
340 """Handle client connection."""
341 # Could log connection or update presence
343 async def on_disconnect(self, request: Any) -> None:
344 """Handle client disconnection."""
345 # Could log disconnection or update presence
348# ============================================================================
349# Bulk Operation Progress Handler
350# ============================================================================
353class BulkOperationProgressHandler(SSEHandler if HAS_SSE else object): # type: ignore[misc]
354 """SSE handler for bulk operation progress.
356 Streams progress updates for long-running bulk operations.
358 Usage:
359 >>> @sse_endpoint("/admin/bulk/{operation_id}/progress")
360 ... class BulkProgressEndpoint(BulkOperationProgressHandler):
361 ... pass
362 """
364 heartbeat_interval: int = 5
365 retry: int = 1000
367 # In-memory progress tracking (should be Redis-backed in production)
368 _progress: dict[str, dict[str, Any]] = {}
370 @classmethod
371 def start_operation(
372 cls,
373 operation_id: str,
374 total: int,
375 description: str = "",
376 ) -> None:
377 """Start tracking a new operation."""
378 cls._progress[operation_id] = {
379 "total": total,
380 "processed": 0,
381 "failed": 0,
382 "description": description,
383 "status": "running",
384 "started_at": datetime.now(UTC).isoformat(),
385 }
387 @classmethod
388 def update_progress(
389 cls,
390 operation_id: str,
391 processed: int,
392 failed: int = 0,
393 ) -> None:
394 """Update operation progress."""
395 if operation_id in cls._progress:
396 cls._progress[operation_id]["processed"] = processed
397 cls._progress[operation_id]["failed"] = failed
399 @classmethod
400 def complete_operation(
401 cls,
402 operation_id: str,
403 success: bool = True,
404 message: str = "",
405 ) -> None:
406 """Mark operation as complete."""
407 if operation_id in cls._progress:
408 cls._progress[operation_id]["status"] = "completed" if success else "failed"
409 cls._progress[operation_id]["message"] = message
410 cls._progress[operation_id]["completed_at"] = datetime.now(
411 UTC,
412 ).isoformat()
414 async def stream(self, request: Any) -> AsyncGenerator[dict[str, Any], None]:
415 """Stream progress for an operation."""
416 operation_id = ""
417 if hasattr(request, "path_params"):
418 operation_id = request.path_params.get("operation_id", "")
420 if not operation_id or operation_id not in self._progress:
421 yield {
422 "event": "error",
423 "data": {"message": "Operation not found"},
424 }
425 return
427 while True:
428 progress = self._progress.get(operation_id)
429 if not progress:
430 break
432 yield {
433 "event": "progress",
434 "data": progress,
435 }
437 if progress["status"] in ("completed", "failed"):
438 # Final event
439 yield {
440 "event": progress["status"],
441 "data": progress,
442 }
443 # Cleanup
444 self._progress.pop(operation_id, None)
445 break
447 await asyncio.sleep(0.5)
450# ============================================================================
451# SSE Response Helpers
452# ============================================================================
455def create_sse_response(
456 event_generator: AsyncGenerator[dict[str, Any], None],
457 sse_factory: SseResponseFactoryProtocol,
458) -> Any:
459 """Create an SSE response from an event generator via an injected factory.
461 The ``sse_factory`` is registered in the DI container by ``lexigram-web``
462 during ``WebProvider.boot()``. Callers must resolve
463 ``SseResponseFactoryProtocol`` from the container and pass it here.
465 Args:
466 event_generator: Async generator yielding event-data dicts. Each dict
467 may contain ``data``, ``event``, ``id``, and ``retry`` keys.
468 sse_factory: Factory that wraps a ``ServerSentEvent`` generator into a
469 framework streaming response.
471 Returns:
472 A framework-specific SSE streaming HTTP response.
473 """
475 async def wrapped_generator() -> AsyncGenerator[ServerSentEvent, None]:
476 async for event_data in event_generator:
477 yield ServerSentEvent(
478 data=event_data.get("data", event_data),
479 event=event_data.get("event"),
480 event_id=event_data.get("id"),
481 retry=event_data.get("retry"),
482 )
484 return sse_factory.create_response(wrapped_generator())
487__all__ = [
488 # Flags
489 "HAS_SSE",
490 "AdminEvent",
491 # Hub
492 "AdminEventHub",
493 # Event types
494 "AdminEventType",
495 # Handlers
496 "AdminEventsHandler",
497 "BulkOperationProgressHandler",
498 # Helpers
499 "create_sse_response",
500]