-
Notifications
You must be signed in to change notification settings - Fork 0
/
database.py
194 lines (182 loc) · 8.2 KB
/
database.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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
import sqlite3
import csv
from datetime import datetime
class Database:
"""This class contains image read and write functions along with population of database,
updation of tables and insert queries for each use case.
All SQLite queries are being run in this python module.
"""
def __init__(self) -> None:
"""Connection to database, along with context managers (WAL: Write-Ahead Logging) to prevent database lockout.
Creation of tables are being executed within the exception handling test case.
"""
self.connection = sqlite3.connect("Library.db")
self.connection.execute("pragma journal_mode=wal")
self.cursor = self.connection.cursor()
with self.connection:
try:
self.cursor.execute(
"CREATE TABLE IF NOT EXISTS Book_Info (ID PRIMARY KEY, Genre, Title, Author, LoanPeriod, PurchaseDate, CurrentLoanStatus)"
)
self.cursor.execute(
"CREATE TABLE IF NOT EXISTS Loan_History (TransactionID INTEGER PRIMARY KEY AUTOINCREMENT, BookID, CheckoutDate DATETIME default CURRENT_DATE, ReturnDate DATE DEFAULT NULL)"
)
self.cursor.execute(
"CREATE TABLE IF NOT EXISTS Overdue_Books (BookID PRIMARY KEY, CurrentLoanStatus, CheckoutDate DATETIME default CURRENT_DATE, ReturnDate, Fines REAL, OverdueDays DATETIME)"
)
self.cursor.execute(
"CREATE TABLE IF NOT EXISTS Image_Data (Photo BLOB NOT NULL)"
)
self.connection.commit()
except Exception as e:
print(e)
finally:
self.populate_DB()
def convert_Image(self, photo) -> bytes:
"""
This function will convert digital data to binary format.
@param photo: Will be used for the image name.
"""
with open(photo, "rb") as books_img:
img_data = books_img.read()
books_img.close()
return img_data
def insert_Image(self, book) -> None:
"""
Connecting to the database where image is being read as bytes and then inserted through the form of a tuple.
@param book: This parameter will take the image and convert it to a tuple.
"""
try:
self.connection = sqlite3.connect("Library.db")
self.connection.execute("pragma journal_mode=wal")
self.cursor = self.connection.cursor()
print("Connected to SQLite")
bookImg = self.convert_Image(book)
data_tuple = bookImg
self.cursor.execute(
"INSERT INTO Image_Data (Photo) VALUES (?)", [data_tuple]
)
self.connection.commit()
print("Image uploaded")
except sqlite3.Error as e:
print(e)
finally:
if self.connection:
self.connection.close()
print("FINISHED.")
def write_Img(self, data, filename) -> None:
"""
This function will convert binary data to proper format.
@param data: Will be used as a buffer for writing the image onto the hard drive.
@param filename: Will be used as the image name.
"""
with open(filename, "wb") as file:
file.write(data)
def write_Img_Data(self) -> None:
"""
Retrieve the image as BLOB data type and write onto the hard drive.
"""
try:
global r_data
r_data = ""
self.connection = sqlite3.connect("Library.db")
self.connection.execute("pragma journal_mode=wal")
self.cursor = self.connection.cursor()
print("Connected to SQLite")
select_books = "SELECT * FROM Image_Data"
data = self.cursor.execute(select_books)
for x in data:
r_data = x[0]
photo_path = "book.jpg"
photo_path1 = "book return.jpg"
self.write_Img(r_data, photo_path)
self.write_Img(r_data, photo_path1)
self.connection.commit()
except sqlite3.Error as e:
print(e)
finally:
if self.connection:
self.connection.close()
print("sqlite connection is closed")
def populate_DB(self) -> None:
"""This function would read the text files and insert into the respective tables."""
try:
with open("Book_Info.txt") as file_open:
for row in csv.reader(file_open, delimiter=","):
self.cursor.execute(
"INSERT or IGNORE INTO Book_Info VALUES (:ID, :Genre, :Title, :Author, :LoanPeriod, :PurchaseDate, :CurrentLoanStatus)",
row,
)
file_open.close()
with open("Loan_History.txt") as file_open1:
for row in csv.reader(file_open1, delimiter=","):
self.cursor.execute(
"INSERT or IGNORE INTO Loan_History VALUES (:TransactionID, :BookID, :CheckoutDate,:ReturnDate)",
row,
)
file_open1.close()
except Exception as e:
print(e)
def book_MemberID_Change(self, member_id_entry: str, book_id_entry: str) -> str:
"""
This function would work during the checkout phase which would update the Book_Info table then insert to the Loan_History and Overdue_Books.
@param member_id_entry: Will take the memberID input.
@param book_id_entry: Will take the bookID input.
"""
with sqlite3.connect("Library.db", isolation_level=None) as connection:
cursor = connection.cursor()
connection.execute("pragma journal_mode=wal")
result = cursor.execute(
"UPDATE Book_Info SET CurrentLoanStatus = ? WHERE ID = ?",
(
member_id_entry,
book_id_entry,
),
)
result1 = cursor.execute(
"INSERT INTO Loan_History (BookID) VALUES (?)", (book_id_entry,)
)
result2 = cursor.execute(
"INSERT INTO Overdue_Books (BookID, CurrentLoanStatus) VALUES (?, ?)",
(
book_id_entry,
member_id_entry,
),
)
connection.commit()
return result.fetchall(), result1.fetchall(), result2.fetchall()
def book_Return(self, book_id_entry: str) -> None:
"""
This function would update the Loan_History table which would then go onto the Overdue_Books update,
where the overdue days and fines are being calculated through SQL query using JULIANDAY conversion.
@param book_id_entry: Will take the bookID input.
"""
with sqlite3.connect("Library.db", isolation_level=None) as connection:
return_date = datetime.now().strftime("%Y-%m-%d")
fine_amount = 0.25
cursor = connection.cursor()
connection.execute("pragma journal_mode=wal")
result = cursor.execute(
"UPDATE Loan_History SET ReturnDate = ? WHERE BookID = ? AND ReturnDate IS NULL",
(
return_date,
book_id_entry,
),
)
# Calculation of fines and overdue days through conversion of data types within the SQL query.
result1 = cursor.execute(
"UPDATE Overdue_Books SET ReturnDate = ? ,OverdueDays = CAST(JULIANDAY(?) - (SELECT JULIANDAY(t.CheckoutDate) FROM Overdue_Books t WHERE t.BookID = Overdue_Books.BookID) AS INTEGER) ,Fines = ? * CAST(JULIANDAY(?) - (SELECT JULIANDAY(t.CheckoutDate) FROM Overdue_Books t WHERE t.BookID = Overdue_Books.BookID) AS INTGER) WHERE BookID = ?",
(
return_date,
return_date,
fine_amount,
return_date,
book_id_entry,
),
)
connection.commit()
return result.fetchall(), result1.fetchall()
if __name__ == "__main__":
DB = Database()
DB.write_Img_Data()
DB.write_Img_Data()