Merge pull request #95 from common-workflow-language/scoped-ref-and-typedsl

Scoped ref and typedsl

Author: mr-c

cwltool/builder.py

@@ -34,9 +34,9 @@ class Builder(object):
     def __init__(self):  # type: () -> None
         self.names = None  # type: avro.schema.Names
         self.schemaDefs = None  # type: Dict[str,Dict[unicode, Any]]
-        self.files = None  # type: List[Dict[str, str]]
+        self.files = None  # type: List[Dict[unicode, unicode]]
         self.fs_access = None  # type: StdFsAccess
-        self.job = None  # type: Dict[str, Any]
+        self.job = None  # type: Dict[unicode, Union[Dict[unicode, Any], List, unicode]]
         self.requirements = None  # type: List[Dict[str,Any]]
         self.outdir = None  # type: str
         self.tmpdir = None  # type: str

cwltool/cwlrdf.py

@@ -6,7 +6,7 @@
 from typing import Any, Union, Dict, IO
 
 def makerdf(workflow, wf, ctx):
-    # type: (Union[str, unicode], Dict[str,Any], Loader.ContextType) -> Graph
+    # type: (Union[str, unicode], Union[List[Dict[unicode, Any]], Dict[unicode, Any]], Loader.ContextType) -> Graph
     prefixes = {}
     for k,v in ctx.iteritems():
         if isinstance(v, dict):
@@ -18,7 +18,11 @@ def makerdf(workflow, wf, ctx):
             p, _ = frg.split("/")
             prefixes[p] = u"%s#%s/" % (doc_url, p)
 
-    wf["@context"] = ctx
+    if isinstance(wf, list):
+        for entry in wf:
+            entry["@context"] = ctx
+    else:
+        wf["@context"] = ctx
     g = Graph().parse(data=json.dumps(wf), format='json-ld', location=workflow)
 
     # Bug in json-ld loader causes @id fields to be added to the graph
@@ -31,7 +35,7 @@ def makerdf(workflow, wf, ctx):
     return g
 
 def printrdf(workflow, wf, ctx, sr, stdout):
-    # type: (Union[str, unicode], Dict[str, Any], Loader.ContextType, str, IO[Any]) -> None
+    # type: (Union[str, unicode], Union[List[Dict[unicode, Any]], Dict[unicode, Any]], Loader.ContextType, str, IO[Any]) -> None
     stdout.write(makerdf(workflow, wf, ctx).serialize(format=sr))
 
 def lastpart(uri):  # type: (Any) -> str
@@ -172,7 +176,7 @@ def dot_without_parameters(g, stdout):  # type: (Graph, IO[Any]) -> None
 
 
 def printdot(workflow, wf, ctx, stdout, include_parameters=False):
-    # type: (Union[str, unicode], Dict[str, Any], Loader.ContextType, Any, bool) -> None
+    # type: (Union[str, unicode], Union[List[Dict[unicode, Any]], Dict[unicode, Any]], Loader.ContextType, Any, bool) -> None
     g = makerdf(workflow, wf, ctx)
 
     stdout.write("digraph {")

cwltool/draft2tool.py

@@ -30,7 +30,7 @@
 
 class ExpressionTool(Process):
     def __init__(self, toolpath_object, **kwargs):
-        # type: (Dict[str,List[None]], **Any) -> None
+        # type: (Dict[unicode, Any], **Any) -> None
         super(ExpressionTool, self).__init__(toolpath_object, **kwargs)
 
     class ExpressionJob(object):
@@ -53,7 +53,7 @@ def run(self, **kwargs):  # type: (**Any) -> None
                 self.output_callback({}, "permanentFail")
 
     def job(self, joborder, output_callback, **kwargs):
-        # type: (Dict[str,str], str, Callable[[Any, Any], Any], **Any) -> Generator[ExpressionTool.ExpressionJob, None, None]
+        # type: (Dict[unicode, unicode], Callable[[Any, Any], Any], **Any) -> Generator[ExpressionTool.ExpressionJob, None, None]
         builder = self._init_job(joborder, **kwargs)
 
         j = ExpressionTool.ExpressionJob()
@@ -113,14 +113,14 @@ def run(self, **kwargs):
 
 class CommandLineTool(Process):
     def __init__(self, toolpath_object, **kwargs):
-        # type: (Dict[str,Any], **Any) -> None
+        # type: (Dict[unicode, Any], **Any) -> None
         super(CommandLineTool, self).__init__(toolpath_object, **kwargs)
 
     def makeJobRunner(self):  # type: () -> CommandLineJob
         return CommandLineJob()
 
     def makePathMapper(self, reffiles, **kwargs):
-        # type: (Set[str], str, **Any) -> PathMapper
+        # type: (Set[unicode], **Any) -> PathMapper
         dockerReq, _ = self.get_requirement("DockerRequirement")
         try:
             if dockerReq and kwargs.get("use_container"):
@@ -132,7 +132,7 @@ def makePathMapper(self, reffiles, **kwargs):
                 raise WorkflowException(u"Missing input file %s" % e)
 
     def job(self, joborder, output_callback, **kwargs):
-        # type: (Dict[str,str], str, Callable[..., Any], **Any) -> Generator[Union[CommandLineJob, CallbackJob], None, None]
+        # type: (Dict[unicode, unicode], Callable[..., Any], **Any) -> Generator[Union[CommandLineJob, CallbackJob], None, None]
 
         jobname = uniquename(kwargs.get("name", shortname(self.tool.get("id", "job"))))
 
@@ -149,7 +149,7 @@ def job(self, joborder, output_callback, **kwargs):
             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
-            keydict = {"cmdline": cmdline}
+            keydict = {u"cmdline": cmdline}
 
             for _,f in cachebuilder.pathmapper.items():
                 st = os.stat(f[0])
@@ -200,7 +200,7 @@ def rm_pending_output_callback(output_callback, jobcachepending,
 
         builder = self._init_job(joborder, **kwargs)
 
-        reffiles = set((f["path"] for f in builder.files))
+        reffiles = set((f[u"path"] for f in builder.files))
 
         j = self.makeJobRunner()
         j.builder = builder
@@ -298,9 +298,9 @@ def _check_adjust(f):  # type: (Dict[str,Any]) -> Dict[str,Any]
         yield j
 
     def collect_output_ports(self, ports, builder, outdir):
-        # type: (Set[Dict[str,Any]], Builder, str) -> Dict[str,Union[str,List[Any],Dict[str,Any]]]
+        # type: (Set[Dict[str,Any]], Builder, str) -> Dict[unicode, Union[unicode, List[Any], Dict[unicode, Any]]]
         try:
-            ret = {}  # type: Dict[str,Union[str,List[Any],Dict[str,Any]]]
+            ret = {}  # type: Dict[unicode, Union[unicode, List[Any], Dict[unicode, Any]]]
             custom_output = os.path.join(outdir, "cwl.output.json")
             if builder.fs_access.exists(custom_output):
                 with builder.fs_access.open(custom_output, "r") as f:
@@ -329,7 +329,7 @@ def collect_output_ports(self, ports, builder, outdir):
             raise WorkflowException("Error validating output record, " + str(e) + "\n in " + json.dumps(ret, indent=4))
 
     def collect_output(self, schema, builder, outdir):
-        # type: (Dict[str,Any], Builder, str) -> Union[Dict[str, Any], List[Union[Dict[str, Any], str]]]
+        # type: (Dict[str,Any], Builder, str) -> Union[Dict[unicode, Any], List[Union[Dict[unicode, Any], unicode]]]
         r = []  # type: List[Any]
         if "outputBinding" in schema:
             binding = schema["outputBinding"]

cwltool/expression.py

@@ -19,7 +19,7 @@ def jshead(engineConfig, rootvars):
     return u"\n".join(engineConfig + [u"var %s = %s;" % (k, json.dumps(v, indent=4)) for k, v in rootvars.items()])
 
 def exeval(ex, jobinput, requirements, outdir, tmpdir, context, pull_image):
-    # type: (Dict[str,Any], Dict[str,str], List[Dict[str, Any]], str, str, Any, bool) -> sandboxjs.JSON
+    # type: (Dict[str, Any], Dict[unicode, Union[Dict, List, unicode]], List[Dict[str, Any]], str, str, Any, bool) -> sandboxjs.JSON
 
     if ex["engine"] == "https://w3id.org/cwl/cwl#JavascriptEngine":
         engineConfig = []  # type: List[unicode]
@@ -126,7 +126,7 @@ def param_interpolate(ex, obj, strip=True):
 
 def do_eval(ex, jobinput, requirements, outdir, tmpdir, resources,
             context=None, pull_image=True, timeout=None):
-    # type: (Any, Dict[str,str], List[Dict[str,Any]], str, str, Dict[str, Union[int, str]], Any, bool, int) -> Any
+    # type: (Union[dict, unicode], Dict[unicode, Union[Dict, List, unicode]], List[Dict[str, Any]], str, str, Dict[str, Union[int, str]], Any, bool, int) -> Any
 
     runtime = resources.copy()
     runtime["tmpdir"] = tmpdir
@@ -140,7 +140,7 @@ def do_eval(ex, jobinput, requirements, outdir, tmpdir, resources,
 
     if isinstance(ex, dict) and "engine" in ex and "script" in ex:
         return exeval(ex, jobinput, requirements, outdir, tmpdir, context, pull_image)
-    if isinstance(ex, basestring):
+    if isinstance(ex, (str, unicode)):
         for r in requirements:
             if r["class"] == "InlineJavascriptRequirement":
                 return sandboxjs.interpolate(str(ex), jshead(r.get("expressionLib", []), rootvars),

cwltool/job.py

@@ -42,7 +42,7 @@ class CommandLineJob(object):
 
     def __init__(self):  # type: () -> None
         self.builder = None  # type: Builder
-        self.joborder = None  # type: Dict[str,str]
+        self.joborder = None  # type: Dict[unicode, Union[Dict[unicode, Any], List, unicode]]
         self.stdin = None  # type: str
         self.stderr = None  # type: str
         self.stdout = None  # type: str

cwltool/load_tool.py

@@ -44,7 +44,7 @@ def fetch_document(argsworkflow):
 
 def validate_document(document_loader, workflowobj, uri,
                       enable_dev=False, strict=True, preprocess_only=False):
-    # type: (Loader, Dict[unicode, Any], unicode, bool, bool, bool) -> Tuple[Loader, Names, Any, Dict[str, str], unicode]
+    # type: (Loader, Dict[unicode, Any], unicode, bool, bool, bool) -> Tuple[Loader, Names, Union[Dict[unicode, Any], List[Dict[unicode, Any]]], Dict[unicode, Any], unicode]
     """Validate a CWL document."""
     jobobj = None
     if "cwl:tool" in workflowobj:
@@ -83,31 +83,34 @@ def validate_document(document_loader, workflowobj, uri,
 
     workflowobj["id"] = fileuri
     processobj, metadata = document_loader.resolve_all(workflowobj, fileuri)
+    if not isinstance(processobj, (dict, list)):
+        raise validate.ValidationException("Workflow must be a dict or list.")
 
     if not metadata:
+        if not isinstance(processobj, dict):
+            raise validate.ValidationException(
+                    "Draft-2 workflows must be a dict.")
         metadata = {"$namespaces": processobj.get("$namespaces", {}),
                    "$schemas": processobj.get("$schemas", []),
                    "cwlVersion": processobj["cwlVersion"]}
 
     if preprocess_only:
         return document_loader, avsc_names, processobj, metadata, uri
 
-    document_loader.validate_links(processobj)
     schema.validate_doc(avsc_names, processobj, document_loader, strict)
 
     if metadata.get("cwlVersion") != update.LATEST:
         processobj = update.update(
             processobj, document_loader, fileuri, enable_dev, metadata)
 
     if jobobj:
-        metadata["cwl:defaults"] = jobobj
+        metadata[u"cwl:defaults"] = jobobj
 
     return document_loader, avsc_names, processobj, metadata, uri
 
 
-def make_tool(document_loader, avsc_names, processobj, metadata, uri, makeTool,
-              kwargs):
-    # type: (Loader, Names, Dict[str, Any], Dict[str, Any], unicode, Callable[..., Process], Dict[str, Any]) -> Process
+def make_tool(document_loader, avsc_names, metadata, uri, makeTool, kwargs):
+    # type: (Loader, Names, Dict[unicode, Any], unicode, Callable[..., Process], Dict[str, Any]) -> Process
     """Make a Python CWL object."""
     resolveduri = document_loader.resolve_ref(uri)[0]
 
@@ -121,7 +124,7 @@ def make_tool(document_loader, avsc_names, processobj, metadata, uri, makeTool,
                     urlparse.urldefrag(i["id"])[1] for i in resolveduri
                     if "id" in i))
     else:
-        processobj = cast(Dict[str, Any], resolveduri)
+        processobj = resolveduri
 
     kwargs = kwargs.copy()
     kwargs.update({
@@ -149,5 +152,5 @@ def load_tool(argsworkflow, makeTool, kwargs=None,
     document_loader, avsc_names, processobj, metadata, uri = validate_document(
         document_loader, workflowobj, uri, enable_dev=enable_dev,
         strict=strict)
-    return make_tool(document_loader, avsc_names, processobj, metadata, uri,
+    return make_tool(document_loader, avsc_names, metadata, uri,
                      makeTool, kwargs if kwargs else {})