labkeyInterface/labkeyDatabaseBrowser.py

import labkeyInterface
import json
import chardet

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

   def selectRows(self,project,schemaName, queryName,qfilter,viewName='default',columns='all'):

      debug=False
      url=self.net.GetLabkeyUrl()+'/'+project
      url+='/query-selectRows.api?schemaName='+schemaName+\
                '&query.queryName='+queryName
      if viewName!='default':
         url+='&query.viewName={}'.format(viewName)
      if columns!='all':
         url+='&query.columns={}'.format(columns)
      for f in qfilter:
         url+="&query."+f['variable']+"~"+f['oper']+"="+f['value']
      if debug:
         print("Sending {}").format(url)
      response=self.net.get(url)
      return self.toJSON(response)

   def modifyRows(self,mode, project,schemaName, queryName, rows):
      #mode can be insert/update/delete
      debug=True
      data={}
      data['schemaName']=schemaName
      data['queryName']=queryName
      data['rows']=rows
      url=self.net.GetLabkeyUrl()+'/'+project
      url+='/query-'+mode+'Rows.api?'
      response=self.net.post(url,json.dumps(data))
      return self.toJSON(response)

   def addQuery(self,project,schemaName,queryName,fields,options=None):
      
      listDomainDefinition={}
      fNames=[f['name'] for f in fields]
      
      if schemaName=='lists':

         listDomainDefinition['kind']='IntList'
         if not options:
            fields.insert(0,{'name':'Key','rangeURI':'int'})
            options={}
            options['keyName']='Key'
            if queryName.find('enum')==0:
               options['keyType']='Integer'
            else:
               options['keyType']='AutoIncrementInteger'
         listDomainDefinition['options']=options
      
      if schemaName=='study':
      
         listDomainDefinition['kind']='StudyDatasetVisit'
      
      domainDesign={}
      domainDesign['name']=queryName
      domainDesign['description']='Some description'
      domainDesign['fields']=fields
      
      listDomainDefinition['domainDesign']=domainDesign
      
      url=self.net.GetLabkeyUrl()+'/'+project
      url+='/property-createDomain.api'
      response=self.net.post(url,json.dumps(listDomainDefinition))
      return self.toJSON(response)

   def getQueryDesign(self,project,schemaName,queryName):
      url=self.net.GetLabkeyUrl()+'/'+project
      url+='/property-getDomain.api?'
      url+='schemaName={}'.format(schemaName)
      url+='&queryName={}'.format(queryName)
      response=self.net.get(url)
      return self.toJSON(response)

   def toJSON(self,response):
      data=response.data
      encoding=chardet.detect(data)["encoding"]
      #try with a set of encodings to maximize probability of success
      encodings=[encoding,'utf_8']
      for x in encodings:
         try:
            return json.loads(data.decode(x))
         except UnicodeDecodeError:
            print(f'Failed to decode with [{x}]: {data}')
      return None