Refactor wip

Author: Peter Amstutz

cwltool/builder.py

@@ -45,7 +45,6 @@ def __init__(self):  # type: () -> None
         self.build_job_script = None  # type: Callable[[List[str]], Text]
         self.debug = False  # type: bool
         self.mutation_manager = None  # type: MutationManager
-        self.tool_id = None  # type: Text
 
         # One of "no_listing", "shallow_listing", "deep_listing"
         # Will be default "no_listing" for CWL v1.1

cwltool/draft2tool.py

@@ -80,7 +80,6 @@ def job(self,
         j.hints = self.hints
         j.outdir = None
         j.tmpdir = None
-        j.tool_id = self.tool["id"]
 
         yield j
 
@@ -176,9 +175,9 @@ def __init__(self, toolpath_object, **kwargs):
         # type: (Dict[Text, Any], **Any) -> None
         super(CommandLineTool, self).__init__(toolpath_object, **kwargs)
 
-    def makeJobRunner(self, **kwargs):  # type: (Optional[bool]) -> JobBase
-        dockerReq, _ = self.get_requirement("DockerRequirement", kwargs)
-        if dockerReq and kwargs.get("use_container", True):
+    def makeJobRunner(self, use_container=True):  # type: (Optional[bool]) -> JobBase
+        dockerReq, _ = self.get_requirement("DockerRequirement")
+        if dockerReq and use_container:
             return DockerCommandLineJob()
         else:
             for t in reversed(self.requirements):
@@ -216,7 +215,7 @@ def job(self,
                        ("File", "Directory"), _check_adjust)
 
             cmdline = flatten(map(cachebuilder.generate_arg, cachebuilder.bindings))
-            (docker_req, docker_is_req) = self.get_requirement("DockerRequirement", kwargs)
+            (docker_req, docker_is_req) = self.get_requirement("DockerRequirement")
             if docker_req and kwargs.get("use_container") is not False:
                 dockerimg = docker_req.get("dockerImageId") or docker_req.get("dockerPull")
                 cmdline = ["docker", "run", dockerimg] + cmdline
@@ -277,7 +276,7 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
 
         reffiles = copy.deepcopy(builder.files)
 
-        j = self.makeJobRunner(**kwargs)
+        j = self.makeJobRunner(kwargs.get("use_container"))
         j.builder = builder
         j.joborder = builder.job
         j.stdin = None
@@ -289,8 +288,6 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
         j.requirements = self.requirements
         j.hints = self.hints
         j.name = jobname
-        j.tool_id = self.tool["id"]
-        j.overrides = kwargs.get("overrides", [])
 
         if _logger.isEnabledFor(logging.DEBUG):
             _logger.debug(u"[job %s] initializing from %s%s",
@@ -335,7 +332,7 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
         if _logger.isEnabledFor(logging.DEBUG):
             _logger.debug(u"[job %s] command line bindings is %s", j.name, json.dumps(builder.bindings, indent=4))
 
-        dockerReq = self.get_requirement("DockerRequirement", kwargs)[0]
+        dockerReq = self.get_requirement("DockerRequirement")[0]
         if dockerReq and kwargs.get("use_container"):
             out_prefix = kwargs.get("tmp_outdir_prefix")
             j.outdir = kwargs.get("outdir") or tempfile.mkdtemp(prefix=out_prefix)
@@ -347,7 +344,7 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
             j.tmpdir = builder.tmpdir
             j.stagedir = builder.stagedir
 
-        initialWorkdir = self.get_requirement("InitialWorkDirRequirement", kwargs)[0]
+        initialWorkdir = self.get_requirement("InitialWorkDirRequirement")[0]
         j.generatefiles = {"class": "Directory", "listing": [], "basename": ""}
         if initialWorkdir:
             ls = []  # type: List[Dict[Text, Any]]
@@ -383,7 +380,7 @@ def rm_pending_output_callback(output_callbacks, jobcachepending,
                         ls[i] = t["entry"]
             j.generatefiles[u"listing"] = ls
 
-        inplaceUpdateReq = self.get_requirement("http://commonwl.org/cwltool#InplaceUpdateRequirement", kwargs)[0]
+        inplaceUpdateReq = self.get_requirement("http://commonwl.org/cwltool#InplaceUpdateRequirement")[0]
 
         if inplaceUpdateReq:
             j.inplace_update = inplaceUpdateReq["inplaceUpdate"]
@@ -417,12 +414,12 @@ def register_reader(f):
             adjustDirObjs(builder.bindings, register_reader)
 
         j.environment = {}
-        evr = self.get_requirement("EnvVarRequirement", kwargs)[0]
+        evr = self.get_requirement("EnvVarRequirement")[0]
         if evr:
             for t in evr["envDef"]:
                 j.environment[t["envName"]] = builder.do_eval(t["envValue"])
 
-        shellcmd = self.get_requirement("ShellCommandRequirement", kwargs)[0]
+        shellcmd = self.get_requirement("ShellCommandRequirement")[0]
         if shellcmd:
             cmd = []  # type: List[Text]
             for b in builder.bindings:

cwltool/job.py

@@ -127,7 +127,6 @@ def __init__(self):  # type: () -> None
         self.generatefiles = None  # type: Dict[Text, Union[List[Dict[Text, Text]], Dict[Text, Text], Text]]
         self.stagedir = None  # type: Text
         self.inplace_update = None  # type: bool
-        self.overrides = None  # type: List[Dict[Text,Dict[Text, Text]]]
 
     def _setup(self):  # type: () -> None
         if not os.path.exists(self.outdir):
@@ -150,7 +149,7 @@ def _setup(self):  # type: () -> None
     def _execute(self, runtime, env, rm_tmpdir=True, move_outputs="move"):
         # type: (List[Text], MutableMapping[Text, Text], bool, Text) -> None
 
-        scr, _ = get_feature(self, "ShellCommandRequirement", {})
+        scr, _ = get_feature(self, "ShellCommandRequirement")
 
         shouldquote = None  # type: Callable[[Any], Any]
         if scr:
@@ -321,7 +320,7 @@ def run(self, pull_image=True, rm_container=True,
             rm_tmpdir=True, move_outputs="move", **kwargs):
         # type: (bool, bool, bool, Text, **Any) -> Union[Tuple[Text, Dict[None, None]], None]
 
-        (docker_req, docker_is_req) = get_feature(self, "DockerRequirement", kwargs)
+        (docker_req, docker_is_req) = get_feature(self, "DockerRequirement")
 
         img_id = None
         env = None  # type: MutableMapping[Text, Text]

cwltool/main.py

@@ -230,19 +230,20 @@ def output_callback(out, processStatus):
     output_dirs.add(kwargs["outdir"])
     kwargs["mutation_manager"] = MutationManager()
 
-    jobReqs = []
+    jobReqs = None
     if "cwl:requirements" in job_order_object:
         jobReqs = job_order_object["cwl:requirements"]
         del job_order_object["cwl:requirements"]
     elif ("cwl:defaults" in t.metadata and "cwl:requirements" in t.metadata["cwl:defaults"]):
         jobReqs = t.metadata["cwl:defaults"]["cwl:requirements"]
+    if jobReqs:
+        for req in jobReqs:
+            t.requirements.append(req)
 
     if "http://commonwl.org/cwltool#overrides" in job_order_object:
         kwargs["overrides"] = resolve_overrides(job_order_object, t.tool["id"])
         del job_order_object["http://commonwl.org/cwltool#overrides"]
 
-    kwargs["requirements"] = jobReqs
-
     if kwargs.get("default_container"):
         t.requirements.insert(0, {
             "class": "DockerRequirement",
@@ -428,15 +429,12 @@ def generate_parser(toolparser, tool, namemap, records):
     return toolparser
 
 
-def load_job_order(args, t, stdin, print_input_deps=False, relative_deps=False,
-                   stdout=sys.stdout, make_fs_access=None, fetcher_constructor=None):
+def load_job_order(args, stdin, fetcher_constructor=None):
     # type: (argparse.Namespace, Process, IO[Any], bool, bool, IO[Any], Callable[[Text], StdFsAccess], Callable[[Dict[unicode, unicode], requests.sessions.Session], Fetcher]) -> Union[int, Tuple[Dict[Text, Any], Text]]
 
-    job_order_object = None
+    job_order_object = {}
 
-    _jobloaderctx = jobloaderctx.copy()
-    _jobloaderctx.update(t.metadata.get("$namespaces", {}))
-    loader = Loader(_jobloaderctx, fetcher_constructor=fetcher_constructor)
+    loader = Loader(jobloaderctx, fetcher_constructor=fetcher_constructor)
 
     if len(args.job_order) == 1 and args.job_order[0][0] != "-":
         job_order_file = args.job_order[0]
@@ -454,48 +452,53 @@ def load_job_order(args, t, stdin, print_input_deps=False, relative_deps=False,
             job_order_object, _ = loader.resolve_ref(job_order_file, checklinks=False)
         except Exception as e:
             _logger.error(Text(e), exc_info=args.debug)
-            return 1
+            return (1, None)
         toolparser = None
-    else:
-        input_basedir = args.basedir if args.basedir else os.getcwd()
-        namemap = {}  # type: Dict[Text, Text]
-        records = []  # type: List[Text]
-        toolparser = generate_parser(
-            argparse.ArgumentParser(prog=args.workflow), t, namemap, records)
-        if toolparser:
-            if args.tool_help:
-                toolparser.print_help()
-                return 0
-            cmd_line = vars(toolparser.parse_args(args.job_order))
-            for record_name in records:
-                record = {}
-                record_items = {
-                    k: v for k, v in cmd_line.iteritems()
-                    if k.startswith(record_name)}
-                for key, value in record_items.iteritems():
-                    record[key[len(record_name) + 1:]] = value
-                    del cmd_line[key]
-                cmd_line[str(record_name)] = record
-
-            if cmd_line["job_order"]:
-                try:
-                    input_basedir = args.basedir if args.basedir else os.path.abspath(
-                        os.path.dirname(cmd_line["job_order"]))
-                    job_order_object = loader.resolve_ref(cmd_line["job_order"])
-                except Exception as e:
-                    _logger.error(Text(e), exc_info=args.debug)
-                    return 1
-            else:
-                job_order_object = {"id": args.workflow}
 
-            del cmd_line["job_order"]
+    return (job_order_object, loader)
+
+def job_order_from_command_line(args, t, loader, input_basedir,
+                                print_input_deps=False,
+                                relative_deps=False,
+                                stdout=sys.stdout):
+    namemap = {}  # type: Dict[Text, Text]
+    records = []  # type: List[Text]
+    toolparser = generate_parser(
+        argparse.ArgumentParser(prog=args.workflow), t, namemap, records)
+    if toolparser:
+        if args.tool_help:
+            toolparser.print_help()
+            return 0
+        cmd_line = vars(toolparser.parse_args(args.job_order))
+        for record_name in records:
+            record = {}
+            record_items = {
+                k: v for k, v in cmd_line.iteritems()
+                if k.startswith(record_name)}
+            for key, value in record_items.iteritems():
+                record[key[len(record_name) + 1:]] = value
+                del cmd_line[key]
+            cmd_line[str(record_name)] = record
+
+        if cmd_line["job_order"]:
+            try:
+                input_basedir = args.basedir if args.basedir else os.path.abspath(
+                    os.path.dirname(cmd_line["job_order"]))
+                job_order_object, _ = loader.resolve_ref(cmd_line["job_order"])
+            except Exception as e:
+                _logger.error(Text(e), exc_info=args.debug)
+                return 1
+        else:
+            job_order_object = {"id": args.workflow}
 
-            job_order_object.update({namemap[k]: v for k, v in cmd_line.items()})
+        del cmd_line["job_order"]
 
-            if _logger.isEnabledFor(logging.DEBUG):
-                _logger.debug(u"Parsed job order from command line: %s", json.dumps(job_order_object, indent=4))
-        else:
-            job_order_object = None
+        job_order_object.update({namemap[k]: v for k, v in cmd_line.items()})
+
+        if _logger.isEnabledFor(logging.DEBUG):
+            _logger.debug(u"Parsed job order from command line: %s", json.dumps(job_order_object, indent=4))
+    else:
+        job_order_object = {}
 
     for inp in t.tool["inputs"]:
         if "default" in inp and (not job_order_object or shortname(inp["id"]) not in job_order_object):
@@ -509,13 +512,16 @@ def load_job_order(args, t, stdin, print_input_deps=False, relative_deps=False,
             toolparser.print_help()
         _logger.error("")
         _logger.error("Input object required, use --help for details")
-        return 1
+        return (1, None)
 
     if print_input_deps:
         printdeps(job_order_object, loader, stdout, relative_deps, "",
                   basedir=file_uri(input_basedir + "/"))
-        return 0
+        return (0, None)
 
+    return (job_order_object, input_basedir)
+
+def normalize_job_order(job_order_object):
     def pathToLoc(p):
         if "location" not in p and "path" in p:
             p["location"] = p["path"]
@@ -530,8 +536,6 @@ def pathToLoc(p):
     if "id" in job_order_object:
         del job_order_object["id"]
 
-    return (job_order_object, input_basedir)
-
 
 def makeRelative(base, ob):
     u = ob.get("location", ob.get("path"))
@@ -701,6 +705,21 @@ def main(argsl=None,  # type: List[str]
         else:
             use_standard_schema("v1.0")
 
+        try:
+            if job_order_object is None:
+                job_order_object, job_order_loader = load_job_order(args, stdin,
+                                                                      fetcher_constructor=fetcher_constructor)
+            else:
+                job_order_loader = None
+        except SystemExit as e:
+            return e.code
+
+        if isinstance(job_order_object, int):
+            return job_order_object
+
+        if args.overrides:
+            args.overrides = load_overrides(file_uri(os.path.abspath(args.overrides)), args.workflow)
+
         try:
             document_loader, workflowobj, uri = fetch_document(args.workflow, resolver=resolver,
                                                                fetcher_constructor=fetcher_constructor)
@@ -774,28 +793,23 @@ def main(argsl=None,  # type: List[str]
                 setattr(args, 'move_outputs', "copy")
             setattr(args, "tmp_outdir_prefix", args.cachedir)
 
-        try:
-            if job_order_object is None:
-                    job_order_object = load_job_order(args, tool, stdin,
-                                                      print_input_deps=args.print_input_deps,
-                                                      relative_deps=args.relative_deps,
-                                                      stdout=stdout,
-                                                      make_fs_access=make_fs_access,
-                                                      fetcher_constructor=fetcher_constructor)
-        except SystemExit as e:
-            return e.code
-
-        if isinstance(job_order_object, int):
-            return job_order_object
+        input_basedir = args.basedir if args.basedir else os.getcwd()
+        if job_order_loader:
+            jofc, input_basedir = job_order_from_command_line(args, tool, job_order_loader, input_basedir,
+                                                              print_input_deps=args.print_input_deps,
+                                                              relative_deps=args.relative_deps,
+                                                              stdout=stdout)
+            if isinstance(jofc, int):
+                return jofc
+            job_order_object.update(jofc)
 
-        if args.overrides:
-            args.overrides = load_overrides(file_uri(os.path.abspath(args.overrides)), tool.tool["id"])
+        normalize_job_order(job_order_object)
 
         try:
-            setattr(args, 'basedir', job_order_object[1])
+            setattr(args, 'basedir', input_basedir)
             del args.workflow
             del args.job_order
-            (out, status) = executor(tool, job_order_object[0],
+            (out, status) = executor(tool, job_order_object,
                                      makeTool=makeTool,
                                      select_resources=selectResources,
                                      make_fs_access=make_fs_access,

cwltool/process.py

@@ -424,10 +424,12 @@ def __init__(self, toolpath_object, **kwargs):
             self.names = names
         self.tool = toolpath_object
         self.requirements = kwargs.get("requirements", []) + self.tool.get("requirements", [])
+        for ov in reversed(kwargs.get("overrides", [])):
+            if ov.get("overrideTarget") == self.tool["id"]:
+                self.requirements.extend(ov.get("override", []))
+
         self.hints = kwargs.get("hints", []) + self.tool.get("hints", [])
         self.formatgraph = None  # type: Graph
-        self.tool_id = self.tool["id"]
-
         if "loader" in kwargs:
             self.formatgraph = kwargs["loader"].graph
 
@@ -534,11 +536,11 @@ def _init_job(self, joborder, **kwargs):
         builder.debug = kwargs.get("debug")
         builder.mutation_manager = kwargs.get("mutation_manager")
 
-        dockerReq, is_req = self.get_requirement("DockerRequirement", kwargs)
+        dockerReq, is_req = self.get_requirement("DockerRequirement")
         builder.make_fs_access = kwargs.get("make_fs_access") or StdFsAccess
         builder.fs_access = builder.make_fs_access(kwargs["basedir"])
 
-        loadListingReq, _ = self.get_requirement("http://commonwl.org/cwltool#LoadListingRequirement", kwargs)
+        loadListingReq, _ = self.get_requirement("http://commonwl.org/cwltool#LoadListingRequirement")
         if loadListingReq:
             builder.loadListing = loadListingReq.get("loadListing")
 
@@ -604,7 +606,7 @@ def _init_job(self, joborder, **kwargs):
 
     def evalResources(self, builder, kwargs):
         # type: (Builder, Dict[AnyStr, Any]) -> Dict[Text, Union[int, Text]]
-        resourceReq, _ = self.get_requirement("ResourceRequirement", kwargs)
+        resourceReq, _ = self.get_requirement("ResourceRequirement")
         if resourceReq is None:
             resourceReq = {}
         request = {
@@ -657,8 +659,8 @@ def validate_hints(self, avsc_names, hints, strict):
                 else:
                     _logger.info(sl.makeError(u"Unknown hint %s" % (r["class"])))
 
-    def get_requirement(self, feature, kwargs={}):  # type: (Any) -> Tuple[Any, bool]
-        return get_feature(self, feature, kwargs)
+    def get_requirement(self, feature):  # type: (Any) -> Tuple[Any, bool]
+        return get_feature(self, feature)
 
     def visit(self, op):  # type: (Callable[[Dict[Text, Any]], None]) -> None
         op(self.tool)