-
Notifications
You must be signed in to change notification settings - Fork 0
/
SQLAccess.py
49 lines (40 loc) · 1.49 KB
/
SQLAccess.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
import pyodbc as pyodbc
class SQLAccess:
def __init__(self):
self.server = 'dubhacks23.database.windows.net'
self.database = 'DubHacks2023'
""" removed username and password for privacy """
self.connectString = 'Driver={ODBC Driver 18 for SQL Server};SERVER=' + self.server + ';DATABASE=' + self.database + ';UID=' + self.username + ';PWD=' + self.password
def execute_non_query(self, query):
cnxn = pyodbc.connect(self.connectString)
cursor = cnxn.cursor()
cursor.execute(query)
cursor.close()
cnxn.commit()
cnxn.close()
def execute_non_query_many(self, query, data):
cnxn = pyodbc.connect(self.connectString)
cursor = cnxn.cursor()
cursor.fast_executemany = True
cursor.executemany(query, data)
cursor.close()
cnxn.commit()
cnxn.close()
def execute_non_query_select(self, query):
cnxn = pyodbc.connect(self.connectString)
cursor = cnxn.cursor()
cursor.execute(query)
rows = []
for row in cursor:
rows.append([elem for elem in row])
cursor.close()
cnxn.commit()
cnxn.close()
return rows
def fetch_all_records(self, table_name):
query = f"SELECT * FROM {table_name}"
return self.execute_non_query_select(query)
def clear_dataset():
db = SQLAccess()
query = "DELETE FROM FaceEmbeddingEntryTable"
db.execute_non_query(query)