feat(fetchResults): deserialize results

Deserialize fetchResults results.

dcp/api/__init__.py

@@ -1,5 +1,6 @@
 from .compute_for import compute_for_maker
 from .compute_do import compute_do_maker
+from .fetch_results import fetch_results_maker
 from .job import job_maker
 from .result_handle import result_handle_maker 
 from .job_fs import JobFS
@@ -9,5 +10,5 @@
     'ResultHandle': result_handle_maker,
 }
 
-__all__ = ['compute_for_maker', 'compute_do_maker' 'sub_classes', 'JobFS']
+__all__ = ['compute_for_maker', 'compute_do_maker' 'sub_classes', 'JobFS', 'fetch_results_maker']
 

dcp/api/fetch_results.py

@@ -0,0 +1,26 @@
+"""
+fetchResults API
+
+Author: Will Pringle <[email protected]>
+Date: September 2024
+"""
+
+import pythonmonkey as pm
+from .. import dry
+from .. import js
+from .job_serializers import default_serializers, deserialize
+
+def fetch_results_maker(fetch_results_js):
+    def fetch_results(*args):
+        serialized_results = dry.aio.blockify(fetch_results_js)(*args)
+
+        results = []
+        for element in serialized_results:
+            deserialized_value = deserialize(element['value'], default_serializers)
+            slice_number = int(element['slice'])
+
+            results.append({ 'slice': slice_number, 'value': deserialized_value })
+
+        return results
+    return fetch_results 
+

dcp/api/job.py

@@ -154,6 +154,7 @@ def handle_complete(resultHandle):
                     result = deserialize(serialized_result, self.serializers)
                     results.append(result)
                 complete_future.set_result(results)
+            # TODO: on cancel, pass to cancel listener
             self.js_ref.on("complete", handle_complete)
             self.js_ref.on("cancel", handle_complete)
             return complete_future

dcp/initialization.py

@@ -69,7 +69,12 @@ def init_dcp_module(py_parent, js_module, js_name):
     setattr(py_parent, underscore_name, module)
 
     for prop_name, prop_ref in js_module.items():
-        setattr(module, prop_name, _wrap_js(prop_name, prop_ref))
+        # modify props if required
+        # TODO: should this be a dict of prop_name and list of modules with overridden ref?
+        if prop_name == 'fetchResults' and (js_name == 'compute' or js_name == 'job'):
+            setattr(module, prop_name, api.fetch_results_maker(prop_ref))
+        else:
+            setattr(module, prop_name, _wrap_js(prop_name, prop_ref))
 
 
 def make_init_fn(dcp_module) -> Callable:
@@ -104,7 +109,7 @@ def init(**kwargs) -> Module:
 
         # add some api top level imports
         setattr(dcp_module, "compute_for", api.compute_for_maker(class_manager.reg.find('Job')), )
-        setattr(dcp_module, "compute_do", api.compute_do_maker(class_manager.reg.find('Job')), )
+        setattr(dcp_module, "compute_do", api.compute_do_maker(class_manager.reg.find('Job')), ) #TODO unspec'd, why is this here?
         setattr(dcp_module, "JobFS", api.JobFS, )