labkeyInterface/labkeyFileBrowser.py

import urllib3
import xml.etree.ElementTree as ET
import shutil
import io

class labkeyFileBrowser:

    def __init__(self,net):
        self.net=net

    def GetRootUrl(self):
        return self.net.GetLabkeyUrl()+"/_webdav"
   
    def entryExists(self,url):
        #check if entry (file/directory) exists
        r=self.net.head(url)
        if r.status==200:
            return True
        return False

    def listRemoteDir(self,url):
        #input is remoteDir, result are remoteDirs
        dirs=[]
        #r=MethodRequest(dirUrl,method="PROPFIND")
        PROPFIND=u"""<?xml version="1.0" encoding="utf-8"?>\n
                    <propfind xmlns="DAV:">\n
                    <prop>\n
                    <getetag/>\n
                    </prop>\n
                    </propfind>"""
        headers=urllib3.util.make_headers(basic_auth=self.net.getBasicAuth())
        headers["Content-Type"]='text/xml; charset="utf-8"'
        headers['content-length']=str(len(PROPFIND))
        headers['Depth']='1'
        #add csrf
        headers["X-LABKEY-CSRF"]=self.net.getCSRF()

        try:
            f=self.net.http.request('PROPFIND',url,headers=headers,body=PROPFIND)
        #f contains json as a return value
        except urllib3.exceptions.HTTPError as e:
            print(e)
            return False,dirs
        
        try:
            tree=ET.XML(f.data)
        except ET.ParseError:
            #print(f.data.decode('utf-8'))
            #if directory is not there, a 404 response will be made by HTTP, which is not an xml file
            #we are safe to assume the directory is not there
            return False,dirs

        rps=tree.findall('{DAV:}response')
        for r in rps:
            hr=r.find('{DAV:}href')
            dirent=hr.text
            #dirent=re.sub('/labkey/_webdav/','',dirent)
            dirent=self.net.connectionConfig['host']+dirent
            dirs.append(dirent)
        del dirs[0]
        return True,dirs

    def buildPathURL(self,project,pathList):
        #build/check remote directory structure
        remoteDir=self.formatPathURL(project,'')
        for p in pathList:
            remoteDir+='/'+p
            if not self.entryExists(remoteDir):
                 self.mkdir(remoteDir)

        return remoteDir


    def readFileToBuffer(self, url):
        response=self.net.get(url,binary=True)
        buffer=io.BytesIO()
        chunk=65532
        for chunk in response.stream(1024):
            buffer.write(chunk)
        response.release_conn()
        buffer.seek(0,0)
        return buffer

    def readFileToFile(self,url,path):
        response=self.net.get(url,binary=True)
        with open(path,'wb') as out_file:
            shutil.copyfileobj(response,out_file)
        response.release_conn()

    def writeFileToFile(self,path,url):
        with open(path,'rb') as f:
            data=f.read()
        self.net.put(url,data)

    
    def formatPathURL(self,project,path):
        url=self.GetRootUrl()+'/'+project+'/@files'
        if len(path)==0:
            return url
        return url+'/'+path

    def mkdir(self,url):
        self.net.mkcol(url)


#listing helping functions
    def baseDir(self,remotePath):
        #strip potential trailing slash
        if remotePath[-1]=='/':
            remotePath=remotePath[:-1]
        #find last slash
        if remotePath.rfind('/')>-1:
            remotePath=remotePath[remotePath.rfind('/')+1:]
        return remotePath