AI 工作流 05：SQL、索引与 Redis 缓存

-- 创建用户表
CREATE TABLE users (
    id          SERIAL PRIMARY KEY,           -- 自增主键
    username    VARCHAR(50) UNIQUE NOT NULL,  -- 唯一非空
    email       VARCHAR(100) UNIQUE NOT NULL,
    age         INTEGER CHECK (age >= 0),     -- 带约束
    created_at  TIMESTAMP DEFAULT NOW()       -- 默认当前时间
);

-- 创建文章表（带外键关联 users）
CREATE TABLE articles (
    id          SERIAL PRIMARY KEY,
    title       VARCHAR(200) NOT NULL,
    content     TEXT,
    author_id   INTEGER REFERENCES users(id), -- 外键
    published   BOOLEAN DEFAULT FALSE,
    created_at  TIMESTAMP DEFAULT NOW()
);

-- 插入数据
INSERT INTO users (username, email, age) 
VALUES ('tom', 'tom@example.com', 25);

-- 批量插入
INSERT INTO users (username, email) 
VALUES ('jerry', 'jerry@example.com'), ('spike', 'spike@example.com');

-- 修改数据（务必带 WHERE，否则全表改！）
UPDATE users SET age = 26 WHERE username = 'tom';

-- 删除数据（务必带 WHERE！）
DELETE FROM users WHERE username = 'spike';

-- 基础查询
SELECT username, email FROM users;
SELECT * FROM users;                                    -- * 表示所有列
SELECT * FROM users WHERE age >= 18;                    -- 条件
SELECT * FROM users WHERE age >= 18 AND username LIKE 't%';  -- 模糊匹配

-- 排序 + 分页（分页必背）
SELECT * FROM users ORDER BY created_at DESC LIMIT 10 OFFSET 0;   -- 第1页
SELECT * FROM users ORDER BY created_at DESC LIMIT 10 OFFSET 10;  -- 第2页

-- 聚合统计
SELECT COUNT(*) FROM users;                             -- 总数
SELECT age, COUNT(*) FROM users GROUP BY age;           -- 按年龄分组统计
SELECT AVG(age), MAX(age), MIN(age) FROM users;         -- 平均/最大/最小

-- JOIN 连表查询（重点！）
-- 需求：查所有文章标题 + 作者名
SELECT articles.title, users.username 
FROM articles 
JOIN users ON articles.author_id = users.id;

-- LEFT JOIN：左表全保留，右表匹配不上的填 NULL
SELECT users.username, COUNT(articles.id) AS article_count
FROM users
LEFT JOIN articles ON users.id = articles.author_id
GROUP BY users.id, users.username;

users 表                articles 表
┌────┬─────────┐       ┌────┬─────────┬───────────┐
│ id │ username│       │ id │ title   │ author_id │
├────┼─────────┤       ├────┼─────────┼───────────┤
│  1 │ tom     │<──────┤  1 │ 文章A   │     1     │
│  2 │ jerry   │<──────┤  2 │ 文章B   │     1     │
│  3 │ spike   │       │  3 │ 文章C   │     2     │
└────┴─────────┘       └────┴─────────┴───────────┘

JOIN 后：
┌─────────┬────────┐
│ title   │ author │
├─────────┼────────┤
│ 文章A   │ tom    │
│ 文章B   │ tom    │
│ 文章C   │ jerry  │
└─────────┴────────┘
spike 没写文章，所以不在结果里（INNER JOIN）
用 LEFT JOIN 的话 spike 会出现，article_count = 0

docker run -d --name my-pg \
  -e POSTGRES_PASSWORD=123456 \
  -e POSTGRES_DB=myapp \
  -p 5432:5432 \
  postgres:16

pip install psycopg2-binary

import psycopg2

# 连接数据库
conn = psycopg2.connect(
    host="localhost",
    port=5432,
    dbname="myapp",
    user="postgres",
    password="123456",
)
cur = conn.cursor()

# 建表
cur.execute("""
    CREATE TABLE IF NOT EXISTS todos (
        id SERIAL PRIMARY KEY,
        text VARCHAR(200) NOT NULL,
        done BOOLEAN DEFAULT FALSE,
        created_at TIMESTAMP DEFAULT NOW()
    )
""")

# 插入
cur.execute("INSERT INTO todos (text) VALUES (%s) RETURNING id", ("学 SQL",))
new_id = cur.fetchone()[0]

# 查询
cur.execute("SELECT id, text, done FROM todos ORDER BY id")
rows = cur.fetchall()
for row in rows:
    print(row)

conn.commit()      # 提交事务
cur.close()
conn.close()

npm install pg

const { Pool } = require("pg");

const pool = new Pool({
    host: "localhost",
    port: 5432,
    database: "myapp",
    user: "postgres",
    password: "123456",
});

// 查询
async function getTodos() {
    const res = await pool.query("SELECT * FROM todos ORDER BY id");
    return res.rows;
}

// 插入
async function addTodo(text) {
    const res = await pool.query(
        "INSERT INTO todos (text) VALUES ($1) RETURNING *",
        [text]
    );
    return res.rows[0];
}

-- 用户表
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    username VARCHAR(50) UNIQUE NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    password_hash VARCHAR(200) NOT NULL,    -- 存哈希不存明文！
    created_at TIMESTAMP DEFAULT NOW()
);

-- 文章表
CREATE TABLE articles (
    id SERIAL PRIMARY KEY,
    title VARCHAR(200) NOT NULL,
    content TEXT,
    author_id INTEGER REFERENCES users(id) ON DELETE CASCADE,  -- 作者删了文章也删
    view_count INTEGER DEFAULT 0,
    created_at TIMESTAMP DEFAULT NOW(),
    updated_at TIMESTAMP DEFAULT NOW()
);

-- 标签表（多对多）
CREATE TABLE tags (
    id SERIAL PRIMARY KEY,
    name VARCHAR(50) UNIQUE NOT NULL
);

-- 文章-标签关联表（多对多关系靠中间表）
CREATE TABLE article_tags (
    article_id INTEGER REFERENCES articles(id) ON DELETE CASCADE,
    tag_id INTEGER REFERENCES tags(id) ON DELETE CASCADE,
    PRIMARY KEY (article_id, tag_id)        -- 联合主键
);

-- 没有 index，查 username='tom' 要扫全表
SELECT * FROM users WHERE username = 'tom';

-- 加索引后，瞬间定位
CREATE INDEX idx_users_username ON users(username);

-- 现在同样的查询快 100 倍

-- 复合索引
CREATE INDEX idx ON articles(author_id, created_at);

-- 这几个查询能用上索引吗？
WHERE author_id = 1                              -- 能（最左前缀）
WHERE author_id = 1 AND created_at > '2025-01-01' -- 能（完整匹配）
WHERE created_at > '2025-01-01'                   -- 不能（缺了 author_id）

-- 看执行计划
EXPLAIN SELECT * FROM users WHERE username = 'tom';

-- 输出（没索引时）：
-- Seq Scan on users  (cost=0.00..25.00 rows=1 width=...)
--   Filter: (username = 'tom'::text)
-- Seq Scan = 顺序全表扫描，慢！

-- 加索引后再 EXPLAIN：
-- Index Scan using idx_users_username on users  (cost=0.00..8.00 rows=1)
-- Index Scan = 用索引，快！

-- 加 ANALYZE 真的执行一次，看实际耗时
EXPLAIN ANALYZE SELECT * FROM users WHERE username = 'tom';

# Docker 一键启动
docker run -d --name my-redis -p 6379:6379 redis:7

# 命令行连进去玩
docker exec -it my-redis redis-cli

# 基础命令
SET name "tom"          # 存字符串
GET name                # 取
DEL name                # 删

SET count 0
INCR count              # 自增（原子操作，限流必备）
EXPIRE count 60         # 60秒后过期

# 哈希（存对象）
HSET user:1 name tom age 25
HGET user:1 name
HGETALL user:1

pip install redis

import redis
import json
import psycopg2

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

def get_user(user_id):
    # 1. 先查缓存
    cache_key = f"user:{user_id}"
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)   # 缓存命中，直接返回
    
    # 2. 缓存没有，查数据库
    conn = psycopg2.connect(...)
    cur = conn.cursor()
    cur.execute("SELECT id, username, email FROM users WHERE id = %s", (user_id,))
    row = cur.fetchone()
    cur.close()
    conn.close()
    
    if not row:
        return None
    
    user = {"id": row[0], "username": row[1], "email": row[2]}
    
    # 3. 写入缓存，设置过期时间（防止数据不一致太久）
    r.setex(cache_key, 300, json.dumps(user))   # 5 分钟过期
    
    return user

def check_rate_limit(user_id, max_calls=10, window=60):
    """每分钟最多调 max_calls 次"""
    key = f"rate:{user_id}"
    count = r.incr(key)
    if count == 1:
        r.expire(key, window)   # 第一次访问，设置窗口
    if count > max_calls:
        return False    # 超限
    return True