Register callback API endpoints on the Flask app. Each key in callback_api_paths is a route, each value is a handler (sync or async). The view function parses the JSON body and passes it to the handler.
(
self, callback_api_paths: Dict[str, Callable[..., Any]]
)
| 316 | self.after_request(_after_request) |
| 317 | |
| 318 | def register_callback_api_routes( |
| 319 | self, callback_api_paths: Dict[str, Callable[..., Any]] |
| 320 | ): |
| 321 | """ |
| 322 | Register callback API endpoints on the Flask app. |
| 323 | Each key in callback_api_paths is a route, each value is a handler (sync or async). |
| 324 | The view function parses the JSON body and passes it to the handler. |
| 325 | """ |
| 326 | for path, handler in callback_api_paths.items(): |
| 327 | endpoint = f"dash_callback_api_{path}" |
| 328 | route = path if path.startswith("/") else f"/{path}" |
| 329 | methods = ["POST"] |
| 330 | |
| 331 | if inspect.iscoroutinefunction(handler): |
| 332 | |
| 333 | async def _async_view_func(*args, handler=handler, **kwargs): |
| 334 | data = request.get_json() |
| 335 | result = await handler(**data) if data else await handler() |
| 336 | return jsonify(result) |
| 337 | |
| 338 | view_func = _async_view_func |
| 339 | else: |
| 340 | |
| 341 | def _sync_view_func(*args, handler=handler, **kwargs): |
| 342 | data = request.get_json() |
| 343 | result = handler(**data) if data else handler() |
| 344 | return jsonify(result) |
| 345 | |
| 346 | view_func = _sync_view_func |
| 347 | |
| 348 | view_func = _sync_view_func |
| 349 | |
| 350 | # Flask 2.x+ supports async views natively |
| 351 | self.server.add_url_rule( |
| 352 | route, endpoint=endpoint, view_func=view_func, methods=methods |
| 353 | ) |
| 354 | |
| 355 | def enable_compression(self) -> None: |
| 356 | try: |
no test coverage detected