feat: transport Exceptions from tasks over our websocket connection

renumics/spotlight/backend/tasks/reduction.py

@@ -7,7 +7,6 @@
 import numpy as np
 import pandas as pd
 
-from renumics.spotlight.dataset.exceptions import ColumnNotExistsError
 from renumics.spotlight.data_store import DataStore
 from renumics.spotlight import dtypes
 
@@ -80,10 +79,8 @@ def compute_umap(
     Prepare data from table and compute U-Map on them.
     """
 
-    try:
-        data, indices = align_data(data_store, column_names, indices)
-    except (ColumnNotExistsError, ColumnNotEmbeddable):
-        return np.empty(0, np.float64), []
+    data, indices = align_data(data_store, column_names, indices)
+
     if data.size == 0:
         return np.empty(0, np.float64), []
 
@@ -116,14 +113,13 @@ def compute_pca(
     Prepare data from table and compute PCA on them.
     """
 
-    from sklearn import preprocessing, decomposition
+    data, indices = align_data(data_store, column_names, indices)
 
-    try:
-        data, indices = align_data(data_store, column_names, indices)
-    except (ColumnNotExistsError, ColumnNotEmbeddable):
-        return np.empty(0, np.float64), []
     if data.size == 0:
         return np.empty(0, np.float64), []
+
+    from sklearn import preprocessing, decomposition
+
     if data.shape[1] == 1:
         return np.hstack((data, np.zeros_like(data))), indices
     if normalization == "standardize":

renumics/spotlight/backend/tasks/task_manager.py

@@ -6,7 +6,7 @@
 import multiprocessing
 from concurrent.futures import Future, ProcessPoolExecutor
 from concurrent.futures.process import BrokenProcessPool
-from typing import Any, Callable, List, Optional, Sequence, TypeVar, Union
+from typing import Any, Callable, Dict, List, Optional, Sequence, TypeVar, Union
 
 from .exceptions import TaskCancelled
 from .task import Task
@@ -30,16 +30,20 @@ def create_task(
         self,
         func: Callable,
         args: Sequence[Any],
+        kwargs: Optional[Dict[str, Any]] = None,
         name: Optional[str] = None,
         tag: Optional[Union[str, int]] = None,
     ) -> Task:
         """
         create and launch a new task
         """
+        if kwargs is None:
+            kwargs = {}
+
         # cancel running task with same name
         self.cancel(name=name)
 
-        future = self.pool.submit(func, *args)
+        future = self.pool.submit(func, *args, **kwargs)
 
         task = Task(name, tag, future)
         self.tasks.append(task)
@@ -59,14 +63,15 @@ async def run_async(
         self,
         func: Callable[..., T],
         args: Sequence[Any],
+        kwargs: Optional[Dict[str, Any]] = None,
         name: Optional[str] = None,
         tag: Optional[Union[str, int]] = None,
     ) -> T:
         """
         Launch a new task. Await and return result.
         """
 
-        task = self.create_task(func, args, name, tag)
+        task = self.create_task(func, args=args, kwargs=kwargs, name=name, tag=tag)
         try:
             return await asyncio.wrap_future(task.future)
         except BrokenProcessPool as e: