Run WebSocket hooks and return rejection tuple or None if all pass. Args: hooks: List of hooks to run ws: The WebSocket connection *args: Additional arguments to pass to hooks default_reason: Default reason if hook returns False Retur
(
self, hooks, ws, *args, default_reason: str = "Rejected"
)
| 464 | ) from error |
| 465 | |
| 466 | async def _run_ws_hooks( |
| 467 | self, hooks, ws, *args, default_reason: str = "Rejected" |
| 468 | ) -> tuple | None: |
| 469 | """Run WebSocket hooks and return rejection tuple or None if all pass. |
| 470 | |
| 471 | Args: |
| 472 | hooks: List of hooks to run |
| 473 | ws: The WebSocket connection |
| 474 | *args: Additional arguments to pass to hooks |
| 475 | default_reason: Default reason if hook returns False |
| 476 | |
| 477 | Returns: |
| 478 | None if all hooks pass, or (code, reason) tuple for rejection |
| 479 | """ |
| 480 | for hook in hooks: |
| 481 | try: |
| 482 | result = hook(ws, *args) |
| 483 | if inspect.iscoroutine(result): |
| 484 | result = await result |
| 485 | if result is False: |
| 486 | return (4001, default_reason) |
| 487 | if isinstance(result, tuple) and len(result) == 2: |
| 488 | return result |
| 489 | except Exception: # pylint: disable=broad-exception-caught |
| 490 | return (4001, "Authentication error") |
| 491 | return None |
| 492 | |
| 493 | def _validate_ws_origin( |
| 494 | self, origin: str | None, host: str | None, allowed_origins: list |