(*args: _P.args, **kwargs: _P.kwargs)
| 44 | # Handle websocket functions. (Always async) |
| 45 | @functools.wraps(func) |
| 46 | async def websocket_wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None: |
| 47 | websocket = kwargs.get("websocket", args[idx] if idx < len(args) else None) |
| 48 | assert isinstance(websocket, WebSocket), ( |
| 49 | "Parameter with name 'websocket' is required to be of type 'WebSocket'" |
| 50 | f" not '{type(websocket).__name__}'" |
| 51 | ) |
| 52 | |
| 53 | if not has_required_scope(websocket, scopes_list): |
| 54 | await websocket.close() |
| 55 | else: |
| 56 | await func(*args, **kwargs) |
| 57 | |
| 58 | return websocket_wrapper |
| 59 |
nothing calls this directly
no test coverage detected