Merge remote-tracking branch 'origin/main' into reda-venv-refactor

flojoy/__init__.py

@@ -13,3 +13,4 @@
 from .job_service import *
 from .node_init import *
 from .config import *
+from .node_preflight import *

flojoy/__init__.pyi

@@ -13,6 +13,7 @@ from .small_memory import *
 from .flojoy_node_venv import *
 from .job_service import *
 from .node_init import *
+from .node_preflight import *
 from .data_container import *
 from .config import *
 
@@ -287,10 +288,10 @@ def snapshot_download(
     ...
 
 def flojoy(
-    original_function: Callable[..., DataContainer | dict[str, Any] | TypedDict]
+    original_function: Callable[..., DataContainer | dict[str, Any] | TypedDict | None]
     | None = None,
     *,
     node_type: Optional[str] = None,
     deps: Optional[dict[str, str]] = None,
     inject_node_metadata: bool = False,
-) -> Callable[..., DataContainer | dict[str, Any]]: ...
+) -> Callable[..., DataContainer | dict[str, Any] | None]: ...

flojoy/flojoy_python.py

@@ -86,7 +86,8 @@ def __init__(
 
 
 def flojoy(
-    original_function: Callable[..., DataContainer | dict[str, Any]] | None = None,
+    original_function: Callable[..., Optional[DataContainer | dict[str, Any]]]
+    | None = None,
     *,
     node_type: Optional[str] = None,
     deps: Optional[dict[str, str]] = None,
@@ -135,7 +136,7 @@ def SINE(dc_inputs:list[DataContainer], params:dict[str, Any]):
     ```
     """
 
-    def decorator(func: Callable[..., DataContainer | dict[str, Any]]):
+    def decorator(func: Callable[..., Optional[DataContainer | dict[str, Any]]]):
         @wraps(func)
         def wrapper(
             node_id: str,
@@ -209,12 +210,13 @@ def wrapper(
                 # some special nodes like LOOP return dict instead of `DataContainer`
                 if isinstance(dc_obj, DataContainer):
                     dc_obj.validate()  # Validate returned DataContainer object
-                else:
+                elif dc_obj is not None:
                     for value in dc_obj.values():
                         if isinstance(value, DataContainer):
                             value.validate()
                 # Response object to send to FE
                 result = get_frontend_res_obj_from_result(dc_obj)
+
                 JobService().post_job_result(
                     job_id, dc_obj
                 )  # post result to the job service before sending result to socket
@@ -249,8 +251,7 @@ def wrapper(
                     json.dumps(
                         {
                             "SYSTEM_STATUS": f"Failed to run: {func.__name__}",
-                            "FAILED_NODES": node_id,
-                            "FAILURE_REASON": e.args[0],
+                            "FAILED_NODES": {node_id: str(e)},
                             "jobsetId": jobset_id,
                         }
                     )

flojoy/job_result_builder.py

@@ -8,7 +8,7 @@ class JobResultBuilder:
     instructions: dict[str, Union[str, list[str]]] | None = None
 
     def __init__(self) -> None:
-        self.data = self.get_default_data()
+        self.data = None
 
     def _add_instructions(self, instruction: dict[str, Union[str, list[str]]]):
         self.instructions = self.instructions if self.instructions is not None else {}
@@ -20,7 +20,7 @@ def _add_instructions(self, instruction: dict[str, Union[str, list[str]]]):
     def from_inputs(self, inputs: list[DataContainer]):
         # if no inputs were provided, construct fake output
         if len(inputs) == 0:
-            self.data = self.get_default_data()
+            self.data = None
         else:
             self.data = inputs[0]
 

flojoy/job_result_utils.py

@@ -2,7 +2,7 @@
 from .plotly_utils import data_container_to_plotly
 from .data_container import DataContainer
 from .dao import Dao
-from typing import Any, cast
+from typing import Any, cast, Optional
 
 __all__ = ["get_job_result", "get_next_directions", "get_next_nodes", "get_job_result"]
 
@@ -72,15 +72,19 @@ def get_text_blob_from_dc(dc: DataContainer) -> str | None:
 
 
 def get_frontend_res_obj_from_result(
-    result: dict[str, Any] | DataContainer
-) -> dict[str, Any]:
-    if not result:
-        return {"plotly_fig": result}
+    result: Optional[dict[str, Any] | DataContainer]
+) -> Optional[dict[str, Any]]:
+    if result is None:
+        return None
+
     if isinstance(result, DataContainer):
         plotly_fig = data_container_to_plotly(data=result)
         return {"plotly_fig": plotly_fig, "text_blob": get_text_blob_from_dc(result)}
     if result.get(FLOJOY_INSTRUCTION.RESULT_FIELD):
         data = result[result[FLOJOY_INSTRUCTION.RESULT_FIELD]]
+        if not data:
+            return result
+
         plotly_fig = None
         if isinstance(data, DataContainer):
             plotly_fig = data_container_to_plotly(data=data)

flojoy/node_preflight.py

@@ -0,0 +1,3 @@
+def node_preflight(func):
+    func.is_flojoy_preflight = True
+    return func