Run WebSocket hooks and return rejection tuple or None if all pass. Args: hooks: List of hooks to run websocket: The WebSocket connection *args: Additional arguments to pass to hooks default_reason: Default reason if hook returns False
(
self, hooks, websocket: "WebSocket", *args, default_reason: str = "Rejected"
)
| 646 | return response |
| 647 | |
| 648 | async def _run_ws_hooks( |
| 649 | self, hooks, websocket: "WebSocket", *args, default_reason: str = "Rejected" |
| 650 | ) -> tuple | None: |
| 651 | """Run WebSocket hooks and return rejection tuple or None if all pass. |
| 652 | |
| 653 | Args: |
| 654 | hooks: List of hooks to run |
| 655 | websocket: The WebSocket connection |
| 656 | *args: Additional arguments to pass to hooks |
| 657 | default_reason: Default reason if hook returns False |
| 658 | |
| 659 | Returns: |
| 660 | None if all hooks pass, or (code, reason) tuple for rejection |
| 661 | """ |
| 662 | for hook in hooks: |
| 663 | try: |
| 664 | result = hook(websocket, *args) |
| 665 | if inspect.iscoroutine(result): |
| 666 | result = await result |
| 667 | if result is False: |
| 668 | return (4001, default_reason) |
| 669 | if isinstance(result, tuple) and len(result) == 2: |
| 670 | return result |
| 671 | except Exception: # pylint: disable=broad-exception-caught |
| 672 | return (4001, "Authentication error") |
| 673 | return None |
| 674 | |
| 675 | def serve_websocket_callback(self, dash_app: "Dash"): |
| 676 | """Set up the WebSocket endpoint for callback handling. |