-
Notifications
You must be signed in to change notification settings - Fork 7
/
my_classes.py
138 lines (107 loc) · 3.53 KB
/
my_classes.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
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
import wx
import threading
import ctypes
import sqlite3
from sqlite3 import Error
class TextPrintout(wx.Printout):
def __init__(self, text, font):
wx.Printout.__init__(self)
self.text = text
self.font = font
def OnPrintPage(self, page):
# This method is called by the printer when it is time to print a page
dc = self.GetDC()
# Set the font and draw the text
# font = wx.Font(10, wx.FONTFAMILY_DEFAULT, wx.FONTSTYLE_NORMAL, wx.FONTWEIGHT_NORMAL)
dc.SetFont(self.font)
dc.DrawText(self.text, 10, 10)
# Return True to indicate that the page was printed successfully
return True
class my_threads(threading.Thread):
def get_id(self):
# returns id of the respective thread
if hasattr(self, '_thread_id'):
return self._thread_id
for id, thread in threading._active.items():
if thread is self:
return id
def stop(self):
self._is_stopped = True
self.thread_id = self.get_id()
self.res = ctypes.pythonapi.PyThreadState_SetAsyncExc(self.thread_id, ctypes.py_object(SystemExit))
if self.res > 1:
ctypes.pythonapi.PyThreadState_SetAsyncExc(self.thread_id, 0)
def is_stopped(self):
return self._is_stopped
#creating class to manage Database.
class DB:
def __init__(self, DBFile):
self.DBFile = DBFile
self.conn = self.CreateConnection(self.DBFile)
# Creating a cursor object using the cursor() method
self.cursor = self.conn.cursor()
#create function to Connect to sqlite and create db file if not found.
def CreateConnection(self, db_file):
conn = None
try:
conn = sqlite3.connect(db_file)
except Error:
return False
return conn
# create function to create table
def CreateTable(self, TableName, columns):
table = f"""CREATE TABLE {TableName}{columns};"""
#Creating table
try:
self.cursor.execute(table)
except Error:
return False
return True
#create function to insert data to table
def InsertData(self, TableName, Data):
#inserting data
try:
self.cursor.execute(f'''INSERT INTO {TableName} VALUES {Data};''')
except sqlite3.OperationalError:
return False
# Commit your changes in the database
self.conn.commit()
#create function to get data from table
def GetData(self, TableName):
# Query for the specified table
try:
data = self.cursor.execute(f'''SELECT * FROM {TableName}''')
except sqlite3.OperationalError:
return False
return data.fetchall()
# creating function to Search for specific data
def SearchData(self, TableName, ColumnName, SearchText):
# Query for the specified Value
data = self.cursor.execute(f"SELECT * FROM {TableName} WHERE {ColumnName} LIKE '%{SearchText}%';")
return data.fetchall()
# Creating function to delete the selected item in any table.
def DeleteItem(self, TableName, ColumnName, Value):
# Query for delete item
self.cursor.execute(f" DELETE FROM {TableName} WHERE {ColumnName} = '{Value}';")
# clean table.
def CleanTable(self, TableName):
# Query for delete table.
self.cursor.execute(f" DELETE FROM {TableName};")
# Commit your changes in the database
self.conn.commit()
# Clean the Database from the deleted items.
self.cursor.execute(f'VACUUM;')
# Update the data.
def UpdateData(self, TableName, ColumName, OldValue, NewValue):
# Query for updating data.
sql = f"""UPDATE {TableName}
SET {ColumName} = '{NewValue}'
WHERE {ColumName} = '{OldValue}';"""
self.cursor.execute(sql)
# Commit your changes in the database
self.conn.commit()
self.cursor.execute(f'VACUUM;')
# Closing the connection
def CloseConnection(self):
self.conn.close()
#End of class