on error without returning outputs, fix celery on_error.

Author: T4rk1n

dash/_callback.py

@@ -233,7 +233,7 @@ def insert_callback(
     long=None,
     manager=None,
     running=None,
-    dynamic_creator=False,
+    dynamic_creator: Optional[bool] = False,
     no_output=False,
 ):
     if prevent_initial_call is None:
@@ -279,6 +279,14 @@ def insert_callback(
     return callback_id
 
 
+def _set_side_update(ctx, response) -> bool:
+    side_update = dict(ctx.updated_props)
+    if len(side_update) > 0:
+        response["sideUpdate"] = side_update
+        return True
+    return False
+
+
 # pylint: disable=too-many-branches,too-many-statements
 def register_callback(
     callback_list, callback_map, config_prevent_initial_callbacks, *_args, **_kwargs
@@ -350,7 +358,7 @@ def add_context(*args, **kwargs):
                 "callback_context", AttributeDict({"updated_props": {}})
             )
             callback_manager = long and long.get("manager", app_callback_manager)
-            app_on_error = kwargs.pop("app_on_error", None)
+            error_handler = on_error or kwargs.pop("app_on_error", None)
 
             if has_output:
                 _validate.validate_output_spec(insert_output, output_spec, Output)
@@ -361,7 +369,7 @@ def add_context(*args, **kwargs):
                 args, inputs_state_indices
             )
 
-            response = {"multi": True}
+            response: dict = {"multi": True}
             has_update = False
 
             if long is not None:
@@ -413,7 +421,6 @@ def add_context(*args, **kwargs):
                             triggered_inputs=callback_ctx.triggered_inputs,
                             ignore_register_page=True,
                         ),
-                        on_error=on_error or app_on_error,
                     )
 
                     data = {
@@ -451,10 +458,19 @@ def add_context(*args, **kwargs):
                     isinstance(output_value, dict)
                     and "long_callback_error" in output_value
                 ):
-                    error = output_value.get("long_callback_error")
-                    raise LongCallbackError(
+                    error = output_value.get("long_callback_error", {})
+                    exc = LongCallbackError(
                         f"An error occurred inside a long callback: {error['msg']}\n{error['tb']}"
                     )
+                    if error_handler:
+                        error_handler(exc)
+                        output_value = NoUpdate()
+                        # set_props from the error handler uses the original ctx
+                        # instead of manager.get_updated_props since it runs in the
+                        # request process.
+                        has_update = _set_side_update(callback_ctx, response)
+                    else:
+                        raise exc
 
                 if job_running and output_value is not callback_manager.UNDEFINED:
                     # cached results.
@@ -478,16 +494,15 @@ def add_context(*args, **kwargs):
                 except PreventUpdate as err:
                     raise err
                 except Exception as err:  # pylint: disable=broad-exception-caught
-                    if on_error:
-                        output_value = on_error(err)
-                    elif app_on_error:
-                        output_value = app_on_error(err)
+                    if error_handler:
+                        error_handler(err)
+                        if not multi:
+                            output_value = NoUpdate()
+                        else:
+                            output_value = [NoUpdate for _ in output_spec]
                     else:
                         raise err
 
-            if NoUpdate.is_no_update(output_value):
-                raise PreventUpdate
-
             component_ids = collections.defaultdict(dict)
 
             if has_output:
@@ -508,12 +523,12 @@ def add_context(*args, **kwargs):
                 )
 
                 for val, spec in zip(flat_output_values, output_spec):
-                    if isinstance(val, NoUpdate):
+                    if NoUpdate.is_no_update(val):
                         continue
                     for vali, speci in (
                         zip(val, spec) if isinstance(spec, list) else [[val, spec]]
                     ):
-                        if not isinstance(vali, NoUpdate):
+                        if not NoUpdate.is_no_update(vali):
                             has_update = True
                             id_str = stringify_id(speci["id"])
                             prop = clean_property_name(speci["property"])
@@ -527,10 +542,7 @@ def add_context(*args, **kwargs):
                 flat_output_values = []
 
             if not long:
-                side_update = dict(callback_ctx.updated_props)
-                if len(side_update) > 0:
-                    has_update = True
-                    response["sideUpdate"] = side_update
+                has_update = _set_side_update(callback_ctx, response) or has_update
 
             if not has_update:
                 raise PreventUpdate

dash/long_callback/managers/__init__.py

@@ -39,7 +39,7 @@ def job_running(self, job):
     def make_job_fn(self, fn, progress, key=None):
         raise NotImplementedError
 
-    def call_job_fn(self, key, job_fn, args, context, on_error=None):
+    def call_job_fn(self, key, job_fn, args, context):
         raise NotImplementedError
 
     def get_progress(self, key):

dash/long_callback/managers/celery_manager.py

@@ -89,8 +89,8 @@ def get_task(self, job):
     def clear_cache_entry(self, key):
         self.handle.backend.delete(key)
 
-    def call_job_fn(self, key, job_fn, args, context, on_error=None):
-        task = job_fn.delay(key, self._make_progress_key(key), args, context, on_error)
+    def call_job_fn(self, key, job_fn, args, context):
+        task = job_fn.delay(key, self._make_progress_key(key), args, context)
         return task.task_id
 
     def get_progress(self, key):
@@ -139,9 +139,7 @@ def _make_job_fn(fn, celery_app, progress, key):
     cache = celery_app.backend
 
     @celery_app.task(name=f"long_callback_{key}")
-    def job_fn(
-        result_key, progress_key, user_callback_args, context=None, on_error=None
-    ):
+    def job_fn(result_key, progress_key, user_callback_args, context=None):
         def _set_progress(progress_value):
             if not isinstance(progress_value, (list, tuple)):
                 progress_value = [progress_value]
@@ -181,21 +179,18 @@ def run():
                     ),
                 )
             except Exception as err:  # pylint: disable=broad-except
-                if on_error:
-                    user_callback_output = on_error(err)
-                else:
-                    errored = True
-                    cache.set(
-                        result_key,
-                        json.dumps(
-                            {
-                                "long_callback_error": {
-                                    "msg": str(err),
-                                    "tb": traceback.format_exc(),
-                                }
-                            },
-                        ),
-                    )
+                errored = True
+                cache.set(
+                    result_key,
+                    json.dumps(
+                        {
+                            "long_callback_error": {
+                                "msg": str(err),
+                                "tb": traceback.format_exc(),
+                            }
+                        },
+                    ),
+                )
 
             if not errored:
                 cache.set(

dash/long_callback/managers/diskcache_manager.py

@@ -115,14 +115,14 @@ def clear_cache_entry(self, key):
         self.handle.delete(key)
 
     # noinspection PyUnresolvedReferences
-    def call_job_fn(self, key, job_fn, args, context, on_error=None):
+    def call_job_fn(self, key, job_fn, args, context):
         # pylint: disable-next=import-outside-toplevel,no-name-in-module,import-error
         from multiprocess import Process
 
         # pylint: disable-next=not-callable
         proc = Process(
             target=job_fn,
-            args=(key, self._make_progress_key(key), args, context, on_error),
+            args=(key, self._make_progress_key(key), args, context),
         )
         proc.start()
         return proc.pid
@@ -169,7 +169,7 @@ def get_updated_props(self, key):
 
 
 def _make_job_fn(fn, cache, progress):
-    def job_fn(result_key, progress_key, user_callback_args, context, on_error):
+    def job_fn(result_key, progress_key, user_callback_args, context):
         def _set_progress(progress_value):
             if not isinstance(progress_value, (list, tuple)):
                 progress_value = [progress_value]
@@ -200,19 +200,16 @@ def run():
                 errored = True
                 cache.set(result_key, {"_dash_no_update": "_dash_no_update"})
             except Exception as err:  # pylint: disable=broad-except
-                if on_error:
-                    user_callback_output = on_error(err)
-                else:
-                    errored = True
-                    cache.set(
-                        result_key,
-                        {
-                            "long_callback_error": {
-                                "msg": str(err),
-                                "tb": traceback.format_exc(),
-                            }
-                        },
-                    )
+                errored = True
+                cache.set(
+                    result_key,
+                    {
+                        "long_callback_error": {
+                            "msg": str(err),
+                            "tb": traceback.format_exc(),
+                        }
+                    },
+                )
 
             if not errored:
                 cache.set(result_key, user_callback_output)

tests/integration/callbacks/test_callback_error.py

@@ -1,9 +1,9 @@
-from dash import Dash, html, Input, Output
+from dash import Dash, html, Input, Output, set_props
 
 
 def test_cber001_error_handler(dash_duo):
     def global_callback_error_handler(err):
-        return f"global: {err}"
+        set_props("output-global", {"children": f"global: {err}"})
 
     app = Dash(on_error=global_callback_error_handler)
 
@@ -15,7 +15,7 @@ def global_callback_error_handler(err):
     ]
 
     def on_callback_error(err):
-        return f"callback: {err}"
+        set_props("output", {"children": f"callback: {err}"})
 
     @app.callback(
         Output("output", "children"),

tests/integration/long_callback/app_bg_on_error.py

@@ -1,13 +1,12 @@
-from dash import Dash, Input, Output, html
+from dash import Dash, Input, Output, html, set_props
 from tests.integration.long_callback.utils import get_long_callback_manager
 
 long_callback_manager = get_long_callback_manager()
 handle = long_callback_manager.handle
 
 
 def global_error_handler(err):
-
-    return f"global: {err}"
+    set_props("global-output", {"children": f"global: {err}"})
 
 
 app = Dash(
@@ -23,7 +22,7 @@ def global_error_handler(err):
 
 
 def callback_on_error(err):
-    return f"callback: {err}"
+    set_props("cb-output", {"children": f"callback: {err}"})
 
 
 @app.callback(

tests/integration/long_callback/test_basic_long_callback018.py

@@ -7,7 +7,7 @@ def test_lcbc018_background_callback_on_error(dash_duo, manager):
 
         dash_duo.find_element("#start-cb-onerror").click()
 
-        dash_duo.wait_for_text_to_equal("#cb-output", "callback: callback error")
+        dash_duo.wait_for_contains_text("#cb-output", "callback error")
 
         dash_duo.find_element("#start-global-onerror").click()
-        dash_duo.wait_for_text_to_equal("#global-output", "global: global error")
+        dash_duo.wait_for_contains_text("#global-output", "global error")

tests/integration/long_callback/utils.py

@@ -18,8 +18,14 @@ def __init__(self, cache=None, cache_by=None, expire=None):
         super().__init__(cache=cache, cache_by=cache_by, expire=expire)
         self.running_jobs = []
 
-    def call_job_fn(self, key, job_fn, args, context, on_error=None):
-        pid = super().call_job_fn(key, job_fn, args, context, on_error=on_error)
+    def call_job_fn(
+        self,
+        key,
+        job_fn,
+        args,
+        context,
+    ):
+        pid = super().call_job_fn(key, job_fn, args, context)
         self.running_jobs.append(pid)
         return pid