Store a set of filters to a text file for import into a different VariantDB instance. Right-click to download the script.
#!/usr/bin/python # default modules import sys import json import urllib import urllib2 import getopt import os.path # non-default packages (install with "pip install") import requests ## SET LOCAL INSTALLATIONS HERE: vdb_url = "http://DevBox/variantdb/api/" def main() : # parse commandline optlist,args = getArgs(sys.argv[1:]) # if no api key provided : exit try: apikey = optlist['k'] except: print('No API key provided.') Usage() ## if both -I and -a are provided : exit if 'f' in optlist and 'I' in optlist: Usage() # check correctness of API. answer = fetch_json_page(vdb_url + 'CheckApiKey?apiKey='+apikey) try: answer == '1' except: print(answer) print("Invalid API Key provided.") print("Log in on VariantDB and check the key under User-Settings (click on you name)") Usage() ## EXPORT if 'f' in optlist: try: optlist['f'].isdigit() except: print("Invalid (non-numeric) Filter_ID provided: '"+optlist['f']+"'") Usage(); try: answer = fetch_json_page(vdb_url+'SavedFilters/'+optlist['f']+'?apiKey='+apikey) except: print("Unknown Filter_ID provided: "+optlist['f']) Usage(); with open(optlist['O'],'w') as outfile: json.dump(answer,outfile) sys.exit(0) ## IMPORT elif 'I' in optlist: try: os.path.isfile(optlist['I']) except: print("Provided file path does not exist") Usage() ## now pass it VariantDB api. answer = requests.post(vdb_url+"LoadFilters?apiKey="+apikey,data={'apiKey':apikey},files={'json_file': open(optlist['I'], 'r')}) fid = json.loads(answer.text) print "Filter Stored under ID: ",fid['id'] ## LIST ALL else: try: answer = fetch_json_page(vdb_url+'SavedFilters?apiKey='+apikey) except: print "Failed to retrieve list of saved filters." for fset in answer: fid = fset['fid'] fname = fset['FilterName'] fcomments = fset['Comments'] fitems = fset['nrRules'] print fname+":" print "=" * (len(fname)+1) print " ID:",fid print " nrRules:",fitems print " Comments:",fcomments,"\n" def getArgs(args): ## arguments # -k : apikey (mandatory) # -f : filter_id (optional) # -I : in_file_path to import (optional) # -O : out_file_path opts, args = getopt.getopt(args, 'k:f:O:I:h') optlist = dict() for opt, arg in opts: optlist[opt[1:]] = arg if 'f' in optlist and 'O' not in optlist: print "Missing argument : -O" Usage() if 'h' in optlist: Usage() return(optlist,args) def Usage(): # print help print "\n\nUsage: python Import_Export_Filter.py -k " print " Default: lists all Filtersets stored by the provided user" print " Optional : -f : get json representation of filterset. Combine with -O." print " Optional : -O : where to write the filterset." print " Optional : -I : import a filterset from a file (exported by -f). " print " \n => -f and -I are mutually exclusive !" print "\n\n" sys.exit(0) def fetch_json_page(url): try: data = urllib2.urlopen(url) j = json.load(data) except: print('Fetching api repsonse failed for following url:') print(url) sys.exit(2) ## return data return j if __name__ == "__main__": main()
Store a set of annotations to a text file for import into a different VariantDB instance. Right-click to download the script.
#!/usr/bin/python # default modules import sys import json import urllib import urllib2 import getopt import os.path # non-default packages (install with "pip install") import requests ## SET LOCAL INSTALLATIONS HERE: vdb_url = "http://143.169.238.104/variantdb/api/" def main() : # parse commandline optlist,args = getArgs(sys.argv[1:]) # if no api key provided : exit try: apikey = optlist['k'] except: print('No API key provided.') Usage() ## if both -I and -a are provided : exit if 'a' in optlist and 'I' in optlist: Usage() # check correctness of API. answer = fetch_json_page(vdb_url + 'CheckApiKey?apiKey='+apikey) try: answer == '1' except: print(answer) print("Invalid API Key provided.") print("Log in on VariantDB and check the key under User-Settings (click on you name)") Usage() ## EXPORT if 'a' in optlist: try: optlist['a'].isdigit() except: print("Invalid (non-numeric) Annotation_ID provided: '"+optlist['a']+"'") Usage(); try: answer = fetch_json_page(vdb_url+'SavedAnnotations/'+optlist['a']+'?apiKey='+apikey) except: print("Unknown Annotation_ID provided: "+optlist['a']) Usage(); with open(optlist['O'],'w') as outfile: json.dump(answer,outfile) sys.exit(0) ## IMPORT elif 'I' in optlist: try: os.path.isfile(optlist['I']) except: print("Provided file path does not exist") Usage() ## now pass it VariantDB api. answer = requests.post(vdb_url+"LoadAnnotations?apiKey="+apikey,data={'apiKey':apikey},files={'json_file': open(optlist['I'], 'r')}) aid = json.loads(answer.text) print "Annotation Set Stored under ID: ",aid['id'] ## LIST ALL else: try: answer = fetch_json_page(vdb_url+'SavedAnnotations?apiKey='+apikey) except: print "Failed to retrieve list of saved annotations." for aset in answer: aid = aset['aid'] aname = aset['AnnotationName'] acomments = aset['Comments'] aitems = aset['nrItems'] print aname+":" print "=" * (len(aname)+1) print " ID:",aid print " nrItems:",aitems print " Comments:",acomments,"\n" def getArgs(args): ## arguments # -k : apikey (mandatory) # -a : annotation_id (optional) # -I : in_file_path to import (optional) # -O : out_file_path opts, args = getopt.getopt(args, 'k:a:O:I:h') optlist = dict() for opt, arg in opts: optlist[opt[1:]] = arg if 'a' in optlist and 'O' not in optlist: print "Missing argument : -O" Usage() if 'h' in optlist: Usage() return(optlist,args) def Usage(): # print help print "\n\nUsage: python Import_Export_Annotation.py -k " print " Default: lists all annotations sets stored by the provided user" print " Optional : -a : print json representation of annotation set. Combine with -O." print " Optional : -O : where to write the annotation set." print " Optional : -I : import an annotation set from a file (exported by -a). " print " \n => -a and -I are mutually exclusive !" print " => In absence of both -a and -I, a summary of all saved anntotations is listed to pick from."; print "\n\n" sys.exit(0) def fetch_json_page(url): try: data = urllib2.urlopen(url) j = json.load(data) except: print('Fetching api repsonse failed for following url:') print(url) sys.exit(2) ## return data return j if __name__ == "__main__": main()
Run the same provided filter set, and optional annotation set on multiple samples. Results are fetched in json format for downstream analysis. Right-click to download the script.
#!/usr/bin/python # default modules import sys import json import urllib import urllib2 import getopt import os.path import time # non-default packages (install with "pip install") import requests ## SET LOCAL INSTALLATIONS HERE: vdb_url = "http://DevBox/variantdb/api/" ## global variables apiKey = "" sid_x_sname = {} q_x_s = {} def main() : # parse commandline optlist,args = getArgs(sys.argv[1:]) apiKey = optlist['k'] print "apikey:",apiKey # check correctness of API. CheckAPIKey(vdb_url,apiKey) # check if filter_id is valid. try: optlist['f'].isdigit() except: print("Invalid (non-numeric) Filter_ID provided: '"+optlist['f']+"'") Usage(); try: answer = fetch_json_page(vdb_url+'SavedFilters/'+optlist['f']+'?apiKey='+apiKey) except: print("Unknown Filter_ID provided: "+optlist['f']) Usage(); fid = optlist['f'] # check anno_id(s) q_anno = '' if 'a' in optlist: annos = optlist['a'].split(',') for aid in annos: # digit? try: aid.isdigit() except: print("Invalid (non-numeric) Annotation_ID provided: '"+aid+"'") Usage(); # known in db? try: answer = fetch_json_page(vdb_url+'SavedAnnotations/'+aid+'?apiKey='+apiKey) except: print("Unknown Annotation_ID provided: "+aid) Usage(); q_anno = '&aid='+optlist['a'] # get list of all samples. answer = fetch_json_page(vdb_url+'Samples?apiKey='+apiKey) db_samples = {} for sample in answer: if not sample['sample_name'] in db_samples: db_samples[sample['sample_name']] = {} db_samples[sample['sample_name']][sample['id']] = sample['project_name'] # sample list to work on q_snames = list() q_sids = list() # get provided sample names if os.path.isfile(optlist['I']): with open(optlist['I'],'r') as f: for line in f: q_snames.append(line.rstrip()) f.close() else: q_snames.append(optlist['I']) # get corresponding sample_ids for sname in q_snames: if sname in db_samples: if len(db_samples[sname]) == 1: q_sids.append(db_samples[sname].keys()[0]) sid_x_sname[db_samples[sname].keys()[0]] = sname elif len(db_samples[sname]) == 0: print "\n ERROR: Sample '"+sname+"' was not found in the database.\n" sys.exit(2) else: print "\nSample name '"+sname+"' is not unique. Please pick the project to use the sample from:" idx = 0 sids = {} for i in db_samples[sname]: idx += 1 print " ",str(idx),": project name",db_samples[sname][i] sids[idx] = i choice = int(raw_input("Your Choice [1]: ")) if not (choice in sids.keys()): print "\n ERROR: Invalid selection." sys.exit(2) q_sids.append(sids[choice]) sid_x_sname[sids[choice]] = sname ## launch queries. q_ids = {} # query_id => status for sid in q_sids: answer = fetch_json_page(vdb_url+'SubmitQuery/Sample/'+sid+'?apiKey='+apiKey+'&fid='+fid+q_anno) q_ids[answer['query_key']] = 0 q_x_s[answer['query_key']] = sid ## wait for them to finish. all_done = 0 while all_done == 0: all_done = 1 for q_id in q_ids.keys(): if q_ids[q_id] == 0: answer = fetch_json_page(vdb_url+'GetStatus/Query/'+q_id+'?apiKey='+apiKey) if answer['status'] == 'finished': q_ids[q_id] = 1 # get query_results FetchQueryResults(q_id,apiKey) else: all_done = 0 if all_done == 0: answer = fetch_json_page(vdb_url+'GetStatus/Queue?apiKey='+apiKey) print "Waiting for queries to finish. Items on queue:",str(answer['nr_queued']) time.sleep(15) def getArgs(args): ## arguments # -k : apikey (mandatory) # -f : filter_id # -a : annotation_id (comma-seperated list, optional) # -O : out_file_path # -I : infile with sample-names. one per line, or one sample name. opts, args = getopt.getopt(args, 'k:f:O:I:a:h') optlist = dict() for opt, arg in opts: optlist[opt[1:]] = arg if 'k' not in optlist: print "ERROR: Missing argument : -k" Usage() if 'f' not in optlist: print "ERROR: Missing argument : -f" Usage() if 'O' not in optlist: print "ERROR: Missing argument : -O" Usage() if 'I' not in optlist: print "ERROR: Missing argument : -I" Usage() if 'h' in optlist: Usage() return(optlist,args) def Usage(): # print help print "\n\nUsage: python Run_Batch_Query.py -k -I -f -a -O " print "\n Output is written to files _ .txt. If " print "\n\n" sys.exit(0) def CheckAPIKey(url,key): # check correctness of API. answer = fetch_json_page(url + 'CheckApiKey?apiKey='+key) try: answer == '1' except: print("Invalid API Key provided.") print("Log in on VariantDB and check the key under User-Settings (click on you name)") Usage() def fetch_json_page(url): try: data = urllib2.urlopen(url) j = json.load(data) except: print('Fetching api repsonse failed for following url:') print(url) sys.exit(2) ## return data return j def FetchQueryResults(q_id,apiKey): answer = fetch_json_page(vdb_url+'GetQueryResults/'+q_id+'?apiKey='+apiKey) sname = sid_x_sname[q_x_s[q_id]] ## todo: parse json to tabular, and write to file for sample 'sname'. print "printed results for "+sname if __name__ == "__main__": main()