Function run

dash/background_callback/managers/celery_manager.py:164–204

()

        ctx = copy_context()

        def run():
            c = AttributeDict(**context)  # type: ignore[reportCallIssue]
            c.ignore_register_page = False
            c.updated_props = ProxySetProps(_set_props)
            context_value.set(c)
            errored = False
            user_callback_output = None  # to help type checking
            try:
                if isinstance(user_callback_args, dict):
                    user_callback_output = fn(*maybe_progress, **user_callback_args)
                elif isinstance(user_callback_args, (list, tuple)):
                    user_callback_output = fn(*maybe_progress, *user_callback_args)
                else:
                    user_callback_output = fn(*maybe_progress, user_callback_args)
            except PreventUpdate:
                # Put NoUpdate dict directly to avoid circular imports.
                errored = True
                cache.set(
                    result_key,
                    json.dumps(
                        {"_dash_no_update": "_dash_no_update"}, cls=PlotlyJSONEncoder
                    ),
                )
            except Exception as err:  # pylint: disable=broad-except
                errored = True
                cache.set(
                    result_key,
                    json.dumps(
                        {
                            "background_callback_error": {
                                "msg": str(err),
                                "tb": traceback.format_exc(),
                            }
                        },
                    ),
                )

            if not errored:
                cache.set(
                    result_key, json.dumps(user_callback_output, cls=PlotlyJSONEncoder)
                )

        async def async_run():
            c = AttributeDict(**context)