Coverage for src/lexigram/web/websocket/decorators.py: 25%

1"""WebSocket Decorators. 

2 

3Provides decorators for marking classes as WebSocket handlers. 

4""" 

5 

6from __future__ import annotations 

7 

8from typing import Any, cast 

9 

10 

11def websocket_handler( 

12 path: str, 

13 *, 

14 ping_interval: int | None = None, 

15 ping_timeout: int | None = None, 

16 max_connections_per_user: int | None = None, 

17) -> Any: 

18 """Decorator to mark a class as a WebSocket handler. 

19 

20 This decorator registers the handler class with the router 

21 and configures the WebSocket path. 

22 

23 Args: 

24 path: URL path for the WebSocket endpoint (can include path parameters) 

25 ping_interval: Override default ping interval 

26 ping_timeout: Override default ping timeout 

27 max_connections_per_user: Override max connections per user 

28 

29 Example: 

30 ```python 

31 @websocket_handler("/ws/chat/{room}") 

32 class ChatHandler(AbstractWebSocketHandler): 

33 async def on_connect(self, websocket): 

34 await websocket.accept() 

35 await self.broadcast({"type": "join", "user": "..."}) 

36 

37 async def on_message(self, websocket, message): 

38 await self.broadcast(message) 

39 

40 async def on_disconnect(self, websocket): 

41 await self.broadcast({"type": "leave", "user": "..."}) 

42 ``` 

43 """ 

44 

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

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

47 cast("Any", cls)._ws_path = path 

48 cast("Any", cls)._ws_metadata = { 

49 "path": path, 

50 "ping_interval": ping_interval, 

51 "ping_timeout": ping_timeout, 

52 "max_connections_per_user": max_connections_per_user, 

53 } 

54 

55 # Override class attributes if provided 

56 if ping_interval is not None: 

57 cast("Any", cls).ping_interval = ping_interval 

58 if ping_timeout is not None: 

59 cast("Any", cls).ping_timeout = ping_timeout 

60 if max_connections_per_user is not None: 

61 cast("Any", cls).max_connections_per_user = max_connections_per_user 

62 

63 # Mark as WebSocket handler for router discovery 

64 cast("Any", cls)._is_websocket_handler = True 

65 

66 return cls 

67 

68 return decorator 

69 

70 

71__all__ = ["websocket_handler"]