match Process.job def in subtypes

Author: mr-c

cwltool/draft2tool.py

@@ -60,14 +60,18 @@ def run(self, **kwargs):  # type: (**Any) -> None
                              e, exc_info=kwargs.get('debug'))
                 self.output_callback({}, "permanentFail")
 
-    def job(self, joborder, output_callback, **kwargs):
-        # type: (Dict[Text, Text], Callable[[Any, Any], Any], **Any) -> Generator[ExpressionTool.ExpressionJob, None, None]
-        builder = self._init_job(joborder, **kwargs)
+    def job(self,
+            job_order,  # type: Dict[Text, Text]
+            output_callbacks,  # type: Callable[[Any, Any], Any]
+            **kwargs  # type: Any
+            ):
+        # type: (...) -> Generator[ExpressionTool.ExpressionJob, None, None]
+        builder = self._init_job(job_order, **kwargs)
 
         j = ExpressionTool.ExpressionJob()
         j.builder = builder
         j.script = self.tool["expression"]
-        j.output_callback = output_callback
+        j.output_callback = output_callbacks
         j.requirements = self.requirements
         j.hints = self.hints
         j.outdir = None
@@ -165,8 +169,12 @@ def makePathMapper(self, reffiles, stagedir, **kwargs):
         dockerReq, _ = self.get_requirement("DockerRequirement")
         return PathMapper(reffiles, kwargs["basedir"], stagedir)
 
-    def job(self, joborder, output_callback, **kwargs):
-        # type: (Dict[Text, Text], Callable[..., Any], **Any) -> Generator[Union[CommandLineJob, CallbackJob], None, None]
+    def job(self,
+            job_order,  # type: Dict[Text, Text]
+            output_callbacks,  # type: Callable[[Any, Any], Any]
+            **kwargs  # type: Any
+            ):
+        # type: (...) -> Generator[Union[CommandLineJob, CallbackJob], None, None]
 
         jobname = uniquename(kwargs.get("name", shortname(self.tool.get("id", "job"))))
 
@@ -175,7 +183,7 @@ def job(self, joborder, output_callback, **kwargs):
             cacheargs["outdir"] = "/out"
             cacheargs["tmpdir"] = "/tmp"
             cacheargs["stagedir"] = "/stage"
-            cachebuilder = self._init_job(joborder, **cacheargs)
+            cachebuilder = self._init_job(job_order, **cacheargs)
             cachebuilder.pathmapper = PathMapper(cachebuilder.files,
                                                  kwargs["basedir"],
                                                  cachebuilder.stagedir,
@@ -222,7 +230,7 @@ def job(self, joborder, output_callback, **kwargs):
                     cachebuilder.outdir = jobcache
 
                 _logger.info("[job %s] Using cached output in %s", jobname, jobcache)
-                yield CallbackJob(self, output_callback, cachebuilder, jobcache)
+                yield CallbackJob(self, output_callbacks, cachebuilder, jobcache)
                 return
             else:
                 _logger.info("[job %s] Output of job will be cached in %s", jobname, jobcache)
@@ -231,19 +239,19 @@ def job(self, joborder, output_callback, **kwargs):
                 kwargs["outdir"] = jobcache
                 open(jobcachepending, "w").close()
 
-                def rm_pending_output_callback(output_callback, jobcachepending,
+                def rm_pending_output_callback(output_callbacks, jobcachepending,
                                                outputs, processStatus):
                     if processStatus == "success":
                         os.remove(jobcachepending)
-                    output_callback(outputs, processStatus)
+                    output_callbacks(outputs, processStatus)
 
-                output_callback = cast(
+                output_callbacks = cast(
                     Callable[..., Any],  # known bug in mypy
                     # https://github.com/python/mypy/issues/797
-                    partial(rm_pending_output_callback, output_callback,
+                    partial(rm_pending_output_callback, output_callbacks,
                             jobcachepending))
 
-        builder = self._init_job(joborder, **kwargs)
+        builder = self._init_job(job_order, **kwargs)
 
         reffiles = copy.deepcopy(builder.files)
 
@@ -265,7 +273,7 @@ def rm_pending_output_callback(output_callback, jobcachepending,
                           j.name,
                           self.tool.get("id", ""),
                           u" as part of %s" % kwargs["part_of"] if "part_of" in kwargs else "")
-            _logger.debug(u"[job %s] %s", j.name, json.dumps(joborder, indent=4))
+            _logger.debug(u"[job %s] %s", j.name, json.dumps(job_order, indent=4))
 
         builder.pathmapper = None
         make_path_mapper_kwargs = kwargs
@@ -377,7 +385,7 @@ def rm_pending_output_callback(output_callback, jobcachepending,
         j.collect_outputs = partial(
             self.collect_output_ports, self.tool["outputs"], builder,
             compute_checksum=kwargs.get("compute_checksum", True))
-        j.output_callback = output_callback
+        j.output_callback = output_callbacks
 
         yield j
 

cwltool/process.py

@@ -599,8 +599,12 @@ def visit(self, op):  # type: (Callable[[Dict[Text, Any]], None]) -> None
         op(self.tool)
 
     @abc.abstractmethod
-    def job(self, job_order, output_callbacks, **kwargs):
-        # type: (Dict[Text, Text], Callable[[Any, Any], Any], **Any) -> Generator[Any, None, None]
+    def job(self,
+            job_order,  # type: Dict[Text, Text]
+            output_callbacks,  # type: Callable[[Any, Any], Any]
+            **kwargs  # type: Any
+            ):
+        # type: (...) -> Generator[Any, None, None]
         return None
 
 

cwltool/workflow.py

@@ -436,15 +436,19 @@ def __init__(self, toolpath_object, **kwargs):
 
         # TODO: statically validate data links instead of doing it at runtime.
 
-    def job(self, joborder, output_callback, **kwargs):
-        # type: (Dict[Text, Text], Callable[[Any, Any], Any], **Any) -> Generator
-        builder = self._init_job(joborder, **kwargs)
+    def job(self,
+            job_order,  # type: Dict[Text, Text]
+            output_callbacks,  # type: Callable[[Any, Any], Any]
+            **kwargs  # type: Any
+            ):
+        # type: (...) -> Generator[Any, None, None]
+        builder = self._init_job(job_order, **kwargs)
         wj = WorkflowJob(self, **kwargs)
         yield wj
 
         kwargs["part_of"] = u"workflow %s" % wj.name
 
-        for w in wj.job(builder.job, output_callback, **kwargs):
+        for w in wj.job(builder.job, output_callbacks, **kwargs):
             yield w
 
     def visit(self, op):
@@ -561,17 +565,23 @@ def receive_output(self, output_callback, jobout, processStatus):
                 processStatus = "permanentFail"
         output_callback(output, processStatus)
 
-    def job(self, joborder, output_callback, **kwargs):
-        # type: (Dict[Text, Any], Callable[...,Any], **Any) -> Generator
+    def job(self,
+            job_order,  # type: Dict[Text, Text]
+            output_callbacks,  # type: Callable[[Any, Any], Any]
+            **kwargs  # type: Any
+            ):
+        # type: (...) -> Generator[Any, None, None]
         for i in self.tool["inputs"]:
             p = i["id"]
             field = shortname(p)
-            joborder[field] = joborder[i["id"]]
-            del joborder[i["id"]]
+            job_order[field] = job_order[i["id"]]
+            del job_order[i["id"]]
 
         try:
-            for t in self.embedded_tool.job(joborder,
-                                            functools.partial(self.receive_output, output_callback),
+            for t in self.embedded_tool.job(job_order,
+                                            functools.partial(
+                                                self.receive_output,
+                                                output_callbacks),
                                             **kwargs):
                 yield t
         except WorkflowException: