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
|
###########################################################################
# Copyright 2019 (C) Hui Lan <hui.lan@cantab.net>
# Written permission must be obtained from the author for commercial uses.
###########################################################################
# Reference: Dusty Phillips. Python 3 Objected-oriented Programming Second Edition. Pages 326-328.
# Copyright (C) 2019 Hui Lan
import sqlite3
class Sqlite3Template:
def __init__(self, db_fname):
self.db_fname = db_fname
def connect(self, db_fname):
self.conn = sqlite3.connect(self.db_fname)
def instructions(self, query_statement):
raise NotImplementedError()
def operate(self):
self.conn.row_factory = sqlite3.Row
self.results = self.conn.execute(self.query) # self.query is to be given in the child classes
self.conn.commit()
def format_results(self):
raise NotImplementedError()
def do(self):
self.connect(self.db_fname)
self.instructions(self.query)
self.operate()
def instructions_with_parameters(self, query_statement, parameters):
self.query = query_statement
self.parameters = parameters
def do_with_parameters(self):
self.connect(self.db_fname)
self.instructions_with_parameters(self.query, self.parameters)
self.operate_with_parameters()
def operate_with_parameters(self):
self.conn.row_factory = sqlite3.Row
self.results = self.conn.execute(self.query, self.parameters) # self.query is to be given in the child classes
self.conn.commit()
class InsertQuery(Sqlite3Template):
def instructions(self, query):
self.query = query
class RecordQuery(Sqlite3Template):
def instructions(self, query):
self.query = query
def format_results(self):
output = []
for row_dict in self.results.fetchall():
lst = []
for k in dict(row_dict):
lst.append( row_dict[k] )
output.append(', '.join(lst))
return '\n\n'.join(output)
def get_results(self):
result = []
for row_dict in self.results.fetchall():
result.append( dict(row_dict) )
return result
if __name__ == '__main__':
#iq = InsertQuery('RiskDB.db')
#iq.instructions("INSERT INTO inspection Values ('FoodSupplies', 'RI2019051301', '2019-05-13', '{}')")
#iq.do()
#iq.instructions("INSERT INTO inspection Values ('CarSupplies', 'RI2019051302', '2019-05-13', '{[{\"risk_name\":\"elevator\"}]}')")
#iq.do()
rq = RecordQuery('wordfreqapp.db')
rq.instructions("SELECT * FROM article WHERE level=3")
rq.do()
#print(rq.format_results())
|
