CREATE TABLE users (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    email       TEXT NOT NULL UNIQUE,
    name        TEXT NOT NULL,
    status      TEXT NOT NULL DEFAULT 'active'
                CHECK (status IN ('active', 'inactive', 'banned')),
    metadata    JSONB DEFAULT '{}',
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- Auto-update updated_at
CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at = NOW();
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_users_updated_at
    BEFORE UPDATE ON users
    FOR EACH ROW EXECUTE FUNCTION update_updated_at();

-- One-to-Many
CREATE TABLE orders (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    user_id     BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    total       DECIMAL(10,2) NOT NULL,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX ix_orders_user_id ON orders(user_id);

-- Many-to-Many
CREATE TABLE user_roles (
    user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    role_id BIGINT NOT NULL REFERENCES roles(id) ON DELETE CASCADE,
    PRIMARY KEY (user_id, role_id)
);

-- Query JSONB fields
SELECT * FROM users WHERE metadata->>'plan' = 'premium';
SELECT * FROM users WHERE metadata @> '{"verified": true}';

-- Update JSONB
UPDATE users SET metadata = metadata || '{"verified": true}' WHERE id = 1;

-- Index JSONB for fast queries
CREATE INDEX ix_users_metadata ON users USING GIN (metadata);

WITH active_users AS (
    SELECT id, name, email
    FROM users
    WHERE status = 'active'
),
user_order_counts AS (
    SELECT user_id, COUNT(*) as order_count
    FROM orders
    GROUP BY user_id
)
SELECT au.name, au.email, COALESCE(uoc.order_count, 0) as orders
FROM active_users au
LEFT JOIN user_order_counts uoc ON au.id = uoc.user_id
ORDER BY orders DESC;

-- Rank users by order count
SELECT
    name,
    COUNT(o.id) as order_count,
    RANK() OVER (ORDER BY COUNT(o.id) DESC) as rank
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
GROUP BY u.id, u.name;

import sqlite3
from contextlib import contextmanager

@contextmanager
def get_db(path: str = "app.db"):
    conn = sqlite3.connect(path)
    conn.row_factory = sqlite3.Row  # Dict-like access
    conn.execute("PRAGMA journal_mode=WAL")  # Better concurrency
    conn.execute("PRAGMA foreign_keys=ON")   # Enforce FK constraints
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Usage
with get_db() as db:
    db.execute(
        "INSERT INTO users (name, email) VALUES (?, ?)",
        (name, email),
    )

import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

# String (cache)
r.setex("user:123:profile", 3600, json.dumps(profile))  # 1hr TTL
cached = json.loads(r.get("user:123:profile") or "{}")

# Hash (object fields)
r.hset("user:123", mapping={"name": "Alice", "email": "[email protected]"})
name = r.hget("user:123", "name")

# Sorted Set (leaderboard)
r.zadd("leaderboard", {"alice": 100, "bob": 85})
top_10 = r.zrevrange("leaderboard", 0, 9, withscores=True)

# List (queue)
r.lpush("task_queue", json.dumps(task))
task = json.loads(r.brpop("task_queue", timeout=30)[1])

def get_user(user_id: str) -> dict:
    # Check cache first
    cached = r.get(f"user:{user_id}")
    if cached:
        return json.loads(cached)

    # Cache miss - fetch from DB
    user = db.query("SELECT * FROM users WHERE id = %s", (user_id,))

    # Cache for 1 hour
    r.setex(f"user:{user_id}", 3600, json.dumps(user))
    return user

def invalidate_user(user_id: str):
    r.delete(f"user:{user_id}")

# Initialize
alembic init migrations

# Create migration
alembic revision --autogenerate -m "add users table"

# Run migrations
alembic upgrade head

# Rollback one step
alembic downgrade -1

-- Single column (equality lookups)
CREATE INDEX ix_users_email ON users(email);

-- Composite (multi-column queries, order matters)
CREATE INDEX ix_orders_user_status ON orders(user_id, status);

-- Partial (subset of rows)
CREATE INDEX ix_users_active ON users(email) WHERE status = 'active';

-- Covering (avoid table lookup)
CREATE INDEX ix_orders_covering ON orders(user_id) INCLUDE (total, status);

EXPLAIN ANALYZE
SELECT u.name, COUNT(o.id)
FROM users u
JOIN orders o ON u.id = o.user_id
WHERE u.status = 'active'
GROUP BY u.id;

# SQLAlchemy
from sqlalchemy import create_engine

engine = create_engine(
    "postgresql://user:pass@localhost/db",
    pool_size=10,          # Maintain 10 connections
    max_overflow=20,       # Allow 20 additional under load
    pool_timeout=30,       # Wait 30s for connection
    pool_recycle=1800,     # Recycle connections every 30min
)