Library Management System — Python + SQL

# Highlights:
# - get_cursor(): returns a shared pyodbc cursor (DSN/connection string configured once)
# - email_validation(s): simple pattern/format checks
# - phone_number_validation(s): digits/length check (normalized)
# - generate_md5(pwd): return md5(pwd).hexdigest() used before INSERT/UPDATE
import pyodbc as odbc
import os
from tkinter import ttk

data_source = 'SQL-ODBC'  # matches the System DSN you created

def get_cursor():
    # For Driver 17: either set Encrypt=No in the DSN,
    # or keep Encrypt=Yes and TrustServerCertificate=Yes.
    conn = odbc.connect(
        f"DSN={data_source};"
        "Database=LibraryDB;"
        "Encrypt=no;"
    )
    return conn.cursor()

cursor = get_cursor()
#hashing and simple validators
import hashlib, re

def generate_md5(password_str: str) -> str:
    md5 = hashlib.md5()
    md5.update(password_str.encode('utf-8'))
    return md5.hexdigest()                # For production use bcrypt/argon2

def email_validation(email_str: str) -> bool:
    regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,7}\b'
    return bool(re.fullmatch(regex, email_str))

def phone_number_validation(phone_str: str) -> bool:
    regex = r'^\+?\d{10,15}$'             # portable E.164-ish; repo uses a stricter version
    return bool(re.fullmatch(regex, phone_str))
#seed runner (simplified)
import os
from typing import Iterable

def insert_data_to_tables():
    cursor = get_cursor()
    for file in os.listdir():
        if file.endswith('.sql'):
            with open(file, encoding='utf-8') as f:
                statements = [s.strip() for s in f.read().split(';') if s.strip()]
            table = file.removesuffix('.sql').split('-')[1]   # insert-TableName.sql
            cursor.execute(f"SELECT COUNT(*) FROM {table};")
            (count,) = cursor.fetchone()
            if count in (0, 1):                               # seed when empty
                for stmt in statements:
                    cursor.execute(stmt)
                cursor.commit()

# Highlights:
# creating_table(), creating_indexes(), add_constraints(), create_admin(), insert_data_to_tables()
#catalog guards (simplified)
def check_table_exists(cursor, table_name: str) -> bool:
    q = ("SELECT COUNT(*) FROM information_schema.tables "
         "WHERE table_name = ? AND table_type = 'BASE TABLE';")
    cursor.execute(q, (table_name,))
    return cursor.fetchone()[0] > 0

def check_index_exists(cursor, index_name: str) -> bool:
    cursor.execute("SELECT COUNT(*) FROM sys.indexes WHERE name = ?;", (index_name,))
    return cursor.fetchone()[0] > 0

def check_constraint_exists(cursor, constraint_name: str) -> bool:
    q = "SELECT COUNT(*) FROM information_schema.table_constraints WHERE constraint_name = ?;"
    cursor.execute(q, (constraint_name,))
    return cursor.fetchone()[0] > 0
#core DDL (excerpt)
CREATE TABLE book (
  book_id INT IDENTITY(1,1) PRIMARY KEY,
  name        NVARCHAR(200) NOT NULL,
  author      NVARCHAR(150),
  date_of_publish DATE,
  price       DECIMAL(10,2) NOT NULL,
  status      NVARCHAR(50) NOT NULL DEFAULT 'In Stock'
              CHECK (status IN ('In Stock','Out of Stock','Borrowed')),
  category_id   INT NOT NULL FOREIGN KEY REFERENCES category(category_id),
  publisher_id  INT NOT NULL FOREIGN KEY REFERENCES publisher(publisher_id)
);
#indexes (excerpt)
CREATE INDEX IX_book_name    ON book(name);
CREATE INDEX IX_book_author  ON book(author);
CREATE INDEX IX_member_user  ON member(username);
CREATE INDEX IX_member_name  ON member(last_name, first_name);
CREATE INDEX IX_pub_name     ON publisher(name);
CREATE INDEX IX_cat_name     ON category(name);
#FK constraints (excerpt)
ALTER TABLE user_tbl
  ADD CONSTRAINT fk_user_tbl_member
  FOREIGN KEY (member_id) REFERENCES member(member_id)
  ON UPDATE CASCADE ON DELETE CASCADE;

ALTER TABLE transactions
  ADD CONSTRAINT fk_tx_book
  FOREIGN KEY (book_id) REFERENCES book(book_id)
  ON UPDATE CASCADE ON DELETE CASCADE;  -- likewise for category/member as designed
#create an initial admin if none exists
from utils import generate_md5

def create_admin():
    # Check if an admin already exists
    check_admin_query = "SELECT COUNT(*) FROM member WHERE role = 'Admin';"
    cursor.execute(check_admin_query)
    admin_count = cursor.fetchone()[0]

    if admin_count == 0:
        # If no admin exists, create a new admin
        creating_admin_query = """
            INSERT INTO member (username, password, first_name, last_name, email, address, phone, role)
            VALUES ('admin', ?, 'Saber', 'Sojudi', 'sojudisaber@gmail.com', 'Germany', '+49151111111', 'Admin')
        """
        hashed_password = generate_md5("123456")
        cursor.execute(creating_admin_query, hashed_password)
        cursor.connection.commit()
        print("Admin Created Successfully!")
    else:
        pass

# Highlights:
# - Validate fields, enforce unique username
# - Hash password; INSERT new member (parameterized); commit
# Required fields + matching passwords + email/phone validation
if firstname_Entry.get()=='' or lastname_Entry.get()=='' or username_Entry.get()=='' \
   or email_entry.get()=='' or phone_num_entry.get()=='' or address_entry.get()=='' \
   or password_Entry.get()=='' or re_enter_password_Entry.get()=='':
    messagebox.showerror('Alert!', 'All Fields must be entered')
elif password_Entry.get() != re_enter_password_Entry.get():
    messagebox.showerror('Alert!', "Passwords didn't Match")
elif not email_validation(email_entry.get()) or not phone_number_validation(phone_num_entry.get()):
    messagebox.showerror('Alert!', 'Email Address Or Phone Number Are not valid!')
# Prevent duplicate usernames, then insert with hashed password
cursor = get_cursor()
cursor.execute("SELECT username, COUNT(*) FROM member WHERE username=? GROUP BY username;", username_Entry.get())
row = cursor.fetchone()
if row and row[1] > 0:
    messagebox.showerror("Alert!", f"This username: {row[0]} Is Taken!")
else:
    cursor.execute(
        "INSERT INTO member (username,password,first_name,last_name,email,address,phone,role) "
        "VALUES (?, ?, ?, ?, ?, ?, ?, 'User')",
        (username_Entry.get(), generate_md5(password_Entry.get()),
         firstname_Entry.get(), lastname_Entry.get(),
         email_entry.get(), address_entry.get(), phone_num_entry.get())
    )
    cursor.commit()
    messagebox.showinfo('Success', 'Account created successfully')

# Highlights:
# - initialize_database()
# - SELECT by username; compare hashes; route by role to admin_panel/user_panel
# One-time, idempotent DB setup before auth
def initialize_database():
    cursor = get_cursor()
    creating_table()
    creating_indexes()
    create_admin()
    insert_data_to_tables()
    add_constraints()
    return cursor
    # Fetch user and verify password hash, then route by role
    cursor.execute("SELECT * FROM member WHERE username=?", (idEntry.get(),))
    user = cursor.fetchone()
    if not user:
        messagebox.showerror('Alert!', 'User not found')
    else:
        stored_hash = user[2]
        if generate_md5(password_entry.get()) == stored_hash:
            role = user[-1]
            if role == 'Admin':
                admin_panel.main(user[1])   # go to admin UI
            elif role == 'User':
                user_panel.main(user[0])    # go to user UI
            else:
                messagebox.showerror('Alert!', 'Unknown role')
        else:
            messagebox.showerror('Alert!', 'Incorrect password')

# Highlights:
# - Validate identity; update hash via parameterized UPDATE; commit
# Lookup user safely, then update to a new hashed password
cursor.execute("SELECT * FROM member WHERE username=?", usernameEntry.get())
user = cursor.fetchone()
if user is None:
    messagebox.showerror('Error', 'User not found')
else:
    if NewpasswordEntry.get() != ReEnterNewPasswordEntry.get():
        messagebox.showerror('Error', "Passwords didn't match")
    else:
        cursor.execute("UPDATE member SET password=? WHERE username=?",
                       (generate_md5(NewpasswordEntry.get()), usernameEntry.get()))
        cursor.commit()
        messagebox.showinfo('Success', 'Password updated successfully')

# Highlights:
  # admin_panel.py / userList.py
  # admin_panel.py — navigate to sub-screens, confirm logout
  def show_book_list(username):
      windows.destroy()
      booklist.main(username)

  def profile(username):
      windows.destroy()
      admin_profile.main(username)

  def user_List(username):
      windows.destroy()
      userList.main(username)

  def logout():
      confirm = messagebox.askyesno("Confirmation", "Are you sure you want to Log Out?")
      if confirm:
          messagebox.showinfo("Success", "Logged Out successfully.")
          windows.destroy()
          login.main()
# admin_profile.py — parameterized lookup then populate Treeview
query = """
    SELECT m.member_id, m.username, m.first_name, m.last_name,
           m.email, m.address, m.phone, m.role
    FROM member m WHERE m.username = ?;
"""
cursor.execute(query, username)
row = cursor.fetchone()
tree.insert('', 'end', values=(row[0], row[1], row[2], row[3], row[4], row[5], row[6], row[7]))
# admin_edit_profile.py — update only fields that were provided
non_empty = []
if firstname_Entry.get(): non_empty.append(("first_name", firstname_Entry.get()))
if password_Entry.get():  non_empty.append(("password", generate_md5(password_Entry.get())))

cursor = get_cursor()
for field, value in non_empty:
    cursor.execute("UPDATE member SET " + field + " = ? WHERE username = ?", (value, username))
    cursor.commit()
# userList.py — list/clear/search in Treeview
def allusers(tree):
    cursor = get_cursor()
    for row in tree.get_children(): tree.delete(row)
    cursor.execute("""
        SELECT m.member_id, m.username, m.first_name, m.last_name,
               m.email, m.address, m.phone, m.role, u.status
        FROM member m LEFT JOIN user_tbl u ON m.member_id = u.member_id
        WHERE u.status IN ('Suspend', 'Valid');
    """)
    for u in cursor.fetchall():
        tree.insert('', 'end', values=(u[0], u[1], u[2], u[3], u[4], u[5], u[6], u[7], u[8]))
# userList.py — mark user as suspended
selected_id = tree.item(tree.selection(), 'values')[0]
cursor.execute("UPDATE user_tbl SET status = 'Suspend' WHERE member_id = ?", selected_id)
cursor.commit()
allusers(tree)
# userList.py — remove user safely (delete child rows first)
selected_id = tree.item(tree.selection(), 'values')[0]
cursor.execute("DELETE FROM transactions WHERE member_id = ?", selected_id)
cursor.execute("DELETE FROM user_tbl     WHERE member_id = ?", selected_id)
cursor.execute("DELETE FROM member       WHERE member_id = ?", selected_id)
cursor.commit()
allusers(tree)
# userList.py — mark user as suspended
selected_id = tree.item(tree.selection(), 'values')[0]
cursor.execute("UPDATE user_tbl SET status = 'Suspend' WHERE member_id = ?", selected_id)
cursor.commit()
allusers(tree)

# Highlights:
# user_panel.py / user_profile.py / user_edit_profile.py
# user_panel.py — pass user_id to target screens
subscription_plans = Button(frame, text='Subscription Plans',
    command=lambda: subscription_btn(user_id))
borrow_book = Button(frame, text='Borrow a Book',
    command=lambda: borrow_btn(user_id))
return_book = Button(frame, text='Return a Book',
    command=bokReturn)
profile = Button(frame, text='Profile',
    command=lambda: userprofile(user_id))
# user_profile.py — get a single user's details
query = """
    SELECT m.username, m.first_name, m.last_name, m.email,
           m.address, m.phone, m.role
    FROM member m WHERE m.member_id = ?;
"""
cursor.execute(query, user_id)
row = cursor.fetchone()
tree.insert('', 'end', values=(row[0], row[1], row[2], row[3], row[4], row[5], row[6]))
# user_edit_profile.py — collect non-empty fields then update
non_empty = []
if username_Entry.get(): non_empty.append(("username", username_Entry.get()))
if password_Entry.get(): non_empty.append(("password", generate_md5(password_Entry.get())))

for field, value in non_empty:
    cursor.execute("UPDATE member SET " + field + " = ? WHERE member_id = ?", (value, user_id))
    cursor.commit()

# Highlights:
# - Lookup publisher_id; INSERT book; link category
# 1) Find publisher_id by name (parameterized)
cursor = get_cursor()
cursor.execute("SELECT publisher_id FROM publisher WHERE name=?", bookPublisher_Entry.get())
row = cursor.fetchone()
if not row:
    messagebox.showerror("Alert!", "Publisher not found."); return
publisher_id = row[0]

# 2) Insert book with initial status
cursor.execute("""
    INSERT INTO book (name, price, publisher_id, author, status, date_of_publish)
    VALUES (?, ?, ?, ?, ?, ?)
""", (bookname_Entry.get(), float(bookprice_Entry.get()), publisher_id,
      bookauthor_Entry.get(), "In Stock", bookdate_Entry.get()))
cursor.commit()

# 3) Retrieve new book_id and attach a category
cursor.execute("SELECT book_id FROM book WHERE name=?", bookname_Entry.get())
book_id = cursor.fetchone()[0]

cursor.execute("INSERT INTO category (book_id, name) VALUES (?, ?)",
               (book_id, bookcategory_Entry.get()))
cursor.commit()

# Highlights:
-- JOIN example
SELECT b.book_id, b.name AS book_name, b.price, p.name AS publisher_name,
       b.author, b.status, c.name AS category_name
FROM   book b
JOIN   publisher p ON b.publisher_id = p.publisher_id
LEFT JOIN category  c ON b.book_id = c.book_id;
# Example: search by author (parameterized LIKE)
cursor.execute("""
  SELECT b.book_id, b.name, b.price, p.name, b.author, b.status, c.name
  FROM   book b JOIN publisher p ON b.publisher_id = p.publisher_id
  LEFT JOIN category c ON b.book_id = c.book_id
  WHERE  LOWER(b.author) LIKE ?;
""", ('%' + search_var.get().lower() + '%',))
# Delete category rows, then the book (parameterized + commit)
cursor.execute("DELETE FROM category WHERE book_id = ?", (selected_book_id,))
cursor.execute("DELETE FROM book     WHERE book_id = ?", (selected_book_id,))
cursor.commit()

# Highlights:
# INSERT Borrow; UPDATE status='Borrowed'
# Borrow flow (parameterized). Requires 'In Stock' selection in the UI.
insert_tx = """
  INSERT INTO transactions (member_id, book_id, status, transaction_date)
  VALUES (?, ?, ?, ?)
  UPDATE book SET status = ? WHERE book_id = ?;
"""
cursor.execute(insert_tx, user_id, book_id, "Borrow", date.today(), "Borrowed", book_id)
cursor.commit()

# Highlights:
# INSERT Return; UPDATE status='In Stock'
# Count Borrow vs Return per book, then list those with Borrow-Return == 1
book_counts = {}
for book_id, name, tx_date, status, member_id in books:
    book_counts.setdefault(book_id, {'Borrow': 0, 'Return': 0})
    book_counts[book_id][status] += 1
borrowed_books = [bid for bid, c in book_counts.items() if c['Borrow'] - c['Return'] == 1]

# Highlights:
# Read/extend expire_date with relativedelta
# Fetch expire_date and render "days remaining"
fetch_user_query = "SELECT expire_date FROM user_tbl WHERE member_id = ?"
cursor.execute(fetch_user_query, user_id)
expire_date = cursor.fetchall()[0][0]
today = date.today()
heading = Label(frame, text=str((expire_date - today).days) + " days of your\nsubscription remains")
# Compute new expiry and persist it safely
from dateutil.relativedelta import relativedelta

def add_months(months: int):
    new_date = expire_date + relativedelta(months=months)
    cursor.execute("UPDATE user_tbl SET expire_date = ? WHERE member_id = ?", (new_date, user_id))
    cursor.commit()
    messagebox.showinfo('Success', 'Your membership has been successfully Renewed.')

# Highlights:
# List suspended; confirm restore → status='Valid'
SELECT m.member_id AS user_id, m.username, m.first_name, m.last_name,
       m.email, m.address, m.phone, m.role, u.status
FROM   member m
LEFT JOIN user_tbl u ON m.member_id = u.member_id
WHERE  u.status IN ('Suspend');
# Confirm, then update status and refresh table
selected_item = tree.selection()
if not selected_item:
    messagebox.showinfo("Information", "Please select a user.")
else:
    if messagebox.askyesno("Confirmation", "Release the user from blacklist?"):
        selected_user_id = tree.item(selected_item, 'values')[0]
        cursor.execute("UPDATE user_tbl SET status = 'Valid' WHERE member_id = ?", (selected_user_id,))
        cursor.commit()
        allusers(tree)  # refresh
        messagebox.showinfo("Success", "User Removed From BlackList successfully.")
# Utility: clear table before re-listing
def clear_the_list(tree):
  for row in tree.get_children():
      tree.delete(row)