I feel like there must be a better way to do the following. I would really appreciate any feedback/pointers so that I can improve my code and skills (or lack thereof!).
The goal is to allow users to query our imapala database using a python script, so that they don't have to learn SQL.
I'd like to take the following dictionary of user arguments (as an example):
args = {
'members':'all',
'subject_id':'all',
'genes':'BRCA1,TRT1,SON%,HOX%',
'gt':'hom',
'chrom':'8',
'search_db':'wgs.comgen_variant',
'search_db_cols':'all',
'annot_db':'ref.ensembl_genes',
'annot_db_cols':'gene_name',
}
And parse them as follows:
If a value is 'all', no statement is added to the query
If a value ends in "%", create a statement as
AND db_name.table_name.column LIKE "value%"
If a value doesn't end in "%", create a statement as
AND db_name.table_name.column = "value"
The desired output should look like (this is just the beginning of the query):
SELECT wgs.comgen_variant.*, ref.ensembl_genes.gene_name
FROM wgs.comgen_variant,ref.ensembl_genes
WHERE (ref.ensembl_genes.gene_name = 'BRCA1'
OR ref.ensembl_genes.gene_name = 'TRT1'
OR ref.ensembl_genes.gene_name LIKE 'SON%'
OR ref.ensembl_genes.gene_name LIKE 'HOX%')
AND wgs.comgen_variant.chrom = "8"
AND wgs.comgen_variant.gt = "hom"
I came up with the following code, to separate the wildcards from the non-wildcards and create statements:
# separate wildcards from args
def find_wildcards(x):
user_args = []
wildcards = []
# separate user args by comma
arg_list = x.replace("'", "").split(',')
for arg in arg_list:
if arg.endswith('%'):
wildcards.append(arg)
else:
user_args.append(arg)
return user_args, wildcards
# function to turn arg_list and wild_list into statements
def process_args(arg_list, wild_list, name_arg, arg_db):
# if there is only one user arg
if (len(arg_list) == 1 and len(wild_list)== 0):
query_args = 'AND {0}.{1} = '.format(arg_db, name_arg) + "'" + ", ".join(str(e) for e in arg_list) + "'"
# if there are more than one user arguments
elif (len(arg_list) > 1 and len(wild_list) == 0):
conditions = ['{0}.{1} = '.format(arg_db, name_arg) + "'" + s + "'" for s in arg_list]
query_args = "AND (" + " OR ".join(map(str, conditions)) + ")"
# if there is only one wildcard
elif (len(wild_list) == 1 and len(arg_list) < 1):
query_args = 'AND {0}.{1} LIKE '.format(arg_db, name_arg) + "'" + ", ".join(str(e) for e in wild_list) + "'"
# if there are more than one wildcard
elif (len(wild_list) > 1 and len(arg_list) == 0):
conditions = ['{0}.{1} LIKE '.format(arg_db, name_arg) + "'" + s + "'" for s in wild_list]
query_args = "AND (" + " OR ".join(map(str, conditions)) + ")"
#if there are more than one wildcard and more than one user arg
elif (len(arg_list) > 1 and len(wild_list) > 1):
reg_conditions = ['{0}.{1} = '.format(arg_db, name_arg) + "'" + s + "'" for s in arg_list]
reg_args = "AND ((" + " OR ".join(map(str, reg_conditions)) + ")"
wild_conditions = ['{0}.{1} LIKE '.format(arg_db, name_arg) + "'" + s + "'" for s in wild_list]
wild_args = "AND (" + " OR ".join(map(str, wild_conditions)) + ")"
query_args = (reg_args + " " + wild_args).replace(' AND ', ' OR ', 2) + ")"
#if there is one of each
elif (len(arg_list) == 1 and len(wild_list) == 1):
reg_conditions = ['{0}.{1} = '.format(arg_db, name_arg) + "'" + s + "'" for s in arg_list]
reg_args = "AND (" + " OR ".join(map(str, reg_conditions)) + ")"
wild_conditions = ['{0}.{1} LIKE '.format(arg_db, name_arg) + "'" + s + "'" for s in wild_list]
wild_args = "AND (" + " OR ".join(map(str, wild_conditions)) + ")"
query_args = (reg_args + " " + wild_args).replace(') AND (', ' OR ', 1)
else:
query_args = ''
return query_args
# run above function on user args
def create_query(argument, arg_name, arg_db):
if argument != 'all':
arg_list, wild_list = find_wildcards(argument)
arg_query = process_args(arg_list, wild_list, arg_name, arg_db)
else:
arg_query = ''
return arg_query
Then, I run the following to create a statement for each argument and merge them together:
# create members argument
member_query = create_query(members,"sample_id", search_db)
# create subject_id argument
subject_query = create_query(subject_id,"sample_id", search_db)
# create genes argument
genes_query = create_query(genes,"gene_name", annot_db)
# create genotype argument
gt_query = create_query(gt,"gt", search_db)
# create chromosome argument
chrom_query = create_query(chrom,"chrom", search_db)
# list to store final query object
final_query = []
# merge query arguments
def merge_query(query_arg):
if len(query_arg) > 0:
final_query.append(query_arg)
# create list to apply function on
query_list = [member_query, subject_query, genes_query, gt_query,chrom_query]
# apply function to list of query statements
for query in query_list:
merge_query(query)
# create final impala query, replacing first item with WHERE instead of AND
impala_query= ' '.join(final_query).replace('AND', 'WHERE', 1)
Then I run the following to determine which columns to return, and write the beginning of the query:
# function to create list of columns to return
def get_cols(db, col_arg):
if col_arg == 'all':
col_query = '{0}.*'.format(db)
else:
col_list = col_arg.replace("'", "").split(',')
if len(col_list) > 1:
for col in col_list:
cols = ["{0}.".format(db) + s for s in col_list]
col_query = ",".join(map(str, cols))
elif len(col_list) == 1:
col_query = "{0}.".format(db) + ",".join(map(str, col_list))
else:
print "Check your search_db_cols and annot_db_cols arguments and try again please."
col_query = ''
return col_query
# merge all query statements
col_statement = get_cols(search_db, search_db_cols) + ", " + get_cols(annot_db, annot_db_cols)
# write beginning of query
query_start = "SELECT " + col_statement + " FROM {0},{1} {2}".format(search_db, annot_db, impala_query)