Share the current websocket context with the function decorated. The websocket context is local per task and hence will not be available in any other task. This decorator can be used to make the context available, .. code-block:: python @copy_current_websocket_context
(func: Callable)
| 406 | |
| 407 | |
| 408 | def copy_current_websocket_context(func: Callable) -> Callable: |
| 409 | """Share the current websocket context with the function decorated. |
| 410 | |
| 411 | The websocket context is local per task and hence will not be |
| 412 | available in any other task. This decorator can be used to make |
| 413 | the context available, |
| 414 | |
| 415 | .. code-block:: python |
| 416 | |
| 417 | @copy_current_websocket_context |
| 418 | async def within_context() -> None: |
| 419 | method = websocket.method |
| 420 | ... |
| 421 | |
| 422 | """ |
| 423 | if not has_websocket_context(): |
| 424 | raise RuntimeError( |
| 425 | "Attempt to copy websocket context outside of a websocket context" |
| 426 | ) |
| 427 | |
| 428 | websocket_context = _cv_websocket.get().copy() |
| 429 | |
| 430 | @wraps(func) |
| 431 | async def wrapper(*args: Any, **kwargs: Any) -> Any: |
| 432 | async with websocket_context: |
| 433 | return await websocket_context.app.ensure_async(func)(*args, **kwargs) |
| 434 | |
| 435 | return wrapper |
| 436 | |
| 437 | |
| 438 | def has_app_context() -> bool: |
searching dependent graphs…