Create callback context from WebSocket message. Args: payload: The callback payload response_adapter: The response adapter instance for the backend websocket_callback: The websocket callback instance for the backend Returns: AttributeDict with callback conte
(
payload: CallbackExecutionBody,
response_adapter: "ResponseAdapter",
websocket_callback: DashWebsocketCallback,
)
| 187 | |
| 188 | |
| 189 | def create_ws_context( |
| 190 | payload: CallbackExecutionBody, |
| 191 | response_adapter: "ResponseAdapter", |
| 192 | websocket_callback: DashWebsocketCallback, |
| 193 | ): |
| 194 | """Create callback context from WebSocket message. |
| 195 | |
| 196 | Args: |
| 197 | payload: The callback payload |
| 198 | response_adapter: The response adapter instance for the backend |
| 199 | websocket_callback: The websocket callback instance for the backend |
| 200 | |
| 201 | Returns: |
| 202 | AttributeDict with callback context |
| 203 | """ |
| 204 | # pylint: disable=import-outside-toplevel |
| 205 | from dash._utils import AttributeDict, inputs_to_dict |
| 206 | |
| 207 | g = AttributeDict({}) |
| 208 | g.inputs_list = payload.get("inputs", []) |
| 209 | g.states_list = payload.get("state", []) |
| 210 | g.outputs_list = payload.get("outputs", []) |
| 211 | g.input_values = inputs_to_dict(g.inputs_list) |
| 212 | g.state_values = inputs_to_dict(g.states_list) |
| 213 | g.triggered_inputs = [ |
| 214 | {"prop_id": x, "value": g.input_values.get(x)} |
| 215 | for x in payload.get("changedPropIds", []) |
| 216 | ] |
| 217 | g.dash_response = response_adapter |
| 218 | g.updated_props = {} |
| 219 | g.dash_websocket = websocket_callback |
| 220 | |
| 221 | return g |
| 222 | |
| 223 | |
| 224 | async def run_ws_sender( |
no test coverage detected
searching dependent graphs…