Refactor add_sizes, add generalized size() call to StdFsAccess. (#855)

* Refactor add_sizes, add generalized size() call to StdFsAccess.

* Fix add_sizes import

* Mypy annotation fixes

Author: tetron

cwltool/builder.py

@@ -104,7 +104,7 @@ def get_requirement(self,
 
 class Builder(HasReqsHints):
     def __init__(self,
-                 job,                       # type: Dict[Text, Union[Dict[Text, Any], List, Text]]
+                 job,                       # type: Dict[Text, Union[Dict[Text, Any], List, Text, None]]
                  files=None,                # type: List[Dict[Text, Text]]
                  bindings=None,             # type: List[Dict[Text, Any]]
                  schemaDefs=None,           # type: Dict[Text, Dict[Text, Any]]
@@ -142,11 +142,6 @@ def __init__(self,
         else:
             self.files = files
 
-        if fs_access is None:
-            self.fs_access = StdFsAccess("")
-        else:
-            self.fs_access = fs_access
-
         self.job = job
         self.requirements = requirements
         self.hints = hints
@@ -171,6 +166,11 @@ def __init__(self,
         else:
             self.make_fs_access = make_fs_access
 
+        if fs_access is None:
+            self.fs_access = self.make_fs_access("")
+        else:
+            self.fs_access = fs_access
+
         self.debug = debug
         self.js_console = js_console
         self.mutation_manager = mutation_manager

cwltool/expression.py

@@ -245,7 +245,7 @@ def needs_parsing(snippet):  # type: (Any) -> bool
         and ("$(" in snippet or "${" in snippet)
 
 def do_eval(ex,                       # type: Union[Text, Dict]
-            jobinput,                 # type: Dict[Text, Union[Dict, List, Text]]
+            jobinput,                 # type: Dict[Text, Union[Dict, List, Text, None]]
             requirements,             # type: List[Dict[Text, Any]]
             outdir,                   # type: Optional[Text]
             tmpdir,                   # type: Optional[Text]

cwltool/job.py

@@ -136,7 +136,7 @@ def relink_initialworkdir(pathmapper, host_outdir, container_outdir, inplace_upd
 class JobBase(with_metaclass(ABCMeta, HasReqsHints)):
     def __init__(self,
                  builder,   # type: Builder
-                 joborder,  # type: Dict[Text, Union[Dict[Text, Any], List, Text]]
+                 joborder,  # type: Dict[Text, Union[Dict[Text, Any], List, Text, None]]
                  make_path_mapper,  # type: Callable[..., PathMapper]
                  requirements,  # type: List[Dict[Text, Text]]
                  hints,  # type: List[Dict[Text, Text]]

cwltool/main.py

@@ -39,15 +39,15 @@
 from .pathmapper import (adjustDirObjs, normalizeFilesDirs, trim_listing,
                          visit_class)
 from .process import (Process, scandeps,   # pylint: disable=unused-import
-                      shortname, use_custom_schema, use_standard_schema)
+                      shortname, use_custom_schema, use_standard_schema, add_sizes)
 from .provenance import ResearchObject
 from .resolver import ga4gh_tool_registries, tool_resolver
 from .secrets import SecretStore
 from .software_requirements import (DependenciesConfiguration,
                                     get_container_from_software_requirements)
 from .stdfsaccess import StdFsAccess
 from .update import ALLUPDATES, UPDATES
-from .utils import (DEFAULT_TMP_PREFIX, add_sizes, json_dumps, onWindows,
+from .utils import (DEFAULT_TMP_PREFIX, json_dumps, onWindows,
                     versionstring, windows_default_container_id,
                     processes_to_kill)
 from .context import LoadingContext, RuntimeContext, getdefault
@@ -183,7 +183,7 @@ def init_job_order(job_order_object,        # type: Optional[MutableMapping[Text
                    print_input_deps=False,  # type: bool
                    provArgs=None,           # type: ResearchObject
                    relative_deps=False,     # type: bool
-                   make_fs_access=None,     # type: Callable[[Text], StdFsAccess]
+                   make_fs_access=StdFsAccess,  # type: Callable[[Text], StdFsAccess]
                    input_basedir="",        # type: Text
                    secret_store=None        # type: SecretStore
                   ):  # type: (...) -> Tuple[MutableMapping[Text, Any], Optional[MutableMapping[Text, Any]]]
@@ -274,7 +274,7 @@ def expand_formats(p):
             p["format"] = ld.expand_url(p["format"], "")
 
     visit_class(job_order_object, ("File", "Directory"), pathToLoc)
-    visit_class(job_order_object, ("File",), add_sizes)
+    visit_class(job_order_object, ("File",), functools.partial(add_sizes, make_fs_access(input_basedir)))
     visit_class(job_order_object, ("File",), expand_formats)
     adjustDirObjs(job_order_object, trim_listing)
     normalizeFilesDirs(job_order_object)
@@ -599,12 +599,14 @@ def main(argsl=None,                   # type: List[str]
             runtimeContext.tmp_outdir_prefix = args.cachedir
 
         runtimeContext.secret_store = getdefault(runtimeContext.secret_store, SecretStore())
+        runtimeContext.make_fs_access = getdefault(runtimeContext.make_fs_access, StdFsAccess)
         try:
             initialized_job_order_object, input_for_prov = init_job_order(
                 job_order_object, args, tool, jobloader, stdout,
                 print_input_deps=args.print_input_deps,
                 provArgs=runtimeContext.research_obj,
                 relative_deps=args.relative_deps,
+                make_fs_access=runtimeContext.make_fs_access,
                 input_basedir=input_basedir,
                 secret_store=runtimeContext.secret_store)
         except SystemExit as err:
@@ -633,7 +635,6 @@ def main(argsl=None,                   # type: List[str]
                 find_default_container,
                 default_container=runtimeContext.default_container,
                 use_biocontainers=args.beta_use_biocontainers)
-            runtimeContext.make_fs_access = getdefault(runtimeContext.make_fs_access, StdFsAccess)
             (out, status) = executor(tool,
                                      initialized_job_order_object,
                                      runtimeContext,

cwltool/process.py

@@ -41,7 +41,7 @@
 from .software_requirements import (  # pylint: disable=unused-import
     DependenciesConfiguration)
 from .stdfsaccess import StdFsAccess
-from .utils import (DEFAULT_TMP_PREFIX, add_sizes, aslist, cmp_like_py2,
+from .utils import (DEFAULT_TMP_PREFIX, aslist, cmp_like_py2,
                     copytree_with_merge, onWindows)
 from .validate_js import validate_js_expressions
 from .context import LoadingContext, RuntimeContext, getdefault
@@ -363,9 +363,20 @@ def cleanIntermediate(output_dirs):  # type: (Set[Text]) -> None
             _logger.debug(u"Removing intermediate output directory %s", a)
             shutil.rmtree(a, True)
 
+def add_sizes(fsaccess, obj):  # type: (StdFsAccess, Dict[Text, Any]) -> None
+    if 'location' in obj:
+        try:
+            obj["size"] = fsaccess.size(obj["location"])
+        except OSError:
+            pass
+    elif 'contents' in obj:
+        obj["size"] = len(obj['contents'])
+    else:
+        return  # best effort
 
-def fillInDefaults(inputs,  # type: List[Dict[Text, Text]]
-                   job      # Dict[Text, Union[Dict[Text, Any], Any, None]]
+def fill_in_defaults(inputs,   # type: List[Dict[Text, Text]]
+                     job,        # type: Dict[Text, Union[Dict[Text, Any], List[Any], Text, None]]
+                     fsaccess    # type: StdFsAccess
                   ):  # type: (...) -> None
     for e, inp in enumerate(inputs):
         with SourceLine(inputs, e, WorkflowException, _logger.isEnabledFor(logging.DEBUG)):
@@ -378,7 +389,7 @@ def fillInDefaults(inputs,  # type: List[Dict[Text, Text]]
                 job[fieldname] = None
             else:
                 raise WorkflowException("Missing required input parameter '%s'" % shortname(inp["id"]))
-    add_sizes(job)
+    visit_class(job, ("File",), functools.partial(add_sizes, fsaccess))
 
 
 def avroize_type(field_type, name_prefix=""):
@@ -440,7 +451,7 @@ def var_spool_cwl_detector(obj,           # type: Union[Dict, List, Text]
 
 def eval_resource(builder, resource_req):  # type: (Builder, Text) -> Any
         if expression.needs_parsing(resource_req):
-            visit_class(builder.job, ("File",), add_sizes)
+            visit_class(builder.job, ("File",), functools.partial(add_sizes, builder.fs_access))
             return builder.do_eval(resource_req)
         return resource_req
 
@@ -572,11 +583,15 @@ def __init__(self,
     def _init_job(self, joborder, runtimeContext):
         # type: (Dict[Text, Text], RuntimeContext) -> Builder
 
-        job = cast(Dict[Text, Union[Dict[Text, Any], List,
-                                    Text]], copy.deepcopy(joborder))
+        job = cast(Dict[Text, Union[Dict[Text, Any], List[Any], Text, None]],
+                   copy.deepcopy(joborder))
+
+        make_fs_access = getdefault(runtimeContext.make_fs_access, StdFsAccess)
+        fs_access = make_fs_access(runtimeContext.basedir)
+
         # Validate job order
         try:
-            fillInDefaults(self.tool[u"inputs"], job)
+            fill_in_defaults(self.tool[u"inputs"], job, fs_access)
             normalizeFilesDirs(job)
             validate.validate_ex(self.names.get_name("input_record_schema", ""),
                                  job, strict=False, logger=_logger_validation_warnings)
@@ -585,8 +600,6 @@ def _init_job(self, joborder, runtimeContext):
 
         files = []  # type: List[Dict[Text, Text]]
         bindings = CommentedSeq()
-        make_fs_access = getdefault(runtimeContext.make_fs_access, StdFsAccess)
-        fs_access = make_fs_access(runtimeContext.basedir)
         tmpdir = u""
         stagedir = u""
 

cwltool/stdfsaccess.py

@@ -51,6 +51,9 @@ def open(self, fn, mode):
     def exists(self, fn):  # type: (Text) -> bool
         return os.path.exists(self._abs(fn))
 
+    def size(self, fn):    # type: (Text) -> int
+        return os.stat(self._abs(fn)).st_size
+
     def isfile(self, fn):  # type: (Text) -> bool
         return os.path.isfile(self._abs(fn))
 

cwltool/utils.py

@@ -211,16 +211,6 @@ def bytes2str_in_dicts(inp  # type: Union[Dict[Text, Any], List[Any], Any]
     # simply return elements itself
     return inp
 
-def add_sizes(obj):  # type: (Dict[Text, Any]) -> None
-    if 'location' in obj:
-        try:
-            obj["size"] = os.stat(obj["location"][7:]).st_size  # strip off file://
-        except OSError:
-            pass
-    elif 'contents' in obj:
-        obj["size"] = len(obj['contents'])
-    else:
-        return  # best effort
 
 
 def visit_class(rec, cls, op):  # type: (Any, Iterable, Union[Callable[..., Any], partial[Any]]) -> None