fixing for lint and progress outputs

Author: BSd3v

dash/_callback.py

@@ -300,7 +300,7 @@ def _set_side_update(ctx, response) -> bool:
     return False
 
 
-# pylint: disable=too-many-branches,too-many-statements
+# pylint: disable=too-many-branches,too-many-statements,too-many-locals
 def register_callback(
     callback_list, callback_map, config_prevent_initial_callbacks, *_args, **_kwargs
 ):
@@ -475,24 +475,9 @@ def update_long_callback(error_handler, callback_ctx, response, kwargs):
         cache_key = flask.request.args.get("cacheKey")
         job_id = flask.request.args.get("job")
 
-        output_value = callback_manager.get_result(cache_key, job_id)
-
         progress_long_callback(response, callback_manager)
 
-        return handle_rest_long_callback(
-            output_value, callback_manager, response, error_handler, callback_ctx
-        )
-
-    async def async_update_long_callback(error_handler, callback_ctx, response, kwargs):
-        """Set up the long callback and manage jobs."""
-        callback_manager = get_callback_manager(kwargs)
-
-        cache_key = flask.request.args.get("cacheKey")
-        job_id = flask.request.args.get("job")
-
-        output_value = await callback_manager.async_get_result(cache_key, job_id)
-
-        progress_long_callback(response, callback_manager)
+        output_value = callback_manager.get_result(cache_key, job_id)
 
         return handle_rest_long_callback(
             output_value, callback_manager, response, error_handler, callback_ctx

dash/long_callback/managers/celery_manager.py

@@ -1,6 +1,8 @@
 import json
 import traceback
 from contextvars import copy_context
+import asyncio
+from functools import partial
 
 from _plotly_utils.utils import PlotlyJSONEncoder
 
@@ -9,8 +11,6 @@
 from dash.exceptions import PreventUpdate
 from dash.long_callback._proxy_set_props import ProxySetProps
 from dash.long_callback.managers import BaseLongCallbackManager
-import asyncio
-from functools import partial
 
 
 class CeleryManager(BaseLongCallbackManager):
@@ -92,14 +92,7 @@ def clear_cache_entry(self, key):
         self.handle.backend.delete(key)
 
     def call_job_fn(self, key, job_fn, args, context):
-        if asyncio.iscoroutinefunction(job_fn):
-            # pylint: disable-next=import-outside-toplevel,no-name-in-module,import-error
-            from asgiref.sync import async_to_sync
-
-            new_job_fun = async_to_sync(job_fn)
-            task = new_job_fun.delay(key, self._make_progress_key(key), args, context)
-        else:
-            task = job_fn.delay(key, self._make_progress_key(key), args, context)
+        task = job_fn.delay(key, self._make_progress_key(key), args, context)
         return task.task_id
 
     def get_progress(self, key):
@@ -144,11 +137,13 @@ def get_updated_props(self, key):
         return json.loads(updated_props)
 
 
-def _make_job_fn(fn, celery_app, progress, key):
+def _make_job_fn(fn, celery_app, progress, key):  # pylint: disable=too-many-statements
     cache = celery_app.backend
 
     @celery_app.task(name=f"long_callback_{key}")
-    def job_fn(result_key, progress_key, user_callback_args, context=None):
+    def job_fn(
+        result_key, progress_key, user_callback_args, context=None
+    ):  # pylint: disable=too-many-statements
         def _set_progress(progress_value):
             if not isinstance(progress_value, (list, tuple)):
                 progress_value = [progress_value]
@@ -224,21 +219,31 @@ async def async_run():
                 else:
                     user_callback_output = await fn(*maybe_progress, user_callback_args)
             except PreventUpdate:
+                # Put NoUpdate dict directly to avoid circular imports.
                 errored = True
-                cache.set(result_key, {"_dash_no_update": "_dash_no_update"})
+                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,
-                    {
-                        "long_callback_error": {
-                            "msg": str(err),
-                            "tb": traceback.format_exc(),
-                        }
-                    },
+                    json.dumps(
+                        {
+                            "long_callback_error": {
+                                "msg": str(err),
+                                "tb": traceback.format_exc(),
+                            }
+                        },
+                    ),
                 )
+
             if asyncio.iscoroutine(user_callback_output):
                 user_callback_output = await user_callback_output
+
             if not errored:
                 cache.set(
                     result_key, json.dumps(user_callback_output, cls=PlotlyJSONEncoder)

dash/long_callback/managers/diskcache_manager.py

@@ -1,13 +1,13 @@
 import traceback
 from contextvars import copy_context
+import asyncio
+from functools import partial
 
 from . import BaseLongCallbackManager
 from .._proxy_set_props import ProxySetProps
 from ..._callback_context import context_value
 from ..._utils import AttributeDict
 from ...exceptions import PreventUpdate
-import asyncio
-from functools import partial
 
 _pending_value = "__$pending__"
 
@@ -131,24 +131,13 @@ 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
 
-        # Check if the job is asynchronous
-        if asyncio.iscoroutinefunction(job_fn):
-            # For async jobs, run in an event loop in a new process
-            process = Process(
-                target=self._run_async_in_process,
-                args=(job_fn, key, args, context),
-            )
-            process.start()
-            return process.pid
-        else:
-            # For sync jobs, use the existing implementation
-            # pylint: disable-next=not-callable
-            process = Process(
-                target=job_fn,
-                args=(key, self._make_progress_key(key), args, context),
-            )
-            process.start()
-            return process.pid
+        # pylint: disable-next=not-callable
+        process = Process(
+            target=job_fn,
+            args=(key, self._make_progress_key(key), args, context),
+        )
+        process.start()
+        return process.pid
 
     @staticmethod
     def _run_async_in_process(job_fn, key, args, context):
@@ -172,7 +161,7 @@ def _run_async_in_process(job_fn, key, args, context):
             loop.run_until_complete(async_job())
         except Exception as e:
             # Handle errors, log them, and cache if necessary
-            raise str(e)
+            raise Exception(str(e)) from e
         finally:
             loop.close()
 
@@ -217,7 +206,9 @@ def get_updated_props(self, key):
         return result
 
 
+# pylint: disable-next=too-many-statements
 def _make_job_fn(fn, cache, progress):
+    # pylint: disable-next=too-many-statements
     def job_fn(result_key, progress_key, user_callback_args, context):
         def _set_progress(progress_value):
             if not isinstance(progress_value, (list, tuple)):