Documentation : API access to VariantDB : Example Scripts

Import/Export Filter Sets

Store a set of filters to a text file for import into a different VariantDB instance. Right-click to download the script.

Code
#!/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()

Import/Export Annotation Sets

Store a set of annotations to a text file for import into a different VariantDB instance. Right-click to download the script.

Code
#!/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()

Submit Batch Queries & Retrieve results

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.

Code
#!/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()