(*args: _P.args, **kwargs: _P.kwargs)
| 44 | class="cm"># 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(class="st">"websocket", args[idx] if idx < len(args) else None) |
| 48 | assert isinstance(websocket, WebSocket), ( |
| 49 | class="st">"Parameter with name &class="cm">#x27;websocket' is required to be of type 'WebSocket'" |
| 50 | fclass="st">" not &class="cm">#x27;{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