if isinstance(src, basestring) and ArvPathMapper.pdh_dirpath.match(src):
self._pathmap[src] = MapperEnt(src, self.collection_pattern % urllib.unquote(src[5:]), srcobj["class"], True)
+ debug = logger.isEnabledFor(logging.DEBUG)
+
if src not in self._pathmap:
if src.startswith("file:"):
# Local FS ref, may need to be uploaded or may be on keep
fnPattern="keep:%s/%s",
dirPattern="keep:%s/%s",
raiseOSError=True)
- with SourceLine(srcobj, "location", WorkflowException):
+ with SourceLine(srcobj, "location", WorkflowException, debug):
if isinstance(st, arvados.commands.run.UploadFile):
uploadfiles.add((src, ab, st))
elif isinstance(st, arvados.commands.run.ArvFile):
else:
self._pathmap[src] = MapperEnt(src, src, srcobj["class"], True)
- with SourceLine(srcobj, "secondaryFiles", WorkflowException):
+ with SourceLine(srcobj, "secondaryFiles", WorkflowException, debug):
for l in srcobj.get("secondaryFiles", []):
self.visit(l, uploadfiles)
- with SourceLine(srcobj, "listing", WorkflowException):
+ with SourceLine(srcobj, "listing", WorkflowException, debug):
for l in srcobj.get("listing", []):
self.visit(l, uploadfiles)
- def addentry(self, obj, c, path, subdirs):
+ def addentry(self, obj, c, path, remap):
if obj["location"] in self._pathmap:
src, srcpath = self.arvrunner.fs_access.get_collection(self._pathmap[obj["location"]].resolved)
if srcpath == "":
srcpath = "."
c.copy(srcpath, path + "/" + obj["basename"], source_collection=src, overwrite=True)
+ remap.append((obj["location"], path + "/" + obj["basename"]))
for l in obj.get("secondaryFiles", []):
- self.addentry(l, c, path, subdirs)
+ self.addentry(l, c, path, remap)
elif obj["class"] == "Directory":
for l in obj.get("listing", []):
- self.addentry(l, c, path + "/" + obj["basename"], subdirs)
- subdirs.append((obj["location"], path + "/" + obj["basename"]))
+ self.addentry(l, c, path + "/" + obj["basename"], remap)
+ remap.append((obj["location"], path + "/" + obj["basename"]))
elif obj["location"].startswith("_:") and "contents" in obj:
with c.open(path + "/" + obj["basename"], "w") as f:
f.write(obj["contents"].encode("utf-8"))
self._pathmap[loc] = MapperEnt(urllib.quote(fn, "/:+@"), self.collection_pattern % fn[5:], cls, True)
for srcobj in referenced_files:
- subdirs = []
+ remap = []
if srcobj["class"] == "Directory" and srcobj["location"] not in self._pathmap:
c = arvados.collection.Collection(api_client=self.arvrunner.api,
keep_client=self.arvrunner.keep_client,
num_retries=self.arvrunner.num_retries)
for l in srcobj.get("listing", []):
- self.addentry(l, c, ".", subdirs)
+ self.addentry(l, c, ".", remap)
check = self.arvrunner.api.collections().list(filters=[["portable_data_hash", "=", c.portable_data_hash()]], limit=1).execute(num_retries=self.arvrunner.num_retries)
if not check["items"]:
c = arvados.collection.Collection(api_client=self.arvrunner.api,
keep_client=self.arvrunner.keep_client,
num_retries=self.arvrunner.num_retries )
- self.addentry(srcobj, c, ".", subdirs)
+ self.addentry(srcobj, c, ".", remap)
check = self.arvrunner.api.collections().list(filters=[["portable_data_hash", "=", c.portable_data_hash()]], limit=1).execute(num_retries=self.arvrunner.num_retries)
if not check["items"]:
ab = self.collection_pattern % c.portable_data_hash()
self._pathmap["_:" + unicode(uuid.uuid4())] = MapperEnt("keep:"+c.portable_data_hash(), ab, "Directory", True)
- if subdirs:
- for loc, sub in subdirs:
- # subdirs will all start with "./", strip it off
- ab = self.file_pattern % (c.portable_data_hash(), sub[2:])
+ if remap:
+ for loc, sub in remap:
+ # subdirs start with "./", strip it off
+ if sub.startswith("./"):
+ ab = self.file_pattern % (c.portable_data_hash(), sub[2:])
+ else:
+ ab = self.file_pattern % (c.portable_data_hash(), sub)
self._pathmap[loc] = MapperEnt("keep:%s/%s" % (c.portable_data_hash(), sub[2:]),
ab, "Directory", True)
self.keepdir = None
def reversemap(self, target):
- if target.startswith("keep:"):
+ p = super(ArvPathMapper, self).reversemap(target)
+ if p:
+ return p
+ elif target.startswith("keep:"):
return (target, target)
elif self.keepdir and target.startswith(self.keepdir):
- return (target, "keep:" + target[len(self.keepdir)+1:])
+ kp = "keep:" + target[len(self.keepdir)+1:]
+ return (kp, kp)
else:
- return super(ArvPathMapper, self).reversemap(target)
+ return None
class StagingPathMapper(PathMapper):
_follow_dirs = True
+ def __init__(self, referenced_files, basedir, stagedir, separateDirs=True):
+ self.targets = set()
+ super(StagingPathMapper, self).__init__(referenced_files, basedir, stagedir, separateDirs)
+
def visit(self, obj, stagedir, basedir, copy=False, staged=False):
# type: (Dict[unicode, Any], unicode, unicode, bool) -> None
loc = obj["location"]
tgt = os.path.join(stagedir, obj["basename"])
+ basetgt, baseext = os.path.splitext(tgt)
+ n = 1
+ if tgt in self.targets and (self.reversemap(tgt)[0] != loc):
+ while tgt in self.targets:
+ n += 1
+ tgt = "%s_%i%s" % (basetgt, n, baseext)
+ self.targets.add(tgt)
if obj["class"] == "Directory":
- self._pathmap[loc] = MapperEnt(loc, tgt, "Directory", staged)
+ if obj.get("writable"):
+ self._pathmap[loc] = MapperEnt(loc, tgt, "WritableDirectory", staged)
+ else:
+ self._pathmap[loc] = MapperEnt(loc, tgt, "Directory", staged)
if loc.startswith("_:") or self._follow_dirs:
self.visitlisting(obj.get("listing", []), tgt, basedir)
elif obj["class"] == "File":
if "contents" in obj and loc.startswith("_:"):
self._pathmap[loc] = MapperEnt(obj["contents"], tgt, "CreateFile", staged)
else:
- if copy:
+ if copy or obj.get("writable"):
self._pathmap[loc] = MapperEnt(loc, tgt, "WritableFile", staged)
else:
self._pathmap[loc] = MapperEnt(loc, tgt, "File", staged)