Coverage for nexios\websockets\errors.py: 40%

1from nexios.websockets import WebSocket 

2from nexios.exceptions import WebSocketException 

3from nexios.types import ASGIApp, Send, Receive, Scope 

4from nexios.logging import getLogger 

5import traceback 

6 

7logger = getLogger("nexios") 

8 

9 

10async def websocket_exception_handler( 

11 websocket: WebSocket, exc: WebSocketException 

12) -> None: 

13 error = traceback.format_exc() 

14 logger.error(f"WebSocket error: {error}") 

15 await websocket.close(code=exc.code, reason=str(exc)) 

16 

17 

18class WebSocketErrorMiddleware: 

19 def __init__(self, app: ASGIApp): 

20 self.app = app 

21 

22 async def __call__(self, scope: Scope, receive: Receive, send: Send): 

23 if scope["type"] == "websocket": 

24 websocket = WebSocket(scope, receive, send) 

25 try: 

26 await self.app(scope, receive, send) 

27 except WebSocketException as exc: 

28 await websocket_exception_handler(websocket, exc) 

29 except Exception as exc: 

30 error = traceback.format_exc() 

31 logger.error(f"Unexpected error: {error}") 

32 await websocket.close(code=1011, reason="Internal Server Error") 

33 else: 

34 await self.app(scope, receive, send)