30. Databases & SQL

# ==============================================================================
# JUNIOR PYTHON TRACK 30 — DATABASES & SQL (sqlite3 & DB-API)
# ==============================================================================
import sqlite3

def init_database() -> sqlite3.Connection:
    """Connect to in-memory SQLite database and set Row factory."""
    conn = sqlite3.connect(":memory:")
    conn.row_factory = sqlite3.Row  # Enables dict-like column access
    return conn

def create_tables(conn: sqlite3.Connection):
    """Execute DDL statements to create users and orders tables."""
    with conn:
        conn.execute("""
            CREATE TABLE users (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                username TEXT UNIQUE NOT NULL,
                email TEXT UNIQUE NOT NULL,
                is_active INTEGER DEFAULT 1
            )
        """)
        conn.execute("""
            CREATE TABLE orders (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                user_id INTEGER NOT NULL,
                amount REAL NOT NULL,
                FOREIGN KEY (user_id) REFERENCES users (id)
            )
        """)

def insert_sample_data(conn: sqlite3.Connection):
    """Insert sample rows using parameterized binding (prevents SQL injection)."""
    users = [
        ("shkar_dev", "shkar@teachcloud.org", 1),
        ("alice_py", "alice@example.com", 1),
        ("bob_db", "bob@database.io", 0)
    ]
    orders = [
        (1, 150.50), (1, 49.99), (2, 299.00)
    ]
    with conn:
        conn.executemany("INSERT INTO users (username, email, is_active) VALUES (?, ?, ?)", users)
        conn.executemany("INSERT INTO orders (user_id, amount) VALUES (?, ?)", orders)

def get_user_summary(conn: sqlite3.Connection) -> list[dict]:
    """Perform SELECT query with JOIN, GROUP BY, and ORDER BY."""
    query = """
        SELECT u.username, COUNT(o.id) AS total_orders, COALESCE(SUM(o.amount), 0.0) AS total_spent
        FROM users u
        LEFT JOIN orders o ON u.id = o.user_id
        WHERE u.is_active = 1
        GROUP BY u.id, u.username
        ORDER BY total_spent DESC
    """
    cursor = conn.cursor()
    cursor.execute(query)
    return [{"username": row["username"], "orders": row["total_orders"], "spent": row["total_spent"]} for row in cursor.fetchall()]

if __name__ == "__main__":
    conn = init_database()
    create_tables(conn)
    insert_sample_data(conn)
    
    summary = get_user_summary(conn)
    print("=== Active User Sales Summary (JOIN & GROUP BY) ===")
    for item in summary:
        print(f"User: {item['username']:<12} | Orders: {item['orders']} | Total Spent: ${item['spent']:.2f}")
    
    conn.close()

$ python3 junior_databases.py
=== Active User Sales Summary (JOIN & GROUP BY) ===
User: alice_py     | Orders: 1 | Total Spent: $299.00
User: shkar_dev    | Orders: 2 | Total Spent: $200.49
Execution completed successfully with returncode 0.

import unittest
import sqlite3

class TestJuniorDatabases(unittest.TestCase):
    def setUp(self):
        self.conn = sqlite3.connect(":memory:")
        self.conn.row_factory = sqlite3.Row
        self.conn.execute("CREATE TABLE users (id INT PRIMARY KEY, name TEXT)")

    def tearDown(self):
        self.conn.close()

    def test_parameterized_insert_and_select(self):
        with self.conn:
            self.conn.execute("INSERT INTO users VALUES (?, ?)", (1, "Alice"))
        
        cursor = self.conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
        row = cursor.fetchone()
        self.assertEqual(row["name"], "Alice")

    def test_transaction_rollback(self):
        try:
            with self.conn:
                self.conn.execute("INSERT INTO users VALUES (?, ?)", (2, "Bob"))
                raise ValueError("Simulated Transaction Error")
        except ValueError:
            pass

        cursor = self.conn.cursor()
        cursor.execute("SELECT * FROM users WHERE id = ?", (2,))
        self.assertIsNone(cursor.fetchone())

if __name__ == "__main__":
    unittest.main()

⚡ Key Concept: SQL Commands & Database Architecture

• SQL DML Primitives: SELECT (retrieve), INSERT (create), UPDATE (modify), DELETE (remove).
• Query Clause Ordering: SELECT ... FROM ... JOIN ... WHERE ... GROUP BY ... HAVING ... ORDER BY ... LIMIT.
• Parameterized Queries: Always use ? bindings (or %s in Postgres) to prevent catastrophic SQL Injection attacks. Never concatenate SQL strings.
• RDBMS Engines: SQLite (embedded file/memory DB), PostgreSQL (enterprise server RDBMS with rich JSON & ACID concurrency), MySQL (widely used web server RDBMS).
• ORM Abstraction: SQLAlchemy bridges Python classes and SQL tables, allowing DB-agnostic declarative models, automated sessions, and migrations (via Alembic).