Coverage for src/lexigram/web/sse/decorators.py: 29%

1from __future__ import annotations 

2 

3from typing import Any, cast 

4 

5 

6def sse_endpoint( 

7 path: str, 

8 *, 

9 heartbeat_interval: int | None = None, 

10 retry: int | None = None, 

11) -> Any: 

12 """Decorator to mark a class as an SSE endpoint. 

13 

14 This decorator registers the handler class with the router 

15 and configures the SSE path. 

16 

17 Args: 

18 path: URL path for the SSE endpoint (can include path parameters) 

19 heartbeat_interval: Override default heartbeat interval 

20 retry: Override default client retry interval 

21 

22 Example: 

23 ```python 

24 @sse_endpoint("/events/{channel}") 

25 class ChannelEventsHandler(AbstractSSEHandler): 

26 async def stream(self, request): 

27 channel = request.path_params["channel"] 

28 async for event in get_events(channel): 

29 yield event 

30 ``` 

31 """ 

32 

33 def decorator(cls: type) -> type: 

34 # Store route metadata on the class using setattr to satisfy mypy 

35 cast("Any", cls)._sse_path = path 

36 cast("Any", cls)._sse_metadata = { 

37 "path": path, 

38 "heartbeat_interval": heartbeat_interval, 

39 "retry": retry, 

40 } 

41 

42 # Override class attributes if provided 

43 if heartbeat_interval is not None: 

44 cast("Any", cls).heartbeat_interval = heartbeat_interval 

45 if retry is not None: 

46 cast("Any", cls).retry = retry 

47 

48 # Mark as SSE handler for router discovery 

49 cast("Any", cls)._is_sse_handler = True 

50 

51 return cls 

52 

53 return decorator 

54 

55 

56__all__ = ["sse_endpoint"]