+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
+
import fnmatch
import os
import errno
def glob(self, pattern):
collection, rest = self.get_collection(pattern)
- if collection and not rest:
+ if collection is not None and not rest:
return [pattern]
patternsegments = rest.split("/")
return self._match(collection, patternsegments, "keep:" + collection.manifest_locator())
def open(self, fn, mode):
collection, rest = self.get_collection(fn)
- if collection:
+ if collection is not None:
return collection.open(rest, mode)
else:
return super(CollectionFsAccess, self).open(self._abs(fn), mode)
def exists(self, fn):
collection, rest = self.get_collection(fn)
- if collection:
+ if collection is not None:
if rest:
return collection.exists(rest)
else:
def isfile(self, fn): # type: (unicode) -> bool
collection, rest = self.get_collection(fn)
- if collection:
+ if collection is not None:
if rest:
return isinstance(collection.find(rest), arvados.arvfile.ArvadosFile)
else:
def isdir(self, fn): # type: (unicode) -> bool
collection, rest = self.get_collection(fn)
- if collection:
+ if collection is not None:
if rest:
return isinstance(collection.find(rest), arvados.collection.RichCollectionBase)
else:
def listdir(self, fn): # type: (unicode) -> List[unicode]
collection, rest = self.get_collection(fn)
- if collection:
+ if collection is not None:
if rest:
dir = collection.find(rest)
else:
if path.startswith("$(task.tmpdir)") or path.startswith("$(task.outdir)"):
return path
collection, rest = self.get_collection(path)
- if collection:
+ if collection is not None:
return path
else:
return os.path.realpath(path)
pipeline_template_uuid_pattern = re.compile(r'[a-z0-9]{5}-p5p6p-[a-z0-9]{15}')
def collectionResolver(api_client, document_loader, uri, num_retries=4):
+ if uri.startswith("keep:") or uri.startswith("arvwf:"):
+ return uri
+
if workflow_uuid_pattern.match(uri):
return "arvwf:%s#main" % (uri)