Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def rm(self, path):
"""Removes folder or job in the path.
To clean all use /* as a path.
"""
try:
path = str(path)
pp = self.__get_path(path)
if len(pp) > 1:
dpath = os.path.join(*pp[:-1])
f = self.__select_dir(dpath)
if pp[-1] != '*':
try:
self.__remove_dir(path=dpath, dir=pp[-1])
except KeyError:
raise TreeError(1, "%s does not exist" % pp[-1])
else:
for k in list(self.__get_folders()):
del self.__get_folders()[k]
else:
raise TreeError(3, "Can not delete the root directory")
self._setDirty()
finally:
self._releaseSessionLockAndFlush()
returnable_folder = folders
top_level = ''
## loop through elements in the path to get to the requested path from the user
for _dir in path:
## catch thie exception
if _dir not in returnable_folder:
clean_path = os.path.join(*path)
raise TreeError(1, "Directory %s does not exist in folder %s, accessing: %s" % (_dir, top_level, clean_path))
## 'cd' into the folder of interest
returnable_folder = returnable_folder[_dir]
if not isType(returnable_folder, type({})):
clean_path = os.path.join(*path)
raise TreeError(2, "%s not a directory, accessing: %s" % (_dir, clean_path))
top_level = _dir
return returnable_folder
"""
try:
job = stripProxy(job)
mydir = self.__select_dir(path)
if isType(job, Job):
mydir[job.getFQID('.')] = job.getFQID('.') # job.id
elif isType(job, JobRegistrySlice):
for sliceKey in job.objects.keys():
mydir[sliceKey] = sliceKey
elif isType(job, list):
for element in job:
mydir[element.id] = element.id
else:
raise TreeError(4, "Not a job/slice/list object")
self._setDirty()
except Exception as err:
logger.error("Error: %s" % err)
raise
finally:
self._releaseSessionLockAndFlush()
def __make_dir(self, path):
_path = self.__get_path(path)
returnable_folder = self.__folder_cd(_path[:-1])
local_dir = _path[-1]
if local_dir not in returnable_folder:
returnable_folder[local_dir] = {}
if not isType(returnable_folder[local_dir], type({})):
clean_path = os.path.join(*_path)
raise TreeError(2, "%s not a directory, accessing: %s" % (local_dir, clean_path))
return returnable_folder[local_dir]
def __folder_cd(self, path):
folders = self.__get_folders()
if path == []:
return folders
returnable_folder = folders
top_level = ''
## loop through elements in the path to get to the requested path from the user
for _dir in path:
## catch thie exception
if _dir not in returnable_folder:
clean_path = os.path.join(*path)
raise TreeError(1, "Directory %s does not exist in folder %s, accessing: %s" % (_dir, top_level, clean_path))
## 'cd' into the folder of interest
returnable_folder = returnable_folder[_dir]
if not isType(returnable_folder, type({})):
clean_path = os.path.join(*path)
raise TreeError(2, "%s not a directory, accessing: %s" % (_dir, clean_path))
top_level = _dir
return returnable_folder