Database Interaction with SQLite

SQLite is like a digital filing cabinet. Each drawer is a database, each folder is a table, and each paper inside is a row of data. Python is the assistant who helps you organize, find, and update the papers in the cabinet.

# Database Interaction with SQLite
import sqlite3
import csv
import os
from datetime import datetime

print("DATABASE INTERACTION WITH SQLITE")
print("=" * 60)

# Example 1: Creating a database and table
print("\n1. CREATING A DATABASE AND TABLE")
print("-" * 30)

# Connect to database (creates if doesn't exist)
db_name = "library.db"
connection = sqlite3.connect(db_name)
cursor = connection.cursor()

print(f"Connected to database: {db_name}")

# Create books table
create_table_query = '''
CREATE TABLE IF NOT EXISTS books (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT NOT NULL,
    author TEXT NOT NULL,
    published_year INTEGER,
    genre TEXT,
    price REAL,
    in_stock BOOLEAN DEFAULT 1,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
'''

cursor.execute(create_table_query)
print("Created 'books' table")

# Show table structure
cursor.execute("PRAGMA table_info(books)")
columns = cursor.fetchall()

print("\nTable Structure:")
print("-" * 60)
for col in columns:
    print(f"{col[1]:15} {col[2]:10} {'PRIMARY KEY' if col[5] == 1 else ''}")

# Example 2: Inserting data
print("\n\n2. INSERTING DATA")
print("-" * 30)

# Insert single book
insert_query = '''
INSERT INTO books (title, author, published_year, genre, price, in_stock)
VALUES (?, ?, ?, ?, ?, ?)
'''

book_data = ("Python Programming", "John Doe", 2023, "Programming", 29.99, True)
cursor.execute(insert_query, book_data)
connection.commit()
print(f"Inserted 1 book: {book_data[0]}")

# Insert multiple books
books = [
    ("Data Science Basics", "Jane Smith", 2022, "Data Science", 34.99, True),
    ("Web Development", "Bob Johnson", 2021, "Web", 24.99, True),
    ("Machine Learning", "Alice Brown", 2023, "AI", 39.99, False),
    ("Database Design", "Charlie Wilson", 2020, "Database", 27.50, True)
]

cursor.executemany(insert_query, books)
connection.commit()
print(f"Inserted {len(books)} more books")

# Example 3: Querying data (SELECT)
print("\n\n3. QUERYING DATA (SELECT)")
print("-" * 30)

# Select all books
print("All Books:")
print("-" * 60)
cursor.execute("SELECT * FROM books")
all_books = cursor.fetchall()

for book in all_books:
    print(f"ID: {book[0]:2} | {book[1]:25} | {book[2]:15} | ${book[5]:5.2f} | {'In Stock' if book[6] else 'Out of Stock'}")

# Select specific columns with WHERE clause
print("\nProgramming Books in Stock:")
print("-" * 60)
query = "SELECT title, author, price FROM books WHERE genre = ? AND in_stock = 1"
cursor.execute(query, ("Programming",))
programming_books = cursor.fetchall()

for book in programming_books:
    print(f"{book[0]:25} by {book[1]:15} - ${book[2]:.2f}")

# Select with ORDER BY and LIMIT
print("\nTop 3 Most Expensive Books:")
print("-" * 60)
cursor.execute("SELECT title, price FROM books ORDER BY price DESC LIMIT 3")
expensive_books = cursor.fetchall()

for book in expensive_books:
    print(f"{book[0]:25} - ${book[1]:.2f}")

# Example 4: Updating data
print("\n\n4. UPDATING DATA")
print("-" * 30)

# Update price of a book
update_query = "UPDATE books SET price = ? WHERE id = ?"
cursor.execute(update_query, (24.99, 1))  # Update book with id=1
connection.commit()
print("Updated price of Python Programming to $24.99")

# Mark all books published before 2022 as out of stock
cursor.execute("UPDATE books SET in_stock = 0 WHERE published_year < 2022")
connection.commit()
affected = cursor.rowcount
print(f"Marked {affected} older books as out of stock")

# Show updated data
print("\nUpdated Book Status:")
print("-" * 60)
cursor.execute("SELECT title, published_year, in_stock FROM books ORDER BY published_year DESC")
updated_books = cursor.fetchall()

for book in updated_books:
    status = "✓ In Stock" if book[2] else "✗ Out of Stock"
    print(f"{book[0]:25} ({book[1]}) - {status}")

# Example 5: Deleting data
print("\n\n5. DELETING DATA")
print("-" * 30)

# Show books before deletion
cursor.execute("SELECT COUNT(*) FROM books")
count_before = cursor.fetchone()[0]
print(f"Books before deletion: {count_before}")

# Delete a specific book (simulating removal of out of stock books)
delete_query = "DELETE FROM books WHERE id = ?"
cursor.execute(delete_query, (4,))  # Delete book with id=4 (Machine Learning)
connection.commit()
print("Deleted book with ID 4 (Machine Learning)")

# Show books after deletion
cursor.execute("SELECT COUNT(*) FROM books")
count_after = cursor.fetchone()[0]
print(f"Books after deletion: {count_after}")
print(f"Total deleted: {count_before - count_after}")

# Example 6: Advanced queries
print("\n\n6. ADVANCED QUERIES")
print("-" * 30)

# Group by genre
print("Books per Genre:")
print("-" * 40)
group_query = """
SELECT genre, COUNT(*) as count, AVG(price) as avg_price
FROM books 
GROUP BY genre
ORDER BY count DESC
"""
cursor.execute(group_query)
genre_stats = cursor.fetchall()

for stat in genre_stats:
    print(f"{stat[0]:15} : {stat[1]} books | Average price: ${stat[2]:.2f}")

# Search with LIKE
print("\nSearch for books with 'Python':")
print("-" * 40)
search_query = "SELECT title, author FROM books WHERE title LIKE ?"
cursor.execute(search_query, ('%Python%',))
search_results = cursor.fetchall()

for result in search_results:
    print(f"{result[0]} by {result[1]}")

# Example 7: Working with multiple tables
print("\n\n7. WORKING WITH MULTIPLE TABLES")
print("-" * 30)

# Create authors table
cursor.execute('''
CREATE TABLE IF NOT EXISTS authors (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    country TEXT,
    birth_year INTEGER
)
''')

# Create book_authors table (junction table for many-to-many)
cursor.execute('''
CREATE TABLE IF NOT EXISTS book_authors (
    book_id INTEGER,
    author_id INTEGER,
    PRIMARY KEY (book_id, author_id),
    FOREIGN KEY (book_id) REFERENCES books(id),
    FOREIGN KEY (author_id) REFERENCES authors(id)
)
''')

print("Created authors and book_authors tables")

# Insert sample authors
authors = [
    ("John Doe", "USA", 1980),
    ("Jane Smith", "UK", 1975),
    ("Bob Johnson", "Canada", 1985)
]

cursor.executemany("INSERT INTO authors (name, country, birth_year) VALUES (?, ?, ?)", authors)
connection.commit()
print(f"Inserted {len(authors)} authors")

# Create relationships
book_author_relationships = [
    (1, 1),  # Python Programming by John Doe
    (2, 2),  # Data Science Basics by Jane Smith
    (3, 3),  # Web Development by Bob Johnson
    (5, 1)   # Database Design also by John Doe
]

cursor.executemany("INSERT INTO book_authors (book_id, author_id) VALUES (?, ?)", book_author_relationships)
connection.commit()
print("Created book-author relationships")

# JOIN query to get books with authors
print("\nBooks with Authors (JOIN):")
print("-" * 60)
join_query = """
SELECT b.title, a.name, a.country
FROM books b
JOIN book_authors ba ON b.id = ba.book_id
JOIN authors a ON ba.author_id = a.id
ORDER BY b.title
"""

cursor.execute(join_query)
books_with_authors = cursor.fetchall()

for book in books_with_authors:
    print(f"{book[0]:25} by {book[1]:15} ({book[2]})")

# Example 8: Database functions and aggregation
print("\n\n8. DATABASE FUNCTIONS AND AGGREGATION")
print("-" * 30)

print("Database Statistics:")
print("-" * 40)

stats_queries = [
    ("Total Books", "SELECT COUNT(*) FROM books"),
    ("Total Authors", "SELECT COUNT(*) FROM authors"),
    ("Average Book Price", "SELECT AVG(price) FROM books"),
    ("Most Expensive Book", "SELECT MAX(price) FROM books"),
    ("Cheapest Book", "SELECT MIN(price) FROM books"),
    ("Total Inventory Value", "SELECT SUM(price) FROM books WHERE in_stock = 1")
]

for stat_name, query in stats_queries:
    cursor.execute(query)
    result = cursor.fetchone()[0]
    if isinstance(result, float):
        print(f"{stat_name:25}: ${result:.2f}")
    else:
        print(f"{stat_name:25}: {result}")

# Example 9: Exporting and importing data
print("\n\n9. EXPORTING AND IMPORTING DATA")
print("-" * 30)

# Export books to CSV
export_query = "SELECT title, author, genre, price, in_stock FROM books"
cursor.execute(export_query)
books_data = cursor.fetchall()

# Write to CSV
export_filename = "books_export.csv"
with open(export_filename, 'w', newline='', encoding='utf-8') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow(['Title', 'Author', 'Genre', 'Price', 'In Stock'])  # Header
    writer.writerows(books_data)

print(f"Exported {len(books_data)} books to {export_filename}")

# Read from CSV and import to new table
print("\nImporting from CSV to new table...")

# Create import table
cursor.execute('''
CREATE TABLE IF NOT EXISTS imported_books (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    title TEXT,
    author TEXT,
    genre TEXT,
    price REAL,
    in_stock BOOLEAN
)
''')

# Read CSV and insert
imported_count = 0
with open(export_filename, 'r', newline='', encoding='utf-8') as csvfile:
    reader = csv.reader(csvfile)
    next(reader)  # Skip header
    
    for row in reader:
        # Convert 'True'/'False' string to boolean
        in_stock = row[4].lower() == 'true' if row[4] else False
        cursor.execute(
            "INSERT INTO imported_books (title, author, genre, price, in_stock) VALUES (?, ?, ?, ?, ?)",
            (row[0], row[1], row[2], float(row[3]), in_stock)
        )
        imported_count += 1

connection.commit()
print(f"Imported {imported_count} books to imported_books table")

# Verify import
cursor.execute("SELECT COUNT(*) FROM imported_books")
imported_total = cursor.fetchone()[0]
print(f"Total books in imported_books table: {imported_total}")

# Example 10: Complete Library Management System
print("\n\n10. COMPLETE LIBRARY MANAGEMENT SYSTEM")
print("-" * 30)

class LibraryDB:
    """Library database management class"""
    
    def __init__(self, db_name="library_system.db"):
        self.connection = sqlite3.connect(db_name)
        self.cursor = self.connection.cursor()
        self.setup_database()
    
    def setup_database(self):
        """Create all necessary tables"""
        # Books table
        self.cursor.execute('''
        CREATE TABLE IF NOT EXISTS books (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            isbn TEXT UNIQUE,
            title TEXT NOT NULL,
            author TEXT NOT NULL,
            year INTEGER,
            available BOOLEAN DEFAULT 1
        )
        ''')
        
        # Members table
        self.cursor.execute('''
        CREATE TABLE IF NOT EXISTS members (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            member_id TEXT UNIQUE,
            name TEXT NOT NULL,
            email TEXT UNIQUE,
            joined_date DATE DEFAULT CURRENT_DATE
        )
        ''')
        
        # Loans table
        self.cursor.execute('''
        CREATE TABLE IF NOT EXISTS loans (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            book_id INTEGER,
            member_id INTEGER,
            loan_date DATE DEFAULT CURRENT_DATE,
            due_date DATE,
            returned_date DATE,
            FOREIGN KEY (book_id) REFERENCES books(id),
            FOREIGN KEY (member_id) REFERENCES members(id)
        )
        ''')
        
        self.connection.commit()
        print("Library database setup complete")
    
    def add_book(self, isbn, title, author, year):
        """Add a new book to the library"""
        try:
            self.cursor.execute('''
            INSERT INTO books (isbn, title, author, year)
            VALUES (?, ?, ?, ?)
            ''', (isbn, title, author, year))
            self.connection.commit()
            print(f"Added book: {title}")
            return True
        except sqlite3.IntegrityError:
            print(f"Book with ISBN {isbn} already exists")
            return False
    
    def add_member(self, member_id, name, email):
        """Add a new library member"""
        try:
            self.cursor.execute('''
            INSERT INTO members (member_id, name, email)
            VALUES (?, ?, ?)
            ''', (member_id, name, email))
            self.connection.commit()
            print(f"Added member: {name}")
            return True
        except sqlite3.IntegrityError as e:
            print(f"Error adding member: {e}")
            return False
    
    def borrow_book(self, book_id, member_id, days=14):
        """Borrow a book"""
        # Check if book is available
        self.cursor.execute("SELECT available FROM books WHERE id = ?", (book_id,))
        book = self.cursor.fetchone()
        
        if not book:
            print("Book not found")
            return False
        
        if not book[0]:  # available = False
            print("Book is already borrowed")
            return False
        
        # Calculate due date
        from datetime import datetime, timedelta
        loan_date = datetime.now().date()
        due_date = loan_date + timedelta(days=days)
        
        # Create loan record
        self.cursor.execute('''
        INSERT INTO loans (book_id, member_id, loan_date, due_date)
        VALUES (?, ?, ?, ?)
        ''', (book_id, member_id, loan_date, due_date))
        
        # Update book availability
        self.cursor.execute("UPDATE books SET available = 0 WHERE id = ?", (book_id,))
        
        self.connection.commit()
        print(f"Book borrowed successfully. Due date: {due_date}")
        return True
    
    def return_book(self, loan_id):
        """Return a borrowed book"""
        # Get loan details
        self.cursor.execute("SELECT book_id FROM loans WHERE id = ? AND returned_date IS NULL", (loan_id,))
        loan = self.cursor.fetchone()
        
        if not loan:
            print("Loan not found or already returned")
            return False
        
        book_id = loan[0]
        
        # Update loan record
        return_date = datetime.now().date()
        self.cursor.execute('''
        UPDATE loans SET returned_date = ? WHERE id = ?
        ''', (return_date, loan_id))
        
        # Update book availability
        self.cursor.execute("UPDATE books SET available = 1 WHERE id = ?", (book_id,))
        
        self.connection.commit()
        print(f"Book returned successfully on {return_date}")
        return True
    
    def search_books(self, keyword):
        """Search books by title or author"""
        search_term = f"%{keyword}%"
        self.cursor.execute('''
        SELECT b.id, b.title, b.author, b.year, 
               CASE WHEN b.available THEN 'Available' ELSE 'Borrowed' END as status
        FROM books b
        WHERE b.title LIKE ? OR b.author LIKE ?
        ORDER BY b.title
        ''', (search_term, search_term))
        
        return self.cursor.fetchall()
    
    def get_overdue_loans(self):
        """Get all overdue loans"""
        self.cursor.execute('''
        SELECT l.id, b.title, m.name, l.due_date,
               julianday('now') - julianday(l.due_date) as days_overdue
        FROM loans l
        JOIN books b ON l.book_id = b.id
        JOIN members m ON l.member_id = m.id
        WHERE l.returned_date IS NULL AND l.due_date < date('now')
        ORDER BY l.due_date
        ''')
        
        return self.cursor.fetchall()
    
    def close(self):
        """Close database connection"""
        self.connection.close()
        print("Database connection closed")

# Use the library system
print("Library Management System Demo:")
library = LibraryDB("library_system.db")

# Add some books
library.add_book("978-0134670942", "Python Crash Course", "Eric Matthes", 2019)
library.add_book("978-1491957660", "Fluent Python", "Luciano Ramalho", 2021)
library.add_book("978-1449369415", "Automate the Boring Stuff", "Al Sweigart", 2020)

# Add members
library.add_member("M001", "Alice Johnson", "alice@email.com")
library.add_member("M002", "Bob Smith", "bob@email.com")

# Search books
print("\nSearching for 'Python' books:")
print("-" * 60)
python_books = library.search_books("Python")
for book in python_books:
    print(f"ID: {book[0]:2} | {book[1]:30} | {book[2]:20} | {book[3]} | {book[4]}")

# Borrow a book
print("\nBorrowing a book:")
library.borrow_book(1, 1)  # Book ID 1, Member ID 1

# Search again to see status change
print("\nSearching again (book should be borrowed):")
books = library.search_books("Crash Course")
for book in books:
    print(f"{book[1]} - {book[4]}")

# Return the book
print("\nReturning the book:")
library.return_book(1)  # Loan ID 1

# Check overdue loans (none yet)
print("\nChecking overdue loans:")
overdue = library.get_overdue_loans()
if overdue:
    for loan in overdue:
        print(f"{loan[1]} borrowed by {loan[2]} is {loan[4]:.0f} days overdue")
else:
    print("No overdue loans")

# Close connection
library.close()

# Clean up demo files
import os
if os.path.exists("library.db"):
    os.remove("library.db")
if os.path.exists("library_system.db"):
    os.remove("library_system.db")
if os.path.exists("books_export.csv"):
    os.remove("books_export.csv")

print("\nDemo databases cleaned up")

→ Run this code interactively