Coverage for src / lexigram / ui / htmx / sse.py: 83%

1"""SSE support for Lexigram UI. 

2 

3Server-side: ``SSEMessage``, ``SSEStream`` — format and stream SSE payloads. 

4Client-side: ``SSE`` component — renders the HTMX SSE connector element with 

5 typed event names and automatic exponential-backoff reconnection. 

6""" 

7 

8from __future__ import annotations 

9 

10import asyncio 

11from collections.abc import AsyncGenerator 

12from enum import Enum 

13from typing import Any 

14 

15from starlette.responses import StreamingResponse 

16 

17from lexigram.serialization import dumps_str 

18from lexigram.ui.core.base import Component, el, raw 

19 

20__all__ = [ 

21 "SSEEventType", 

22 "SSEMessage", 

23 "SSEStream", 

24 "SSE", 

25] 

26 

27 

28# --------------------------------------------------------------------------- 

29# Typed event names 

30# --------------------------------------------------------------------------- 

31 

32 

33class SSEEventType(str, Enum): 

34 """Well-known SSE event type names. 

35 

36 Use these instead of bare strings when configuring the ``SSE`` component 

37 or emitting ``SSEMessage`` objects so the names stay in sync across the 

38 client and server. 

39 """ 

40 

41 MESSAGE = "message" 

42 UPDATE = "update" 

43 ERROR = "error" 

44 PING = "ping" 

45 CLOSE = "close" 

46 CONNECT = "connect" 

47 DISCONNECT = "disconnect" 

48 

49 

50# --------------------------------------------------------------------------- 

51# Server-side helpers 

52# --------------------------------------------------------------------------- 

53 

54 

55class SSEMessage: 

56 """Helper to format SSE messages.""" 

57 

58 def __init__( 

59 self, 

60 data: Any, 

61 event: str | None = None, 

62 event_id: str | None = None, 

63 retry: int | None = None, 

64 ): 

65 self.data = data 

66 self.event = event 

67 self.id = event_id 

68 self.retry = retry 

69 

70 def __str__(self) -> str: 

71 lines = [] 

72 if self.id: 

73 lines.append(f"id: {self.id}") 

74 if self.event: 

75 lines.append(f"event: {self.event}") 

76 if self.retry: 

77 lines.append(f"retry: {self.retry}") 

78 

79 data = self.data 

80 if not isinstance(data, str): 

81 data = dumps_str(data) 

82 

83 for line in data.split("\n"): 

84 lines.append(f"data: {line}") 

85 

86 return "\n".join(lines) + "\n\n" 

87 

88 

89class SSEStream(StreamingResponse): 

90 """Streaming response for Server-Sent Events.""" 

91 

92 def __init__(self, generator: AsyncGenerator[SSEMessage, None], **kwargs: Any): 

93 async def event_generator() -> Any: 

94 try: 

95 async for message in generator: 

96 yield str(message) 

97 except asyncio.CancelledError: 

98 pass 

99 

100 super().__init__(event_generator(), media_type="text/event-stream", **kwargs) 

101 self.headers["Cache-Control"] = "no-cache" 

102 self.headers["Connection"] = "keep-alive" 

103 self.headers["X-Accel-Buffering"] = "no" 

104 

105 

106# --------------------------------------------------------------------------- 

107# Client-side HTMX SSE connector component 

108# --------------------------------------------------------------------------- 

109 

110 

111class SSE(Component): 

112 """Client-side HTMX SSE connector with typed event support and auto-reconnect. 

113 

114 Renders a ``<div>`` wired up with the ``hx-ext="sse"`` extension plus an 

115 inline ``<script>`` that listens for ``htmx:sseError`` and schedules a 

116 reconnect with exponential back-off, capped at 30 s. 

117 

118 Args: 

119 url: SSE endpoint URL (``sse-connect`` attribute). 

120 target: CSS selector or ``"this"`` for the HTMX swap target. 

121 event_type: SSE event name to swap on (``sse-swap`` attribute). 

122 Accepts a plain string or an ``SSEEventType`` member. 

123 retry_ms: Initial reconnect delay in milliseconds (default 3000). 

124 Doubles on each failed attempt, capped at 30 000 ms. 

125 """ 

126 

127 def __init__( 

128 self, 

129 url: str, 

130 target: str = "this", 

131 event_type: str = SSEEventType.MESSAGE, 

132 retry_ms: int = 3000, 

133 **props: Any, 

134 ) -> None: 

135 super().__init__( 

136 url=url, target=target, event_type=event_type, retry_ms=retry_ms, **props 

137 ) 

138 self.url = url 

139 self.target = target 

140 # Normalise SSEEventType members to their string value 

141 self.event_type = ( 

142 event_type.value if isinstance(event_type, SSEEventType) else event_type 

143 ) 

144 self.retry_ms = max(100, int(retry_ms)) 

145 

146 def render(self) -> Any: 

147 # Use the Python object id to produce a unique DOM id per render. 

148 component_id = f"sse-{id(self)}" 

149 

150 reconnect_script = raw( 

151 f"<script>" 

152 f"(function(){{" 

153 f"var _el=document.getElementById('{component_id}');" 

154 f"var _retry={self.retry_ms};" 

155 f"var _max=30000;" 

156 f"var _n=0;" 

157 f"document.addEventListener('htmx:sseError',function(e){{" 

158 f"if(!_el||!e.detail)return;" 

159 f"var delay=Math.min(_retry*Math.pow(2,_n),_max);" 

160 f"_n++;" 

161 f"setTimeout(function(){{if(_el&&typeof htmx!=='undefined')htmx.process(_el);}},delay);" 

162 f"}});" 

163 f"document.addEventListener('htmx:sseOpen',function(){{{{" 

164 f"_n=0;" 

165 f"}}}});" 

166 f"}})();" 

167 f"</script>" 

168 ) 

169 

170 return el( 

171 "div", 

172 { 

173 "id": component_id, 

174 "hx-ext": "sse", 

175 "sse-connect": self.url, 

176 "sse-swap": self.event_type, 

177 "hx-target": self.target, 

178 }, 

179 *self.children, 

180 reconnect_script, 

181 )