Refactor to use loading / runtime context instead of kwargs (#788)

* Major refactor to get rid of kwargs and use well defined context objects.
* Unwind incorrect changes to Process object constructor interface.
* Remove get_feature, use get_requirement everywhere.

Author: tetron

README.rst

@@ -541,13 +541,13 @@ executor
   A toplevel workflow execution loop, should synchronously execute a process
   object and return an output object.
 
-makeTool
+construct_tool_object
   ::
 
-    makeTool(toolpath_object, **kwargs)
+    construct_tool_object(toolpath_object, **kwargs)
       (Dict[Text, Any], **Any) -> Process
 
-  Construct a Process object from a document.
+  Hook to construct a Process object (eg CommandLineTool) object from a document.
 
 selectResources
   ::

cwltool/builder.py

@@ -3,7 +3,7 @@
 import copy
 import logging
 from typing import (Any, Callable, Dict, List,  # pylint: disable=unused-import
-                    Optional, Set, Text, Type, Union)
+                    Optional, Set, Text, Type, Union, Tuple)
 
 from rdflib import Graph, URIRef  # pylint: disable=unused-import
 from rdflib.namespace import OWL, RDFS
@@ -20,7 +20,7 @@
 from .pathmapper import (PathMapper,  # pylint: disable=unused-import
                          get_listing, normalizeFilesDirs, visit_class)
 from .stdfsaccess import StdFsAccess  # pylint: disable=unused-import
-from .utils import (aslist, docker_windows_path_adjust, get_feature,
+from .utils import (aslist, docker_windows_path_adjust,
                     json_dumps, onWindows)
 
 CONTENT_LIMIT = 64 * 1024
@@ -85,7 +85,23 @@ def check_format(actual_file,    # type: Union[Dict[Text, Any], List, Text]
             u"File has an incompatible format: {}".format(
                 json_dumps(afile, indent=4)))
 
-class Builder(object):
+class HasReqsHints(object):
+    def __init__(self):
+        self.requirements = []  # List[Dict[Text, Any]]
+        self.hints = []         # List[Dict[Text, Any]]
+
+    def get_requirement(self,
+                    feature  # type: Text
+                   ):  # type: (...) -> Tuple[Optional[Any], Optional[bool]]
+        for item in reversed(self.requirements):
+            if item["class"] == feature:
+                return (item, True)
+        for item in reversed(self.hints):
+            if item["class"] == feature:
+                return (item, False)
+        return (None, None)
+
+class Builder(HasReqsHints):
     def __init__(self,
                  job,                  # type: Dict[Text, Union[Dict[Text, Any], List, Text]]
                  files,                # type: List[Dict[Text, Text]]
@@ -96,9 +112,9 @@ def __init__(self,
                  hints,                # type: List[Dict[Text, Any]]
                  timeout,              # type: float
                  debug,                # type: bool
-                 resources,            # type: Dict[Text, Union[int, Text, None]]
+                 resources,            # type: Dict[Text, int]
                  js_console,           # type: bool
-                 mutation_manager,     # type: MutationManager
+                 mutation_manager,     # type: Optional[MutationManager]
                  formatgraph,          # type: Optional[Graph]
                  make_fs_access,       # type: Type[StdFsAccess]
                  fs_access,            # type: StdFsAccess
@@ -107,7 +123,7 @@ def __init__(self,
                  outdir,               # type: Text
                  tmpdir,               # type: Text
                  stagedir,             # type: Text
-                 job_script_provider,  # type: Optional[Any]
+                 job_script_provider   # type: Optional[Any]
                 ):  # type: (...) -> None
         self.names = names
         self.schemaDefs = schemaDefs
@@ -299,7 +315,7 @@ def tostr(self, value):  # type: (Any) -> Text
                 raise WorkflowException(u"%s object missing \"path\": %s" % (value["class"], value))
 
             # Path adjust for windows file path when passing to docker, docker accepts unix like path only
-            (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
+            (docker_req, docker_is_req) = self.get_requirement("DockerRequirement")
             if onWindows() and docker_req is not None:
                 # docker_req is none only when there is no dockerRequirement
                 # mentioned in hints and Requirement

cwltool/command_line_tool.py

@@ -40,6 +40,7 @@
 from .utils import (aslist, convert_pathsep_to_unix,
                     docker_windows_path_adjust, json_dumps, onWindows,
                     windows_default_container_id)
+from .context import LoadingContext, RuntimeContext, getdefault
 
 ACCEPTLIST_EN_STRICT_RE = re.compile(r"^[a-zA-Z0-9._+-]+$")
 ACCEPTLIST_EN_RELAXED_RE = re.compile(r".*")  # Accept anything
@@ -80,25 +81,23 @@ def __init__(self,
             self.tmpdir = tmpdir
             self.script = script
 
-        def run(self, **kwargs):  # type: (**Any) -> None
+        def run(self, runtimeContext):  # type: (RuntimeContext) -> None
             try:
                 ev = self.builder.do_eval(self.script)
                 normalizeFilesDirs(ev)
                 self.output_callback(ev, "success")
             except Exception as e:
                 _logger.warning(u"Failed to evaluate expression:\n%s",
-                             e, exc_info=kwargs.get('debug'))
+                             e, exc_info=runtimeContext.debug)
                 self.output_callback({}, "permanentFail")
 
     def job(self,
             job_order,         # type: Dict[Text, Text]
             output_callbacks,  # type: Callable[[Any, Any], Any]
-            mutation_manager,  # type: MutationManager
-            basedir,           # type: Text
-            **kwargs           # type: Any
+            runtimeContext     # type: RuntimeContext
            ):
         # type: (...) -> Generator[ExpressionTool.ExpressionJob, None, None]
-        builder = self._init_job(job_order, mutation_manager, basedir, **kwargs)
+        builder = self._init_job(job_order, runtimeContext)
 
         yield ExpressionTool.ExpressionJob(
             builder, self.tool["expression"], output_callbacks,
@@ -169,13 +168,13 @@ def __init__(self, job, output_callback, cachebuilder, jobcache):
         self.cachebuilder = cachebuilder
         self.outdir = jobcache
 
-    def run(self, **kwargs):
-        # type: (**Any) -> None
+    def run(self, runtimeContext):
+        # type: (RuntimeContext) -> None
         self.output_callback(self.job.collect_output_ports(
             self.job.tool["outputs"],
             self.cachebuilder,
             self.outdir,
-            kwargs.get("compute_checksum", True)), "success")
+            getdefault(runtimeContext.compute_checksum, True)), "success")
 
 
 def check_adjust(builder, file_o):
@@ -216,33 +215,29 @@ def make_path_mapper(reffiles, stagedir, basedir, separateDirs=True):
 OutputPorts = Dict[Text, Union[None, Text, List[Union[Dict[Text, Any], Text]], Dict[Text, Any]]]
 
 class CommandLineTool(Process):
-    def __init__(self, toolpath_object, eval_timeout, debug, js_console,
-                 force_docker_pull, job_script_provider, **kwargs):
-        # type: (Dict[Text, Any], float, bool, bool, bool, Optional[DependenciesConfiguration], **Any) -> None
+    def __init__(self, toolpath_object, loadingContext):
+        # type: (Dict[Text, Any], LoadingContext) -> None
         super(CommandLineTool, self).__init__(
-            toolpath_object, eval_timeout, debug, js_console,
-            force_docker_pull, job_script_provider, **kwargs)
-        self.find_default_container = kwargs.get("find_default_container", None)
+            toolpath_object, loadingContext)
 
     def make_job_runner(self,
-                        use_container=True,  # type: Optional[bool]
-                        **kwargs             # type: Any
+                        runtimeContext       # type: RuntimeContext
                        ):  # type: (...) -> Type[JobBase]
         dockerReq, _ = self.get_requirement("DockerRequirement")
-        if not dockerReq and use_container:
-            if self.find_default_container:
-                default_container = self.find_default_container(self)
+        if not dockerReq and runtimeContext.use_container:
+            if runtimeContext.find_default_container:
+                default_container = runtimeContext.find_default_container(self)
                 if default_container:
                     self.requirements.insert(0, {
                         "class": "DockerRequirement",
                         "dockerPull": default_container
                     })
                     dockerReq = self.requirements[0]
-                    if default_container == windows_default_container_id and use_container and onWindows():
+                    if default_container == windows_default_container_id and runtimeContext.use_container and onWindows():
                         _logger.warning(DEFAULT_CONTAINER_MSG % (windows_default_container_id, windows_default_container_id))
 
-        if dockerReq and use_container:
-            if kwargs.get('singularity'):
+        if dockerReq and runtimeContext.use_container:
+            if runtimeContext.singularity:
                 return SingularityCommandLineJob
             else:
                 return DockerCommandLineJob
@@ -268,9 +263,7 @@ def updatePathmap(self, outdir, pathmap, fn):
     def job(self,
             job_order,         # type: Dict[Text, Text]
             output_callbacks,  # type: Callable[[Any, Any], Any]
-            mutation_manager,  # type: MutationManager
-            basedir,           # type: Text
-            **kwargs           # type: Any
+            runtimeContext     # RuntimeContext
             ):
         # type: (...) -> Generator[Union[JobBase, CallbackJob], None, None]
 
@@ -281,15 +274,15 @@ def job(self,
         workReuse = self.get_requirement(require_prefix+"WorkReuse")[0]
         enableReuse = workReuse.get("enableReuse", True) if workReuse else True
 
-        jobname = uniquename(kwargs.get("name", shortname(self.tool.get("id", "job"))))
-        if kwargs.get("cachedir") and enableReuse:
-            cacheargs = kwargs.copy()
-            cacheargs["outdir"] = "/out"
-            cacheargs["tmpdir"] = "/tmp"
-            cacheargs["stagedir"] = "/stage"
-            cachebuilder = self._init_job(job_order, mutation_manager, basedir, **cacheargs)
+        jobname = uniquename(runtimeContext.name or shortname(self.tool.get("id", "job")))
+        if runtimeContext.cachedir and enableReuse:
+            cachecontext = runtimeContext.copy()
+            cachecontext.outdir = "/out"
+            cachecontext.tmpdir = "/tmp"
+            cachecontext.stagedir = "/stage"
+            cachebuilder = self._init_job(job_order, cachecontext)
             cachebuilder.pathmapper = PathMapper(cachebuilder.files,
-                                                 basedir,
+                                                 runtimeContext.basedir,
                                                  cachebuilder.stagedir,
                                                  separateDirs=False)
             _check_adjust = partial(check_adjust, cachebuilder)
@@ -298,10 +291,10 @@ def job(self,
 
             cmdline = flatten(list(map(cachebuilder.generate_arg, cachebuilder.bindings)))
             (docker_req, docker_is_req) = self.get_requirement("DockerRequirement")
-            if docker_req and kwargs.get("use_container"):
+            if docker_req and runtimeContext.use_container:
                 dockerimg = docker_req.get("dockerImageId") or docker_req.get("dockerPull")
-            elif kwargs.get("default_container", None) is not None and kwargs.get("use_container"):
-                dockerimg = kwargs.get("default_container")
+            elif runtimeContext.default_container is not None and runtimeContext.use_container:
+                dockerimg = runtimeContext.default_container
             else:
                 dockerimg = None
 
@@ -340,12 +333,12 @@ def job(self,
             _logger.debug("[job %s] keydictstr is %s -> %s", jobname,
                           keydictstr, cachekey)
 
-            jobcache = os.path.join(kwargs["cachedir"], cachekey)
+            jobcache = os.path.join(runtimeContext.cachedir, cachekey)
             jobcachepending = jobcache + ".pending"
 
             if os.path.isdir(jobcache) and not os.path.isfile(jobcachepending):
-                if docker_req and kwargs.get("use_container"):
-                    cachebuilder.outdir = kwargs.get("docker_outdir") or "/var/spool/cwl"
+                if docker_req and runtimeContext.use_container:
+                    cachebuilder.outdir = runtimeContext.docker_outdir or "/var/spool/cwl"
                 else:
                     cachebuilder.outdir = jobcache
 
@@ -356,7 +349,8 @@ def job(self,
                 _logger.info("[job %s] Output of job will be cached in %s", jobname, jobcache)
                 shutil.rmtree(jobcache, True)
                 os.makedirs(jobcache)
-                kwargs["outdir"] = jobcache
+                runtimeContext = runtimeContext.copy()
+                runtimeContext.outdir = jobcache
                 open(jobcachepending, "w").close()
 
                 def rm_pending_output_callback(output_callbacks, jobcachepending,
@@ -367,18 +361,12 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
 
                 output_callbacks = partial(
                     rm_pending_output_callback, output_callbacks, jobcachepending)
-                # output_callbacks = cast(
-                #     Callable[..., Any],  # known bug in mypy
-                #     # https://github.com/python/mypy/issues/797
-                #     partial(rm_pending_output_callback, output_callbacks,
-                #             jobcachepending))
 
-
-        builder = self._init_job(job_order, mutation_manager, basedir, **kwargs)
+        builder = self._init_job(job_order, runtimeContext)
 
         reffiles = copy.deepcopy(builder.files)
 
-        j = self.make_job_runner(**kwargs)(
+        j = self.make_job_runner(runtimeContext)(
             builder, builder.job, make_path_mapper, self.requirements,
             self.hints, jobname)
         j.successCodes = self.tool.get("successCodes")
@@ -391,13 +379,13 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
             _logger.debug(u"[job %s] initializing from %s%s",
                           j.name,
                           self.tool.get("id", ""),
-                          u" as part of %s" % kwargs["part_of"]
-                          if "part_of" in kwargs else "")
+                          u" as part of %s" % runtimeContext.part_of
+                          if runtimeContext.part_of else "")
             _logger.debug(u"[job %s] %s", j.name, json_dumps(job_order,
                                                              indent=4))
 
         builder.pathmapper = make_path_mapper(
-            reffiles, builder.stagedir, basedir, kwargs.get("separateDirs", True))
+            reffiles, builder.stagedir, runtimeContext.basedir, True)
         builder.requirements = j.requirements
 
         _check_adjust = partial(check_adjust, builder)
@@ -475,12 +463,12 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
                           json_dumps(builder.bindings, indent=4))
 
         dockerReq = self.get_requirement("DockerRequirement")[0]
-        if dockerReq and kwargs.get("use_container"):
-            out_prefix = kwargs.get("tmp_outdir_prefix", 'tmp')
-            j.outdir = kwargs.get("outdir") or \
+        if dockerReq and runtimeContext.use_container:
+            out_prefix = getdefault(runtimeContext.tmp_outdir_prefix, 'tmp')
+            j.outdir = runtimeContext.outdir or \
                 tempfile.mkdtemp(prefix=out_prefix)  # type: ignore
-            tmpdir_prefix = kwargs.get('tmpdir_prefix', 'tmp')
-            j.tmpdir = kwargs.get("tmpdir") or \
+            tmpdir_prefix = getdefault(runtimeContext.tmpdir_prefix, 'tmp')
+            j.tmpdir = runtimeContext.tmpdir or \
                 tempfile.mkdtemp(prefix=tmpdir_prefix)  # type: ignore
             j.stagedir = tempfile.mkdtemp(prefix=tmpdir_prefix)
         else:
@@ -494,8 +482,8 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
             j.inplace_update = inplaceUpdateReq["inplaceUpdate"]
         normalizeFilesDirs(j.generatefiles)
 
-        readers = {}
-        muts = set()
+        readers = {}  # type: Dict[Text, Any]
+        muts = set()  # type: Set[Text]
 
         if builder.mutation_manager:
             def register_mut(f):
@@ -558,7 +546,7 @@ def register_reader(f):
         j.pathmapper = builder.pathmapper
         j.collect_outputs = partial(
             self.collect_output_ports, self.tool["outputs"], builder,
-            compute_checksum=kwargs.get("compute_checksum", True),
+            compute_checksum=getdefault(runtimeContext.compute_checksum, True),
             jobname=jobname,
             readers=readers)
         j.output_callback = output_callbacks