-
Notifications
You must be signed in to change notification settings - Fork 4
/
Copy pathdbapiutil.py
73 lines (57 loc) · 2.24 KB
/
dbapiutil.py
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
import contextlib, logging, time, datetime
logger = logging.getLogger(__name__)
def param_to_sql_literal(param):
if param is None:
return 'NULL'
elif isinstance(param, str):
return "'" + param.replace("'", "''") + "'"
elif isinstance(param, int):
return str(param)
elif isinstance(param, float):
return str(param)
else:
raise Exception(type(param))
class Connection:
def __init__(self, dbapi_module, connect_args, connect_kwargs, init_statements):
self.dbapi_module = dbapi_module
self.dbapi_conn = dbapi_module.connect(*connect_args, **connect_kwargs)
if init_statements:
for statement in init_statements:
self.execute(statement, [])
def execute(self, sql, params = (), convert_datetimes = True):
cursor = self.dbapi_conn.cursor()
if params and convert_datetimes:
conv_params = [ ]
for el in params:
if isinstance(el, datetime.date) or isinstance(el, datetime.datetime):
if el.tzinfo is not None:
raise NotImplementedError
conv_params.append(el.isoformat())
else:
conv_params.append(el)
params = conv_params
params = tuple(param_to_sql_literal(p) for p in params)
sql = sql % params
t0 = time.time()
logger.debug('Starting execution of query: %s', sql)
cursor.execute(sql)
t1 = time.time()
logger.debug('Query execution took %.3f seconds', t1 - t0)
return cursor
def explain(self, sql, params = ()):
return self.execute('explain ' + sql, params)
def close(self):
return self.dbapi_conn.close()
def commit(self):
return self.dbapi_conn.commit()
def rollback(self):
return self.dbapi_conn.rollback()
@contextlib.contextmanager
def connect(dbapi_module, connect_args=(), connect_kwargs={}, init_statements=()):
conn = None
try:
conn = Connection(dbapi_module, connect_args, connect_kwargs, init_statements)
yield conn
finally:
if conn is not None:
conn.close()