-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathautolector_db.py
63 lines (55 loc) · 1.78 KB
/
autolector_db.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
import time
import sqlite3
# Set up sqllite3 db class.
class AutolectorDB:
'''
A neat little DB class so we know it disconnects because __del__
'''
db_file = 'autolector.db'
table_name = 'autolector'
dt_pattern = '%Y-%m-%d %H:%M:%S'
def __init__(self):
'''
Get a connection to self.db_file and create a cursor
'''
self.connection = sqlite3.connect(self.db_file, check_same_thread=False)
self.cursor = self.connection.cursor()
def save(self):
'''
Have to do this after everything apparently.
'''
self.connection.commit()
def insert(self, question, context, answer):
'''
Insert question, context, answer, and timestamp into the table.
'''
date = time.strftime(self.dt_pattern)
qry = "INSERT INTO autolector VALUES (?, ?, ?, ?)"
self.cursor.execute(qry, (date, question, context, answer))
self.save()
def get_all(self):
'''
Return all the rows in the self.table_name table.
'''
qry = f'SELECT * FROM {self.table_name}'
rows = [x for x in self.cursor.execute(qry)]
return rows
def create_table(self):
'''
Creates a table named self.table_name: default-> autolector.
'''
qry = f'''CREATE TABLE {self.table_name}
(date text, question text, context text, answer text)'''
self.cursor.execute(qry)
def drop_table(self):
'''
Drop table self.table_name from database.
'''
qry = f"DROP TABLE {self.table_name}"
self.cursor.execute(qry)
def __del__(self):
'''
The whole reason we are using OOP. Close connection automatically.
'''
self.save()
self.connection.close()