Lesson 8: SQLAlchemy — Database-Agnostic Connections

from sqlalchemy import create_engine
 
# SQLite in-memory — works in-browser (Pyodide)
engine = create_engine("sqlite:///:memory:", echo=False)
 
print(type(engine))  # <class 'sqlalchemy.engine.base.Engine'>
print(engine.dialect.name)  # sqlite
 
# echo=True logs every SQL statement SQLAlchemy generates — useful for debugging
# engine_debug = create_engine("sqlite:///:memory:", echo=True)

from sqlalchemy import create_engine, text
 
engine = create_engine("sqlite:///:memory:", echo=False)
 
# Connect and run a simple test query
with engine.connect() as conn:
    result = conn.execute(text("SELECT 1 AS alive"))
    row = result.fetchone()
    print(f"Connection alive: {row[0] == 1}")
    print(f"Dialect: {engine.dialect.name}")

from sqlalchemy import create_engine, text
 
engine = create_engine("sqlite:///:memory:", echo=False)
 
# Create table
with engine.begin() as conn:
    conn.execute(text("""
        CREATE TABLE employees (
            id         INTEGER PRIMARY KEY,
            name       TEXT NOT NULL,
            department TEXT,
            salary     REAL,
            hire_date  TEXT
        )
    """))
print("Table created")

# INSERT with named parameters — :param_name syntax works on ALL database dialects
with engine.begin() as conn:
    conn.execute(text("""
        INSERT INTO employees (id, name, department, salary, hire_date)
        VALUES (:id, :name, :dept, :salary, :date)
    """), {
        "id": 1, "name": "Alice", "dept": "Engineering",
        "salary": 90000, "date": "2022-01-15"
    })
    print("One row inserted")

# Insert multiple rows using a list of dicts
employee_data = [
    {"id": 2,  "name": "Bob",   "dept": "Marketing",   "salary": 65000, "date": "2023-03-01"},
    {"id": 3,  "name": "Carol", "dept": "Engineering", "salary": 85000, "date": "2021-06-01"},
    {"id": 4,  "name": "Dave",  "dept": "HR",          "salary": 55000, "date": "2020-11-15"},
    {"id": 5,  "name": "Eve",   "dept": "Engineering", "salary": 95000, "date": "2019-08-20"},
    {"id": 6,  "name": "Frank", "dept": "Marketing",   "salary": 70000, "date": "2022-07-10"},
    {"id": 7,  "name": "Grace", "dept": "Data",        "salary": 88000, "date": "2021-03-05"},
    {"id": 8,  "name": "Hank",  "dept": "HR",          "salary": 52000, "date": "2023-01-20"},
    {"id": 9,  "name": "Ivy",   "dept": "Data",        "salary": 92000, "date": "2020-09-30"},
    {"id": 10, "name": "Jack",  "dept": "Engineering", "salary": 78000, "date": "2022-12-01"},
]
 
with engine.begin() as conn:
    conn.execute(
        text("INSERT INTO employees (id, name, department, salary, hire_date) VALUES (:id, :name, :dept, :salary, :date)"),
        employee_data  # pass list of dicts for bulk insert
    )
    print(f"Inserted {len(employee_data)} rows")

# engine.begin() — auto-commits on exit, auto-rollbacks on exception
# Use for write operations (INSERT, UPDATE, DELETE, CREATE)
with engine.begin() as conn:
    conn.execute(text("UPDATE employees SET salary = salary * 1.05 WHERE department = 'Engineering'"))
    # commits automatically when the with block exits normally
    # rolls back automatically if an exception is raised
print("Engineering salaries updated by 5%")
 
# engine.connect() — does NOT auto-commit; requires explicit conn.commit()
# Use when you need fine-grained transaction control or for read-only queries
with engine.connect() as conn:
    result = conn.execute(text("SELECT COUNT(*) FROM employees"))
    print(f"Total employees: {result.scalar()}")
    # No commit needed for SELECT

# SELECT with named parameter — safe for user-provided values
with engine.connect() as conn:
    result = conn.execute(
        text("SELECT name, salary FROM employees WHERE department = :dept ORDER BY salary DESC"),
        {"dept": "Engineering"}
    )
    rows = result.fetchall()
    print("Engineering employees:")
    for name, salary in rows:
        print(f"  {name}: ${salary:,.0f}")

import pandas as pd
from sqlalchemy import create_engine, text
 
engine = create_engine("sqlite:///:memory:", echo=False)
 
# Setup: create and populate table
with engine.begin() as conn:
    conn.execute(text("CREATE TABLE employees (id INTEGER PRIMARY KEY, name TEXT, department TEXT, salary REAL, hire_date TEXT)"))
    conn.execute(text("INSERT INTO employees VALUES (:id, :name, :dept, :salary, :date)"), [
        {"id": 1, "name": "Alice",  "dept": "Engineering", "salary": 90000, "date": "2022-01-15"},
        {"id": 2, "name": "Bob",    "dept": "Marketing",   "salary": 65000, "date": "2023-03-01"},
        {"id": 3, "name": "Carol",  "dept": "Engineering", "salary": 85000, "date": "2021-06-01"},
        {"id": 4, "name": "Dave",   "dept": "HR",          "salary": 55000, "date": "2020-11-15"},
        {"id": 5, "name": "Eve",    "dept": "Engineering", "salary": 95000, "date": "2019-08-20"},
    ])
 
# pd.read_sql() with engine (SQL string form)
df = pd.read_sql("SELECT * FROM employees ORDER BY salary DESC", engine)
print("read_sql with engine:")
print(df)

# pd.read_sql() with text() + named params via engine.connect()
with engine.connect() as conn:
    df_eng = pd.read_sql(
        text("SELECT name, salary FROM employees WHERE department = :dept"),
        conn,
        params={"dept": "Engineering"}
    )
print("\nEngineering employees via text():")
print(df_eng)

import pandas as pd
from sqlalchemy import create_engine, Integer, Text, Float
 
engine = create_engine("sqlite:///:memory:", echo=False)
 
products = pd.DataFrame({
    "product_id": [101, 102, 103, 104, 105],
    "name":       ["Widget A", "Gadget Pro", "Tool Kit", "Smart Device", "Basic Pack"],
    "category":   ["Hardware", "Electronics", "Hardware", "Electronics", "Accessories"],
    "unit_price":  [29.99, 149.99, 49.99, 299.99, 9.99],
})
 
# to_sql() with SQLAlchemy engine — supports dtype parameter
products.to_sql(
    "products",
    engine,
    if_exists="replace",
    index=False,
    dtype={
        "product_id": Integer(),
        "name":       Text(),
        "category":   Text(),
        "unit_price":  Float(),
    }
)
print(f"Wrote {len(products)} products")
 
# Read back with a JOIN query using engine
orders = pd.DataFrame({
    "order_id":   [1, 2, 3, 4, 5],
    "product_id": [101, 102, 101, 103, 104],
    "quantity":   [2, 1, 3, 2, 1],
})
orders.to_sql("orders", engine, if_exists="replace", index=False)
 
# Full round-trip: to_sql then read_sql with JOIN
df_result = pd.read_sql("""
    SELECT
        o.order_id,
        p.name,
        p.category,
        o.quantity,
        ROUND(o.quantity * p.unit_price, 2) AS line_total
    FROM orders o
    JOIN products p ON o.product_id = p.product_id
    ORDER BY line_total DESC
""", engine)
print("\nOrder details:")
print(df_result)

# Read entire products table without writing SQL
df_products = pd.read_sql_table("products", engine)
print("\nAll products via read_sql_table:")
print(df_products)

import pandas as pd
from sqlalchemy import create_engine, text
 
engine = create_engine("sqlite:///:memory:", echo=False)
 
# Create schema and insert seed data using SQLAlchemy
with engine.begin() as conn:
    conn.execute(text("""
        CREATE TABLE products (
            id        INTEGER PRIMARY KEY,
            name      TEXT NOT NULL,
            category  TEXT,
            price     REAL,
            stock_qty INTEGER
        )
    """))
    conn.execute(text("INSERT INTO products VALUES (:id, :name, :cat, :price, :stock)"), [
        {"id": 1, "name": "Widget A",    "cat": "Hardware",     "price": 29.99, "stock": 150},
        {"id": 2, "name": "Gadget Pro",  "cat": "Electronics", "price": 149.99, "stock": 45},
        {"id": 3, "name": "Tool Kit",    "cat": "Hardware",     "price": 49.99, "stock": 30},
        {"id": 4, "name": "Smart Device","cat": "Electronics", "price": 299.99, "stock": 12},
        {"id": 5, "name": "Basic Pack",  "cat": "Accessories",  "price": 9.99, "stock": 200},
    ])
 
# Query using named parameter
category = "Electronics"
df = pd.read_sql(
    text("SELECT name, price, stock_qty FROM products WHERE category = :cat ORDER BY price DESC"),
    engine.connect(),
    params={"cat": category}
)
print(f"Electronics products:")
print(df)

import pandas as pd
from sqlalchemy import create_engine, text
 
def database_summary(engine) -> pd.DataFrame:
    """
    Returns a DataFrame with one row per table:
    table_name, row_count, column_count.
    Works on any SQLAlchemy engine.
    """
    # Get list of user tables
    with engine.connect() as conn:
        tables_result = conn.execute(
            text("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")
        )
        table_names = [row[0] for row in tables_result.fetchall()]
 
    # For each table, get row count and column count
    rows = []
    for table_name in table_names:
        with engine.connect() as conn:
            count = conn.execute(text(f"SELECT COUNT(*) FROM {table_name}")).scalar()
            # PRAGMA table_info returns one row per column
            col_info = conn.execute(text(f"PRAGMA table_info({table_name})")).fetchall()
            col_count = len(col_info)
        rows.append({"table_name": table_name, "row_count": count, "column_count": col_count})
 
    return pd.DataFrame(rows)
 
 
# Test with a multi-table database
engine = create_engine("sqlite:///:memory:", echo=False)
 
customers = pd.DataFrame({"id": [1, 2, 3], "name": ["Alice", "Bob", "Carol"]})
products  = pd.DataFrame({"id": [1, 2],    "name": ["Widget", "Gadget"], "price": [29.99, 149.99]})
orders    = pd.DataFrame({"id": [1, 2, 3, 4, 5], "customer_id": [1, 1, 2, 3, 1], "product_id": [1, 2, 1, 2, 1]})
 
customers.to_sql("customers", engine, if_exists="replace", index=False)
products.to_sql("products",   engine, if_exists="replace", index=False)
orders.to_sql("orders",       engine, if_exists="replace", index=False)
 
summary = database_summary(engine)
print("Database summary:")
print(summary)