1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
"""module contains performance tests for queries"""
import time
import sys
from inspect import getmembers
from inspect import isfunction
from functools import wraps
from gn3.db_utils import database_connector
def timer(func):
"""time function"""
@wraps(func)
def wrapper_time(*args, **kwargs):
"""time wrapper"""
start_time = time.perf_counter()
results = func(*args, **kwargs)
end_time = time.perf_counter()
run_time = end_time - start_time
print(f"the time taken is {run_time:.3f} seconds")
return results
return wrapper_time
def query_executor(query, fetch_all=True):
"""function to execute a query"""
conn, _ = database_connector()
with conn:
cursor = conn.cursor()
cursor.execute(query)
if fetch_all:
return cursor.fetchall()
return cursor.fetchone()
def fetch_probeset_query(dataset_name):
"""contains queries for datasets"""
query = """SELECT * from ProbeSetData
where StrainID in (4, 5, 6, 7, 8, 9, 10, 11, 12,
14, 15, 17, 18, 19, 20, 21, 22, 24, 25, 26, 28,
29, 30, 31, 35, 36, 37, 39, 98, 99, 100, 103,
487, 105, 106, 110, 115,116, 117, 118, 119,
120, 919, 147, 121, 40, 41, 124, 125, 128, 135,
129, 130, 131, 132, 134, 138, 139, 140, 141, 142,
144, 145, 148, 149, 920, 922, 2, 3, 1, 1100)
and id in (SELECT ProbeSetXRef.DataId
FROM (ProbeSet, ProbeSetXRef, ProbeSetFreeze)
WHERE ProbeSetXRef.ProbeSetFreezeId = ProbeSetFreeze.Id
and ProbeSetFreeze.Name = '{}'
and ProbeSet.Id = ProbeSetXRef.ProbeSetId)""".format(dataset_name)
return query
@timer
def perf_simple_query():
"""initial simple query test"""
query = """select * from ProbeSetData limit 1"""
_results = query_executor(query)
return {}
@timer
def perf_hc_m2_dataset():
"""test the default dataset HC_M2_0606_P"""
query = fetch_probeset_query("HC_M2_0606_P")
_results = query_executor(query)
return {}
@timer
def perf_umutaffyexon_dataset():
"""largest dataset in gn"""
query = fetch_probeset_query("UMUTAffyExon_0209_RMA")
_results = query_executor(query)
return {}
def fetch_perf_functions():
"""function to filter all functions strwith perf_"""
name_func_dict = {name: obj for name, obj in
getmembers(sys.modules[__name__], isfunction)if isfunction(
obj) and obj.__module__ == __name__ and name.startswith('perf_')}
return name_func_dict
def fetch_cmd_args():
"""function to fetch cmd args"""
cmd_args = sys.argv[1:]
name_func_dict = fetch_perf_functions()
if len(cmd_args) > 0:
callables = [func_call for name,
func_call in name_func_dict.items() if name in cmd_args]
return callables
return list(name_func_dict.values())
if __name__ == '__main__':
func_list = fetch_cmd_args()
for call_func in func_list:
call_func()
|
