sdk/cwl/arvados_cwl/fsaccess.py

   1 import fnmatch
   2 import os
   3 import errno
   4 import urlparse
   5
   6 import cwltool.stdfsaccess
   7 from cwltool.pathmapper import abspath
   8
   9 import arvados.util
  10 import arvados.collection
  11 import arvados.arvfile
  12
  13 from schema_salad.ref_resolver import DefaultFetcher
  14
  15 class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess):
  16     """Implement the cwltool FsAccess interface for Arvados Collections."""
  17
  18     def __init__(self, basedir, api_client=None, keep_client=None):
  19         super(CollectionFsAccess, self).__init__(basedir)
  20         self.api_client = api_client
  21         self.keep_client = keep_client
  22         self.collections = {}
  23
  24     def get_collection(self, path):
  25         p = path.split("/")
  26         if p[0].startswith("keep:") and arvados.util.keep_locator_pattern.match(p[0][5:]):
  27             pdh = p[0][5:]
  28             if pdh not in self.collections:
  29                 self.collections[pdh] = arvados.collection.CollectionReader(pdh, api_client=self.api_client,
  30                                                                             keep_client=self.keep_client)
  31             return (self.collections[pdh], "/".join(p[1:]))
  32         else:
  33             return (None, path)
  34
  35     def _match(self, collection, patternsegments, parent):
  36         if not patternsegments:
  37             return []
  38
  39         if not isinstance(collection, arvados.collection.RichCollectionBase):
  40             return []
  41
  42         ret = []
  43         # iterate over the files and subcollections in 'collection'
  44         for filename in collection:
  45             if patternsegments[0] == '.':
  46                 # Pattern contains something like "./foo" so just shift
  47                 # past the "./"
  48                 ret.extend(self._match(collection, patternsegments[1:], parent))
  49             elif fnmatch.fnmatch(filename, patternsegments[0]):
  50                 cur = os.path.join(parent, filename)
  51                 if len(patternsegments) == 1:
  52                     ret.append(cur)
  53                 else:
  54                     ret.extend(self._match(collection[filename], patternsegments[1:], cur))
  55         return ret
  56
  57     def glob(self, pattern):
  58         collection, rest = self.get_collection(pattern)
  59         if collection and not rest:
  60             return [pattern]
  61         patternsegments = rest.split("/")
  62         return self._match(collection, patternsegments, "keep:" + collection.manifest_locator())
  63
  64     def open(self, fn, mode):
  65         collection, rest = self.get_collection(fn)
  66         if collection:
  67             return collection.open(rest, mode)
  68         else:
  69             return super(CollectionFsAccess, self).open(self._abs(fn), mode)
  70
  71     def exists(self, fn):
  72         collection, rest = self.get_collection(fn)
  73         if collection:
  74             return collection.exists(rest)
  75         else:
  76             return super(CollectionFsAccess, self).exists(fn)
  77
  78     def isfile(self, fn):  # type: (unicode) -> bool
  79         collection, rest = self.get_collection(fn)
  80         if collection:
  81             if rest:
  82                 return isinstance(collection.find(rest), arvados.arvfile.ArvadosFile)
  83             else:
  84                 return False
  85         else:
  86             return super(CollectionFsAccess, self).isfile(fn)
  87
  88     def isdir(self, fn):  # type: (unicode) -> bool
  89         collection, rest = self.get_collection(fn)
  90         if collection:
  91             if rest:
  92                 return isinstance(collection.find(rest), arvados.collection.RichCollectionBase)
  93             else:
  94                 return True
  95         else:
  96             return super(CollectionFsAccess, self).isdir(fn)
  97
  98     def listdir(self, fn):  # type: (unicode) -> List[unicode]
  99         collection, rest = self.get_collection(fn)
 100         if collection:
 101             if rest:
 102                 dir = collection.find(rest)
 103             else:
 104                 dir = collection
 105             if dir is None:
 106                 raise IOError(errno.ENOENT, "Directory '%s' in '%s' not found" % (rest, collection.portable_data_hash()))
 107             if not isinstance(dir, arvados.collection.RichCollectionBase):
 108                 raise IOError(errno.ENOENT, "Path '%s' in '%s' is not a Directory" % (rest, collection.portable_data_hash()))
 109             return [abspath(l, fn) for l in dir.keys()]
 110         else:
 111             return super(CollectionFsAccess, self).listdir(fn)
 112
 113     def join(self, path, *paths): # type: (unicode, *unicode) -> unicode
 114         if paths and paths[-1].startswith("keep:") and arvados.util.keep_locator_pattern.match(paths[-1][5:]):
 115             return paths[-1]
 116         return os.path.join(path, *paths)
 117
 118     def realpath(self, path):
 119         if path.startswith("$(task.tmpdir)") or path.startswith("$(task.outdir)"):
 120             return path
 121         collection, rest = self.get_collection(path)
 122         if collection:
 123             return path
 124         else:
 125             return os.path.realpath(path)
 126
 127 class CollectionFetcher(DefaultFetcher):
 128     def __init__(self, cache, session, api_client=None, keep_client=None):
 129         super(CollectionFetcher, self).__init__(cache, session)
 130         self.fsaccess = CollectionFsAccess("", api_client=api_client, keep_client=keep_client)
 131
 132     def fetch_text(self, url):
 133         if url.startswith("keep:"):
 134             with self.fsaccess.open(url) as f:
 135                 return f.read()
 136         return super(CollectionFetcher, self).fetch_text(url)
 137
 138     def check_exists(self, url):
 139         if url.startswith("keep:"):
 140             return self.fsaccess.exists(url)
 141         return super(CollectionFetcher, self).check_exists(url)
 142
 143     def urljoin(self, base_url, url):
 144         if not url:
 145             return base_url
 146
 147         urlsp = urlparse.urlsplit(url)
 148         if urlsp.scheme:
 149             return url
 150
 151         basesp = urlparse.urlsplit(base_url)
 152         if basesp.scheme == "keep":
 153             if not basesp.path:
 154                 raise IOError(errno.EINVAL, "Invalid Keep locator", base_url)
 155
 156             baseparts = basesp.path.split("/")
 157             urlparts = urlsp.path.split("/")
 158
 159             pdh = baseparts.pop(0)
 160
 161             if not arvados.util.keep_locator_pattern.match(pdh):
 162                 raise IOError(errno.EINVAL, "Invalid Keep locator", base_url)
 163
 164             if urlsp.path.startswith("/"):
 165                 baseparts = []
 166
 167             if baseparts and urlparts:
 168                 baseparts.pop()
 169
 170             path = "/".join([pdh] + baseparts + urlparts)
 171             return urlparse.urlunsplit(("keep", "", path, "", urlsp.fragment))
 172
 173         return super(CollectionFetcher, self).urljoin(base_url, url)