X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/a4c37c079728f2013f5a299acc97f1b24c1e2884..9f0854e394baf712f1fbcf3ebd21af215276c0fb:/sdk/cwl/arvados_cwl/fsaccess.py diff --git a/sdk/cwl/arvados_cwl/fsaccess.py b/sdk/cwl/arvados_cwl/fsaccess.py index 0970e72d71..9d3fe1c1ba 100644 --- a/sdk/cwl/arvados_cwl/fsaccess.py +++ b/sdk/cwl/arvados_cwl/fsaccess.py @@ -1,5 +1,7 @@ import fnmatch import os +import errno +import urlparse import cwltool.stdfsaccess from cwltool.pathmapper import abspath @@ -8,12 +10,15 @@ import arvados.util import arvados.collection import arvados.arvfile +from schema_salad.ref_resolver import DefaultFetcher + class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess): """Implement the cwltool FsAccess interface for Arvados Collections.""" - def __init__(self, basedir, api_client=None): + def __init__(self, basedir, api_client=None, keep_client=None): super(CollectionFsAccess, self).__init__(basedir) self.api_client = api_client + self.keep_client = keep_client self.collections = {} def get_collection(self, path): @@ -21,7 +26,8 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess): if p[0].startswith("keep:") and arvados.util.keep_locator_pattern.match(p[0][5:]): pdh = p[0][5:] if pdh not in self.collections: - self.collections[pdh] = arvados.collection.CollectionReader(pdh, api_client=self.api_client) + self.collections[pdh] = arvados.collection.CollectionReader(pdh, api_client=self.api_client, + keep_client=self.keep_client) return (self.collections[pdh], "/".join(p[1:])) else: return (None, path) @@ -83,7 +89,7 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess): collection, rest = self.get_collection(fn) if collection: if rest: - return isinstance(collection.find(rest), arvados.collection.Collection) + return isinstance(collection.find(rest), arvados.collection.RichCollectionBase) else: return True else: @@ -96,6 +102,10 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess): dir = collection.find(rest) else: dir = collection + if dir is None: + raise IOError(errno.ENOENT, "Directory '%s' in '%s' not found" % (rest, collection.portable_data_hash())) + if not isinstance(dir, arvados.collection.RichCollectionBase): + raise IOError(errno.ENOENT, "Path '%s' in '%s' is not a Directory" % (rest, collection.portable_data_hash())) return [abspath(l, fn) for l in dir.keys()] else: return super(CollectionFsAccess, self).listdir(fn) @@ -113,3 +123,51 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess): return path else: return os.path.realpath(path) + +class CollectionFetcher(DefaultFetcher): + def __init__(self, cache, session, api_client=None, keep_client=None): + super(CollectionFetcher, self).__init__(cache, session) + self.fsaccess = CollectionFsAccess("", api_client=api_client, keep_client=keep_client) + + def fetch_text(self, url): + if url.startswith("keep:"): + with self.fsaccess.open(url) as f: + return f.read() + return super(CollectionFetcher, self).fetch_text(url) + + def check_exists(self, url): + if url.startswith("keep:"): + return self.fsaccess.exists(url) + return super(CollectionFetcher, self).check_exists(url) + + def urljoin(self, base_url, url): + if not url: + return base_url + + urlsp = urlparse.urlsplit(url) + if urlsp.scheme: + return url + + basesp = urlparse.urlsplit(base_url) + if basesp.scheme == "keep": + if not basesp.path: + raise IOError(errno.EINVAL, "Invalid Keep locator", base_url) + + baseparts = basesp.path.split("/") + urlparts = urlsp.path.split("/") + + pdh = baseparts.pop(0) + + if not arvados.util.keep_locator_pattern.match(pdh): + raise IOError(errno.EINVAL, "Invalid Keep locator", base_url) + + if urlsp.path.startswith("/"): + baseparts = [] + + if baseparts and urlparts: + baseparts.pop() + + path = "/".join([pdh] + baseparts + urlparts) + return urlparse.urlunsplit(("keep", "", path, "", urlsp.fragment)) + + return super(CollectionFetcher, self).urljoin(base_url, url)