#dynamicly create callable sql functions
#put sql in sql directory
#make sure all parameters are in the long %(name)s form
#weep silently
import importlib
import os
import re
import sys
debug = 0
class q:
'''sql query and query parameter manager
execute database query with q.<name>(cursor, args=value...)
sql query files found at location "sql/<name>.sql" are automaticly loaded as the function q.<name>(cursor, arg=value...)
all parameters found in the form %(param_name)s are enforced as named args in the function call
unnamed parameters in the form %s are not handled by this interface
"sql/<name>.py" can be used where the query needs to be generated dynamicly.
There must be a <name>(dbc, ...) function to be called.
To reload a query: delete <name>
del sqlq.get_employee_records
'''
#note using _ prefix for all parameters to keep them out of the loadable namespace
def __init__(self):
self._sql = {}
self._func = {}
self._param = {}
self._match_param_exp = re.compile(r'%\((?P<id>[a-zA-Z_][a-zA-Z0-9_]+)\)s')
def __getattr__(self, name):
debug and self._log('loading function:', name)
return self._load(name)
def _load(self, query_name):
query_path = os.path.join('sql', '{}.sql'.format(query_name) )
mod_path = os.path.join('sql', '{}.py'.format(query_name) )
mod_name = '.'.join(['sql', '{}'.format(query_name)] )
if os.path.exists(query_path):
with open(query_path, 'r') as query_file:
self._sql[query_name] = query_file.read()
query_func = self._make_func(query_name)
elif os.path.exists(mod_path):
importlib.import_module(mod_name)
importlib.reload(sys.modules[mod_name])
query_func = sys.modules[mod_name].__getattribute__(query_name)
else:
raise ValueError('unable to find "{}" in sql'.format(query_name))
self.__setattr__(query_name, query_func)
return query_func
def _make_func(self, query_name):
def make_checked_func(check_set):
def checked_func(dbc, **parameter):
parameter_keys = set(parameter.keys())
if parameter_keys == check_set:
debug and self._log('exec:', query_name, parameter)
debug and self._log(dbc.mogrify(self._sql[query_name], parameter))
return self._execute(dbc, self._sql[query_name], parameter)
else:
raise ValueError('incorrect args for sql "{}": missing {}: provided {}: required {}'.format(
query_name,
check_set.difference(parameter),
parameter,
check_set))
debug and self._log(query_name, 'enforced parameters', check_set)
return checked_func
query = self._sql[query_name]
param_set = set()
for param_match in self._match_param_exp.finditer(query):
param_set.add(param_match['id'])
query_func = make_checked_func(param_set)
self._func[query_name] = query_func
self._param[query_name] = param_set;
return query_func
def _execute(self, dbc, query, parameters):
return dbc.execute(query, parameters)
def _log(self, *message):
print(*message)
#q_er is a special case where the dbc has an additional er_execute parameter added that returns errors instead of raising them
class q_er(q):
def _execute(self, dbc, query, parameters):
return dbc.er_execute(query, parameters)
#q_gen is used as a generator, for r in q.select_name(dbc, id = 'bob', dept = 'manpower')
class q_gen(q):
def _execute(self, dbc, query, parameters):
dbc.execute(query, parameters)
for row in dbc:
yield row
class fake_dbc:
def mogrify(self, query, parameters):
print(query)
print(parameters)
def execute(self, query, parameters):
print('fake_dbc exec')
if __name__ == '__main__':
debug = 0
dbc = fake_dbc()
sql = q()
sql.feastday(dbc, org_like='%%')
sql.feastday(dbc, org_like = 'aesir')
sql.compare_lookup(dbc, pop = 'fertality', role = 'aesir')
sql.compare_lookup(dbc, role = 'aesir', pop = 'fertality')
del sql.feastday
sql.feastday(dbc, org_like='')
sql.col_info(dbc, db_name = 'norse', schema = 'valhala', table = 'people')