Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def upload(request, path):
path = os.path.realpath("%s/%s" % (os.path.sep, path))
hdfs = HDFS()
return hdfs.upload(path, request)
def list_status(request, path):
path = os.path.realpath("/%s/%s" % (os.path.sep, path))
hdfs = HDFS()
return hdfs.list_status(path, request)
def make_dir(request, path):
hdfs_logger.info("make_dir")
path = os.path.realpath("/%s/%s" % (os.path.sep, path))
hdfs = HDFS()
return hdfs.make_dir(path, request)
def showShare(request,path):
result = {}
try:
path = os.path.realpath("%s/%s" %(os.path.sep,path))
hdfs = HDFS()
result = hdfs.list_status_share(request,path)
except Exception,e:
hdfs_logger.error(traceback.format_exc())
result["code"] = StatusCode["SUCCESS"]
result["data"] = TableNoData
finally:
return result
def list_status_tree(request,path):
path = os.path.realpath("/%s/%s" % (os.path.sep, path))
hdfs = HDFS()
baseData = hdfs.list_status(path, request)
result = {}
try:
if baseData:
data=[{
"name":k["name"],
"isParent":"true",
} for k in baseData["data"]["totalList"]
if k["is_dir"]==1 ]
else:
data=[]
hdfs_logger.info("liststatustree:%s" %data)
result["code"] = StatusCode["SUCCESS"]
result["data"] = data
except Exception,e:
hdfs_logger.error("%s" %e)
def download(request, path):
path = os.path.realpath("%s/%s" % (os.path.sep, path))
hdfs = HDFS()
return hdfs.download(path, request)