Spaces:
Runtime error
Runtime error
File size: 2,559 Bytes
022601f 016285f |
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 |
import sqlite3
from sqlite3 import Connection
import datetime
class ResponseDb:
DB_PATH = "response.db"
def __init__(self):
# establish connection
self.con = self.get_connection()
cur = self.con.cursor()
# create table if it doesnt already exist
table_string = "CREATE TABLE IF NOT EXISTS responses (dialogue_id text, task_id text, turn integer, question text, response text, datetime date)"
cur.execute(table_string)
self.con.commit()
cur.close()
def get_connection(self):
"""Put the connection in cache to reuse if path does not change between Streamlit reruns.
NB : https://stackoverflow.com/questions/48218065/programmingerror-sqlite-objects-created-in-a-thread-can-only-be-used-in-that-sa
"""
return sqlite3.connect(self.DB_PATH, check_same_thread=False)
#def __del__(self):
#self.con.close()
def add(self, dialogue_id, task_id, turn, question, response):
cur = self.con.cursor()
curr_datetime = datetime.datetime.now()
cur.execute(
"insert into responses values (?, ?, ?, ?, ?, ?)",
(dialogue_id, int(task_id), turn, question, response, curr_datetime),
)
self.con.commit()
cur.close()
def get_id(self, dialogue_id):
cur = self.con.cursor()
cur.execute(
"select * from responses where conv=:id",
{"id": dialogue_id},
)
results = cur.fetchall()
cur.close()
return results
def get_id_turn(self, dialogue_id, turn):
cur = self.con.cursor()
cur.execute(
"select * from responses where conv=:id and turn=:turn",
{
"id": dialogue_id,
"turn": turn,
},
)
results = cur.fetchall()
cur.close()
return results
def get_all(self):
cur = self.con.cursor()
cur.execute(
"select * from responses",
)
results = cur.fetchall()
cur.close()
return results
class StResponseDb(ResponseDb):
def get_connection(self):
"""Put the connection in cache to reuse if path does not change between Streamlit reruns.
NB : https://stackoverflow.com/questions/48218065/programmingerror-sqlite-objects-created-in-a-thread-can-only-be-used-in-that-sa
"""
return sqlite3.connect(self.DB_PATH, check_same_thread=False)
if __name__ == "__main__":
db = ResponseDb()
print(db.get_all()) |