Stored Procedures, Triggers & Functions

-- A function is an expression, so it is legal anywhere an expression is:
-- in the SELECT list (above), in WHERE, and in a JOIN condition.
SELECT order_id, calculate_order_total(order_id) AS total
FROM orders
WHERE calculate_order_total(order_id) > 100
ORDER BY order_id;

SELECT o.order_id, c.full_name
FROM orders o
JOIN customers c
  ON c.customer_id = o.customer_id
 AND calculate_order_total(o.order_id) > 100
ORDER BY o.order_id;

-- Legal is not the same as free. Ask what it costs:
EXPLAIN (ANALYZE, COSTS OFF, TIMING OFF)
SELECT order_id FROM orders WHERE calculate_order_total(order_id) > 100;

-- And the fix you will reach for does not work:
CREATE INDEX orders_total_idx ON orders (calculate_order_total(order_id));

import psycopg

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
cursor = conn.cursor()

# Call function directly
order_id = 1
cursor.execute("SELECT calculate_order_total(%s)", (order_id,))
total = cursor.fetchone()[0]
print(f"Order {order_id} total: ${total:.2f}")

# Use function in query
cursor.execute("""
    SELECT
        o.order_id,
        o.customer_id,
        calculate_order_total(o.order_id) as total_amount
    FROM orders o
    WHERE o.status = 'pending'
    ORDER BY total_amount DESC
    LIMIT 10
""")

print("\nTop 10 pending orders by value:")
for order_id, customer_id, total in cursor.fetchall():
    print(f"  Order {order_id} (Customer {customer_id}): ${total:.2f}")

cursor.close()
conn.close()

-- Function that returns table of customer purchase statistics
CREATE OR REPLACE FUNCTION get_customer_stats(
    customer_id_param INTEGER
)
RETURNS TABLE (
    total_orders INTEGER,
    total_spent DECIMAL(10, 2),
    avg_order_value DECIMAL(10, 2),
    last_order_date DATE
)
LANGUAGE plpgsql
AS $$
BEGIN
    RETURN QUERY
    SELECT
        COUNT(*)::INTEGER as total_orders,
        COALESCE(SUM(total_amount), 0) as total_spent,
        COALESCE(AVG(total_amount), 0) as avg_order_value,
        MAX(order_date)::DATE as last_order_date
    FROM orders
    WHERE customer_id = customer_id_param
        AND status = 'completed';
END;
$$;

-- Use like a regular table
SELECT * FROM get_customer_stats(1);

-- CAREFUL: the DECIMAL(10, 2) in the RETURNS TABLE list above is NOT
-- enforced. PostgreSQL uses those declarations for column names and
-- base types only; precision and scale are discarded. AVG here comes
-- back as 708.9066666666666667, not 708.91. Round explicitly inside
-- the function if the scale matters:
--     ROUND(COALESCE(AVG(total_amount), 0), 2)

-- Join with other tables
SELECT
    c.customer_id,
    c.email,
    s.total_orders,
    s.total_spent
FROM customers c
CROSS JOIN LATERAL get_customer_stats(c.customer_id) s
WHERE s.total_spent > 1000
ORDER BY s.total_spent DESC;

import psycopg

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
cursor = conn.cursor()

# Get stats for specific customer
customer_id = 1
cursor.execute("SELECT * FROM get_customer_stats(%s)", (customer_id,))

stats = cursor.fetchone()
if stats:
    total_orders, total_spent, avg_order, last_order = stats
    print(f"Customer {customer_id} Statistics:")
    print(f"  Total Orders: {total_orders}")
    print(f"  Total Spent: ${total_spent:.2f}")
    print(f"  Average Order: ${avg_order:.2f}")
    print(f"  Last Order: {last_order}")

# Get all high-value customers
cursor.execute("""
    SELECT
        c.customer_id,
        c.email,
        s.total_orders,
        s.total_spent
    FROM customers c
    CROSS JOIN LATERAL get_customer_stats(c.customer_id) s
    WHERE s.total_spent > 1000
    ORDER BY s.total_spent DESC
    LIMIT 10
""")

print("\nTop 10 High-Value Customers:")
for cust_id, email, orders, spent in cursor.fetchall():
    print(f"  {email}: {orders} orders, ${spent:.2f} spent")

cursor.close()
conn.close()

import psycopg

DSN = "host=localhost dbname=ecommerce user=demo password=demo"


def order_count(cur):
    cur.execute("SELECT count(*) FROM orders")
    return cur.fetchone()[0]


# --- Happy path -----------------------------------------------------------
with psycopg.connect(DSN) as conn, conn.cursor() as cur:
    cur.execute("CALL process_order(%s, %s, %s, NULL, NULL)",
                (1, [1, 2, 3], [2, 1, 3]))

    # psycopg 3 hands the procedure's OUT parameters back as an ordinary
    # result row, so there is no need to guess which order was just created
    # by re-querying and sorting.
    new_order_id, total = cur.fetchone()
    conn.commit()

print("Order created successfully!")
print(f"  Order ID: {new_order_id}")
print(f"  Total: ${total:.2f}")

# --- Failure path: ask for more stock than exists -------------------------
with psycopg.connect(DSN) as conn, conn.cursor() as cur:
    before = order_count(cur)
    try:
        cur.execute("CALL process_order(%s, %s, %s, NULL, NULL)",
                    (1, [1, 2], [1, 999999]))
    except psycopg.errors.RaiseException as e:
        conn.rollback()
        print(f"\nOrder failed: {str(e).strip().splitlines()[0]}")
    after = order_count(cur)

print(f"orders before failed call: {before}, after: {after}")

import json
import psycopg

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
cursor = conn.cursor()

items = [
    {"product_id": 101, "quantity": 2},
    {"product_id": 102, "quantity": 1}
]

# The function returns one row, so SELECT * FROM it and fetch once.
cursor.execute("SELECT * FROM create_order(%s, %s::jsonb)",
               (1, json.dumps(items)))

order_id, total, count = cursor.fetchone()
print(f"Created order {order_id}: ${total:.2f} ({count} items)")

conn.commit()
cursor.close()
conn.close()

import psycopg

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
cursor = conn.cursor()

# Insert a user (trigger fires automatically)
cursor.execute("""
    INSERT INTO users (email, username, status)
    VALUES ('alice@example.com', 'alice', 'active')
    RETURNING user_id
""")
user_id = cursor.fetchone()[0]
conn.commit()

print(f"Created user {user_id}")

# Update the user (trigger fires again)
cursor.execute("""
    UPDATE users
    SET status = 'inactive'
    WHERE user_id = %s
""", (user_id,))
conn.commit()

print(f"Updated user {user_id}")

# Check audit log
cursor.execute("""
    SELECT
        audit_id,
        action,
        old_data->>'status' as old_status,
        new_data->>'status' as new_status,
        changed_at
    FROM user_audit_log
    WHERE user_id = %s
    -- changed_at can tie when several changes share a transaction,
    -- so break the tie on the serial audit_id.
    ORDER BY changed_at, audit_id
""", (user_id,))

print(f"\nAudit trail for user {user_id}:")
for audit_id, action, old_status, new_status, changed_at in cursor.fetchall():
    if action == 'INSERT':
        print(f"  [{changed_at}] INSERT - New status: {new_status}")
    elif action == 'UPDATE':
        print(f"  [{changed_at}] UPDATE - Status: {old_status} → {new_status}")

cursor.close()
conn.close()

import psycopg

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
cursor = conn.cursor()

# Valid email (will be normalized to lowercase)
try:
    cursor.execute("""
        INSERT INTO users (email, username)
        VALUES ('Alice@Example.COM', 'alice')
        RETURNING email
    """)
    normalized_email = cursor.fetchone()[0]
    conn.commit()
    print(f"✓ Email normalized: {normalized_email}")
except psycopg.Error as e:
    print(f"✗ Error: {e}")
    conn.rollback()

# Invalid email format (will fail)
try:
    cursor.execute("""
        INSERT INTO users (email, username)
        VALUES ('not-an-email', 'bob')
    """)
    conn.commit()
except psycopg.Error as e:
    print(f"✗ Validation failed (expected): {e.diag.message_primary}")
    conn.rollback()

# Blocked domain (will fail)
try:
    cursor.execute("""
        INSERT INTO users (email, username)
        VALUES ('user@spam.com', 'charlie')
    """)
    conn.commit()
except psycopg.Error as e:
    print(f"✗ Domain blocked (expected): {e.diag.message_primary}")
    conn.rollback()

cursor.close()
conn.close()

import psycopg

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
cursor = conn.cursor()

# Create an order
cursor.execute("""
    INSERT INTO orders (customer_id, item_count, total_amount)
    VALUES (1, 0, 0)
    RETURNING order_id
""")
order_id = cursor.fetchone()[0]
conn.commit()

print(f"Created order {order_id}")

# Add items (trigger updates order totals automatically)
items = [
    (order_id, 101, 2, 49.99),  # 2 items at $49.99
    (order_id, 102, 1, 99.99),  # 1 item at $99.99
]

for order, product, qty, price in items:
    cursor.execute("""
        INSERT INTO order_items (order_id, product_id, quantity, unit_price)
        VALUES (%s, %s, %s, %s)
    """, (order, product, qty, price))

conn.commit()

# Check order totals (updated by trigger)
cursor.execute("""
    SELECT item_count, total_amount
    FROM orders
    WHERE order_id = %s
""", (order_id,))

item_count, total = cursor.fetchone()
print(f"\nOrder {order_id} totals (auto-updated by trigger):")
print(f"  Item count: {item_count}")
print(f"  Total amount: ${total:.2f}")

# Delete an item (trigger updates totals again)
cursor.execute("""
    DELETE FROM order_items
    WHERE order_id = %s AND product_id = 102
""", (order_id,))
conn.commit()

cursor.execute("""
    SELECT item_count, total_amount
    FROM orders
    WHERE order_id = %s
""", (order_id,))

item_count, total = cursor.fetchone()
print(f"\nAfter deleting one item:")
print(f"  Item count: {item_count}")
print(f"  Total amount: ${total:.2f}")

cursor.close()
conn.close()

import time
from decimal import Decimal

import psycopg

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
cursor = conn.cursor()

# Task: Update prices for 100,000 products with 10% increase

# APPROACH 1: Application Logic (SLOW)
start = time.time()

cursor.execute("SELECT product_id, price FROM products")
products = cursor.fetchall()  # Transfer 100K rows to Python

for product_id, price in products:
    # A NUMERIC column arrives as decimal.Decimal, and Decimal refuses to
    # mix with float: price * 1.10 raises TypeError. Multiply by another
    # Decimal, never by a float literal, or you silently reintroduce the
    # binary-floating-point error that NUMERIC exists to avoid.
    new_price = price * Decimal("1.10")
    cursor.execute("UPDATE products SET price = %s WHERE product_id = %s",
                   (new_price, product_id))

conn.commit()
app_time = time.time() - start
print(f"Application logic: {app_time:.2f} seconds")

# APPROACH 2: Database Logic (FAST)
start = time.time()

cursor.execute("UPDATE products SET price = price * 1.10")
conn.commit()

db_time = time.time() - start
print(f"Database logic: {db_time:.2f} seconds")

print(f"\nSpeedup: {app_time / db_time:.1f}x faster")

cursor.close()
conn.close()

import statistics
import time

import psycopg

conn = psycopg.connect("host=localhost dbname=ecommerce user=demo password=demo")
cursor = conn.cursor()

N = 10_000
REPS = 7


def with_trigger(on):
    cursor.execute("DROP TRIGGER IF EXISTS users_audit_trigger ON users")
    if on:
        cursor.execute("""
            CREATE TRIGGER users_audit_trigger
            AFTER INSERT ON users
            FOR EACH ROW EXECUTE FUNCTION audit_user_changes()
        """)
    conn.commit()


def reset():
    cursor.execute("TRUNCATE user_audit_log; DELETE FROM users")
    conn.commit()


def time_median(fn):
    times = []
    for _ in range(REPS):
        reset()
        start = time.perf_counter()
        fn()
        conn.commit()
        times.append((time.perf_counter() - start) * 1000)
    return statistics.median(times)


def one_statement_per_row():
    for i in range(N):
        cursor.execute("INSERT INTO users (email, username) VALUES (%s, %s)",
                       (f'user{i}@test.com', f'user{i}'))


def one_statement_total():
    cursor.execute("""
        INSERT INTO users (email, username)
        SELECT 'user' || g || '@test.com', 'user' || g
        FROM generate_series(1, %s) g
    """, (N,))


for label, work in (("10,000 round trips", one_statement_per_row),
                    ("1 INSERT ... SELECT", one_statement_total)):
    with_trigger(False)
    off = time_median(work)
    with_trigger(True)
    on = time_median(work)
    print(f"{label}")
    print(f"  without trigger: {off:8.1f} ms")
    print(f"  with trigger:    {on:8.1f} ms")
    print(f"  overhead:        {(on / off - 1) * 100:7.1f}%   "
          f"({(on - off) / N * 1000:.1f} us per row)")

cursor.close()
conn.close()