Module 24 - Database Access

Python's sqlite3 module provides a lightweight disk-based database that doesn't require a separate server. It's perfect for learning SQL and building small to medium applications.

---

1. Introduction to SQLite

SQLite is a self-contained, serverless SQL database engine embedded in Python's standard library.

Advantages

✅ No separate server required
✅ Zero configuration
✅ Cross-platform
✅ Lightweight and fast
✅ ACID compliant

When to Use SQLite

• Desktop applications
• Mobile apps
• Small to medium websites
• Embedded systems
• Development and testing

---

2. Connecting to a Database

import sqlite3

# Connect to database (creates if doesn't exist)
conn = sqlite3.connect('mydata base.db')

# Create cursor object
cursor = conn.cursor()

# Execute SQL commands...

# Close connection
conn.close()

Using Context Manager (Recommended)

import sqlite3

with sqlite3.connect('mydatabase.db') as conn:
    cursor = conn.cursor()
    # Execute SQL commands
    # Automatically commits and closes

In-Memory Database

# Useful for testing
conn = sqlite3.connect(':memory:')

---

3. Creating Tables

Basic Table Creation

import sqlite3

with sqlite3.connect('mydatabase.db') as conn:
    cursor = conn.cursor()
    
    # Create table
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            username TEXT NOT NULL UNIQUE,
            email TEXT NOT NULL,
            age INTEGER,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    ''')
    
    conn.commit()
    print("Table created successfully")

Data Types in SQLite

SQLite Type	Python Type	Description
NULL	None	Null value
INTEGER	int	Signed integer
REAL	float	Floating point
TEXT	str	Text string
BLOB	bytes	Binary data

---

4. CRUD Operations

4.1 Create (INSERT)

import sqlite3

with sqlite3.connect('mydatabase.db') as conn:
    cursor = conn.cursor()
    
    # Insert single record
    cursor.execute('''
        INSERT INTO users (username, email, age)
        VALUES (?, ?, ?)
    ''', ('alice', 'alice@example.com', 25))
    
    # Insert multiple records
    users = [
        ('bob', 'bob@example.com', 30),
        ('charlie', 'charlie@example.com', 35),
        ('diana', 'diana@example.com', 28)
    ]
    cursor.executemany('''
        INSERT INTO users (username, email, age)
        VALUES (?, ?, ?)
    ''', users)
    
    conn.commit()
    print(f"Inserted {cursor.rowcount} rows")

SQL Injection

Always use parameterized queries with ? placeholders. Never use string formatting!

# ❌ NEVER DO THIS (SQL injection risk)
username = "alice'; DROP TABLE users; --"
cursor.execute(f"SELECT * FROM users WHERE username = '{username}'")

# ✅ ALWAYS DO THIS
cursor.execute("SELECT * FROM users WHERE username = ?", (username,))

4.2 Read (SELECT)

import sqlite3

with sqlite3.connect('mydatabase.db') as conn:
    cursor = conn.cursor()
    
    # Fetch all records
    cursor.execute('SELECT * FROM users')
    rows = cursor.fetchall()
    for row in rows:
        print(row)
    
    # Fetch one record
    cursor.execute('SELECT * FROM users WHERE username = ?', ('alice',))
    user = cursor.fetchone()
    print(user)
    
    # Fetch multiple records
    cursor.execute('SELECT * FROM users WHERE age > ?', (25,))
    users = cursor.fetchmany(2)  # Fetch 2 records
    print(users)
    
    # Iterate over results
    cursor.execute('SELECT username, email FROM users')
    for username, email in cursor:
        print(f"{username}: {email}")

4.3 Update (UPDATE)

import sqlite3

with sqlite3.connect('mydatabase.db') as conn:
    cursor = conn.cursor()
    
    # Update single record
    cursor.execute('''
        UPDATE users
        SET email = ?
        WHERE username = ?
    ''', ('newemail@example.com', 'alice'))
    
    # Update multiple fields
    cursor.execute('''
        UPDATE users
        SET age = ?, email = ?
        WHERE username = ?
    ''', (26, 'alice_new@example.com', 'alice'))
    
    conn.commit()
    print(f"Updated {cursor.rowcount} rows")

4.4 Delete (DELETE)

import sqlite3

with sqlite3.connect('mydatabase.db') as conn:
    cursor = conn.cursor()
    
    # Delete specific record
    cursor.execute('DELETE FROM users WHERE username = ?', ('bob',))
    
    # Delete with condition
    cursor.execute('DELETE FROM users WHERE age < ?', (25,))
    
    conn.commit()
    print(f"Deleted {cursor.rowcount} rows")

---

5. Advanced Queries

5.1 WHERE Clauses

import sqlite3

with sqlite3.connect('mydatabase.db') as conn:
    cursor = conn.cursor()
    
    # Single condition
    cursor.execute('SELECT * FROM users WHERE age > ?', (25,))
    
    # Multiple conditions (AND)
    cursor.execute('''
        SELECT * FROM users
        WHERE age > ? AND username LIKE ?
    ''', (25, '%a%'))
    
    # OR condition
    cursor.execute('''
        SELECT * FROM users
        WHERE age < ? OR username = ?
    ''', (25, 'alice'))
    
    # IN clause
    cursor.execute('''
        SELECT * FROM users
        WHERE username IN (?, ?, ?)
    ''', ('alice', 'bob', 'charlie'))
    
    results = cursor.fetchall()

5.2 Sorting and Limiting

# ORDER BY
cursor.execute('SELECT * FROM users ORDER BY age DESC')

# LIMIT
cursor.execute('SELECT * FROM users LIMIT 5')

# OFFSET (for pagination)
cursor.execute('SELECT * FROM users LIMIT 10 OFFSET 20')

# Combine
cursor.execute('''
    SELECT username, age FROM users
    WHERE age > ?
    ORDER BY age DESC
    LIMIT 5
''', (25,))

5.3 Aggregate Functions

import sqlite3

with sqlite3.connect('mydatabase.db') as conn:
    cursor = conn.cursor()
    
    # COUNT
    cursor.execute('SELECT COUNT(*) FROM users')
    count = cursor.fetchone()[0]
    print(f"Total users: {count}")
    
    # AVG
    cursor.execute('SELECT AVG(age) FROM users')
    avg_age = cursor.fetchone()[0]
    print(f"Average age: {avg_age:.1f}")
    
    # MIN and MAX
    cursor.execute('SELECT MIN(age), MAX(age) FROM users')
    min_age, max_age = cursor.fetchone()
    print(f"Age range: {min_age}-{max_age}")
    
    # GROUP BY
    cursor.execute('''
        SELECT age, COUNT(*) as count
        FROM users
        GROUP BY age
        ORDER BY count DESC
    ''')
    for age, count in cursor:
        print(f"Age {age}: {count} users")

---

6. Working with Multiple Tables

6.1 Creating Related Tables

import sqlite3

with sqlite3.connect('mydatabase.db') as conn:
    cursor = conn.cursor()
    
    # Create posts table
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS posts (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id INTEGER NOT NULL,
            title TEXT NOT NULL,
            content TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (user_id) REFERENCES users (id)
        )
    ''')
    
    conn.commit()

6.2 INNER JOIN

# Get posts with user information
cursor.execute('''
    SELECT users.username, posts.title, posts.content
    FROM posts
    INNER JOIN users ON posts.user_id = users.id
''')

for username, title, content in cursor:
    print(f"{username}: {title}")

6.3 LEFT JOIN

# Get all users and their posts (including users with no posts)
cursor.execute('''
    SELECT users.username, posts.title
    FROM users
    LEFT JOIN posts ON users.id = posts.user_id
''')

---

7. Practical Example: User Management System

import sqlite3
from typing import List, Dict, Optional
from datetime import datetime

class UserDatabase:
    def __init__(self, db_name: str = 'users.db'):
        self.db_name = db_name
        self._create_tables()
    
    def _create_tables(self):
        """Create database tables"""
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS users (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    username TEXT NOT NULL UNIQUE,
                    email TEXT NOT NULL,
                    age INTEGER,
                    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            ''')
            conn.commit()
    
    def add_user(self, username: str, email: str, age: int) -> int:
        """Add new user"""
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()
            cursor.execute('''
                INSERT INTO users (username, email, age)
                VALUES (?, ?, ?)
            ''', (username, email, age))
            conn.commit()
            return cursor.lastrowid
    
    def get_user(self, user_id: int) -> Optional[Dict]:
        """Get user by ID"""
        with sqlite3.connect(self.db_name) as conn:
            conn.row_factory = sqlite3.Row  # Return rows as dictionaries
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM users WHERE id = ?', (user_id,))
            row = cursor.fetchone()
            return dict(row) if row else None
    
    def get_all_users(self) -> List[Dict]:
        """Get all users"""
        with sqlite3.connect(self.db_name) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute('SELECT * FROM users')
            return [dict(row) for row in cursor.fetchall()]
    
    def update_user(self, user_id: int, **kwargs) -> bool:
        """Update user fields"""
        fields = ', '.join(f"{k} = ?" for k in kwargs.keys())
        values = list(kwargs.values()) + [user_id]
        
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()
            cursor.execute(f'''
                UPDATE users
                SET {fields}
                WHERE id = ?
            ''', values)
            conn.commit()
            return cursor.rowcount > 0
    
    def delete_user(self, user_id: int) -> bool:
        """Delete user"""
        with sqlite3.connect(self.db_name) as conn:
            cursor = conn.cursor()
            cursor.execute('DELETE FROM users WHERE id = ?', (user_id,))
            conn.commit()
            return cursor.rowcount > 0
    
    def search_users(self, keyword: str) -> List[Dict]:
        """Search users by username or email"""
        with sqlite3.connect(self.db_name) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.cursor()
            cursor.execute('''
                SELECT * FROM users
                WHERE username LIKE ? OR email LIKE ?
            ''', (f'%{keyword}%', f'%{keyword}%'))
            return [dict(row) for row in cursor.fetchall()]

# Usage
db = UserDatabase()

# Add users
user_id = db.add_user('alice', 'alice@example.com', 25)
print(f"Added user with ID: {user_id}")

# Get user
user = db.get_user(user_id)
print(f"User: {user}")

# Update user
db.update_user(user_id, email='alice_new@example.com', age=26)

# Search users
results = db.search_users('alice')
print(f"Search results: {results}")

# Delete user
db.delete_user(user_id)

---

8. Best Practices

8.1 Connection Pooling

import sqlite3
from contextlib import contextmanager

class DatabasePool:
    def __init__(self, db_name: str):
        self.db_name = db_name
    
    @contextmanager
    def get_connection(self):
        """Context manager for database connections"""
        conn = sqlite3.connect(self.db_name)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()

# Usage
pool = DatabasePool('mydatabase.db')

with pool.get_connection() as conn:
    cursor = conn.cursor()
    cursor.execute('SELECT * FROM users')
    users = cursor.fetchall()

8.2 Transactions

import sqlite3

with sqlite3.connect('mydatabase.db') as conn:
    cursor = conn.cursor()
    
    try:
        # Start transaction
        cursor.execute('BEGIN')
        
        # Multiple operations
        cursor.execute('UPDATE users SET age = age + 1 WHERE id = ?', (1,))
        cursor.execute('INSERT INTO logs (message) VALUES (?)', ('Age updated',))
        
        # Commit transaction
        conn.commit()
        print("Transaction committed")
    
    except Exception as e:
        # Rollback on error
        conn.rollback()
        print(f"Transaction rolled back: {e}")

---

9. SQLite Limitations

Feature	SQLite	MySQL/PostgreSQL
Concurrent writes	❌ Limited	✅ Full support
User permissions	❌ None	✅ Granular
Size limit	~140 TB	Much larger
Network access	❌ File-based	✅ Client-server
Use case	Small apps	Large apps

---

10. Migrating to Other Databases

Using SQLAlchemy (ORM)

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    
    id = Column(Integer, primary_key=True)
    username = Column(String, unique=True, nullable=False)
    email = Column(String, nullable=False)
    age = Column(Integer)

# SQLite
engine = create_engine('sqlite:///mydatabase.db')

# Easily switch to PostgreSQL or MySQL
# engine = create_engine('postgresql://user:pass@localhost/dbname')
# engine = create_engine('mysql://user:pass@localhost/dbname')

Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

---

Summary

✅ SQLite is embedded in Python's standard library
✅ Use parameterized queries to prevent SQL injection
✅ CRUD operations: INSERT, SELECT, UPDATE, DELETE
✅ Support for joins, aggregates, and complex queries
✅ Perfect for small to medium applications
✅ Use context managers for automatic cleanup

---

Next Steps

In Module 25, you'll learn:

• NumPy basics
• Array operations
• Mathematical computations
• Scientific computing with Python

---

Practice Exercises

1. Create a todo list application with SQLite backend
2. Build a contact management system with search functionality
3. Implement a simple blog with posts and comments tables
4. Create a library management system (books, authors, borrowers)
5. Build a database migration script to upgrade schema versions

Challenge

Create a complete inventory management system that:

• Tracks products, categories, and suppliers
• Supports transactions (sales, purchases, returns)
• Maintains stock levels with alerts
• Generates reports (sales by period, low stock)
• Includes full CRUD operations with proper error handling
• Uses transactions for data integrity