Coverage for src / lexigram / contracts / web / sse.py: 100%

1"""Server-Sent Events (SSE) contracts. 

2 

3Defines the data transfer types and factory protocol for Server-Sent Events 

4so that packages outside of ``lexigram-web`` can produce SSE responses without 

5creating a cross-extension dependency. 

6 

7Usage:: 

8 

9 from lexigram.contracts.web.sse import ServerSentEvent, SseResponseFactoryProtocol 

10 

11 async def my_handler( 

12 factory: SseResponseFactoryProtocol, 

13 event_generator: AsyncGenerator[dict[str, Any], None], 

14 ) -> Any: 

15 async def wrapped(): 

16 async for event_data in event_generator: 

17 yield ServerSentEvent( 

18 data=event_data.get("data", event_data), 

19 event=event_data.get("event"), 

20 event_id=event_data.get("id"), 

21 retry=event_data.get("retry"), 

22 ) 

23 

24 return factory.create_response(wrapped()) 

25""" 

26 

27from __future__ import annotations 

28 

29from dataclasses import dataclass, field 

30from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

31 

32if TYPE_CHECKING: 

33 from collections.abc import AsyncGenerator 

34 

35 

36@dataclass(frozen=True) 

37class ServerSentEvent: 

38 """Data transfer object for a single Server-Sent Event. 

39 

40 Serialisation / encoding is the responsibility of the web-layer 

41 implementation (e.g. ``lexigram-web``). 

42 

43 Attributes: 

44 data: Event payload — any JSON-serialisable object or plain string. 

45 event: Optional event type name (sent as ``event:`` field). 

46 event_id: Optional event identifier (sent as ``id:`` field). 

47 retry: Optional reconnect interval in milliseconds. 

48 """ 

49 

50 data: Any 

51 event: str | None = field(default=None) 

52 event_id: str | None = field(default=None) 

53 retry: int | None = field(default=None) 

54 

55 

56@runtime_checkable 

57class SseResponseFactoryProtocol(Protocol): 

58 """Factory that wraps an async generator of ``ServerSentEvent`` objects 

59 into a framework HTTP response suitable for streaming SSE. 

60 

61 Implementations are provided by ``lexigram-web`` and registered in the 

62 DI container during ``WebProvider.boot()``. 

63 

64 Example:: 

65 

66 factory = await container.resolve(SseResponseFactoryProtocol) 

67 response = factory.create_response(my_event_generator()) 

68 """ 

69 

70 def create_response( 

71 self, 

72 generator: AsyncGenerator[ServerSentEvent, None], 

73 *, 

74 status_code: int = 200, 

75 headers: dict[str, str] | None = None, 

76 ) -> Any: 

77 """Wrap *generator* into a streaming SSE HTTP response. 

78 

79 Args: 

80 generator: Async generator yielding ``ServerSentEvent`` objects. 

81 status_code: HTTP status code for the response (default 200). 

82 headers: Optional additional response headers. 

83 

84 Returns: 

85 A framework-specific streaming response object. 

86 """ 

87 ... 

88 

89 

90__all__ = [ 

91 "ServerSentEvent", 

92 "SseResponseFactoryProtocol", 

93]