SQLite 完整教程

# 安装（macOS）
brew install sqlite3

# 安装（Ubuntu/Debian）
sudo apt-get install sqlite3

# 创建/打开数据库文件
sqlite3 mydb.db

# 进入交互式 shell 后的常用命令
.help         -- 查看帮助
.tables       -- 显示所有表
.schema       -- 显示表结构
.mode column  -- 列格式输出
.headers on   -- 显示列名
.quit         -- 退出

-- 创建用户表（完整约束示例）
CREATE TABLE IF NOT EXISTS users (
    id         INTEGER  PRIMARY KEY AUTOINCREMENT, -- 自增主键
    username   TEXT     NOT NULL UNIQUE,            -- 非空且唯一
    email      TEXT     NOT NULL,
    age        INTEGER  CHECK(age >= 0),             -- 检查约束
    score      REAL     DEFAULT 0.0,               -- 默认值
    created_at TEXT     DEFAULT (datetime('now'))   -- 默认当前时间
);

-- 创建订单表（外键示例）
CREATE TABLE IF NOT EXISTS orders (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id   INTEGER NOT NULL,
    amount    REAL    NOT NULL,
    status    TEXT    DEFAULT 'pending',
    FOREIGN KEY (user_id) REFERENCES users(id)  -- 外键
);

-- 插入单条记录
INSERT INTO users (username, email, age, score)
VALUES ('alice', 'alice@example.com', 25, 98.5);

-- 批量插入多条（性能更好）
INSERT INTO users (username, email, age, score) VALUES
    ('bob',     'bob@example.com',   30, 88.0),
    ('charlie', 'charlie@example.com', 22, 72.3),
    ('diana',   'diana@example.com',  28, 95.1);

-- INSERT OR REPLACE：存在则替换，不存在则插入
INSERT OR REPLACE INTO users (username, email) 
VALUES ('alice', 'newalice@example.com');

-- 查询所有列
SELECT * FROM users;

-- 查询指定列
SELECT username, age, score FROM users;

-- WHERE 条件过滤
SELECT * FROM users WHERE age > 25;
SELECT * FROM users WHERE username LIKE 'a%';  -- 模糊匹配

-- ORDER BY 排序
SELECT * FROM users ORDER BY score DESC;     -- 降序
SELECT * FROM users ORDER BY age ASC;       -- 升序

-- LIMIT / OFFSET 分页
SELECT * FROM users LIMIT 10 OFFSET 20;       -- 第3页，每页10条

-- UPDATE：更新指定记录（务必加 WHERE！）
UPDATE users
SET    score = 99.0, age = 26
WHERE  username = 'alice';

-- DELETE：删除指定记录（务必加 WHERE！）
DELETE FROM users WHERE id = 3;

-- 清空整张表（不可回滚）
DELETE FROM users;  -- ← 危险！没有 WHERE 就是清空！

-- 基础聚合
SELECT
    COUNT(*)      AS total_users,    -- 总行数
    AVG(age)      AS avg_age,        -- 平均年龄
    MAX(score)    AS max_score,      -- 最高分
    MIN(score)    AS min_score,      -- 最低分
    SUM(score)    AS total_score     -- 总分
FROM users;

-- GROUP BY：按年龄段分组统计
SELECT
    age,
    COUNT(*)    AS count,
    AVG(score) AS avg_score
FROM  users
GROUP BY age
ORDER BY age;

-- HAVING：过滤分组（平均分 > 80 的年龄组）
SELECT age, AVG(score) AS avg_score
FROM   users
GROUP BY age
HAVING AVG(score) > 80
ORDER BY avg_score DESC;

-- INNER JOIN：查有订单的用户及订单信息
SELECT
    u.username,
    u.email,
    o.amount,
    o.status
FROM       users  u
INNER JOIN orders o ON u.id = o.user_id;

-- LEFT JOIN：查所有用户，即使没有订单也显示
SELECT
    u.username,
    COUNT(o.id)  AS order_count,    -- 订单数（无订单为0）
    SUM(o.amount) AS total_spent    -- 总消费（无订单为NULL）
FROM      users  u
LEFT JOIN orders o ON u.id = o.user_id
GROUP BY  u.id, u.username
ORDER BY  total_spent DESC NULLS LAST;

-- ① WHERE 中：查分数高于平均分的用户
SELECT username, score
FROM   users
WHERE  score > (SELECT AVG(score) FROM users);

-- ② IN 子查询：查有订单的用户
SELECT * FROM users
WHERE id IN (
    SELECT DISTINCT user_id FROM orders
);

-- ③ FROM 子查询（派生表）：按分数分级统计
SELECT grade, COUNT(*) AS count
FROM (
    SELECT
        username,
        CASE
            WHEN score >= 90 THEN 'A'
            WHEN score >= 80 THEN 'B'
            WHEN score >= 60 THEN 'C'
            ELSE 'D'
        END AS grade
    FROM users
) AS graded
GROUP BY grade;

-- ① 手动事务
BEGIN TRANSACTION;  -- 或 BEGIN;

    UPDATE accounts SET balance = balance - 100 WHERE id = 1; -- A 扣款
    UPDATE accounts SET balance = balance + 100 WHERE id = 2; -- B 到账

COMMIT;   -- 提交，两步都成功则永久保存

-- ② 回滚示例（遇到错误时）
BEGIN TRANSACTION;
    UPDATE accounts SET balance = balance - 100 WHERE id = 1;
    -- 如果 B 不存在，或余额不足...
ROLLBACK; -- 回滚，A 的扣款也撤销，保持一致性

import sqlite3

conn = sqlite3.connect('mydb.db')
try:
    cursor = conn.cursor()
    cursor.execute("UPDATE accounts SET balance=balance-100 WHERE id=1")
    cursor.execute("UPDATE accounts SET balance=balance+100 WHERE id=2")
    conn.commit()   # ✅ 两步都成功，提交
    print("转账成功")
except Exception as e:
    conn.rollback()  # ❌ 出错，回滚
    print(f"转账失败，已回滚: {e}")
finally:
    conn.close()

-- 创建普通索引
CREATE INDEX idx_users_age ON users(age);

-- 创建唯一索引（同时保证唯一性约束）
CREATE UNIQUE INDEX idx_users_email ON users(email);

-- 创建复合索引（联合多列，注意列顺序！）
CREATE INDEX idx_orders_user_status ON orders(user_id, status);

-- 查看查询是否用到索引（执行计划）
EXPLAIN QUERY PLAN
SELECT * FROM users WHERE age = 25;
-- 输出: SEARCH TABLE users USING INDEX idx_users_age (age=?)

-- 删除索引
DROP INDEX IF EXISTS idx_users_age;

-- 用 CTE 分步计算：找出消费前3名的用户详情
WITH user_spending AS (
    -- Step 1: 计算每个用户的总消费
    SELECT
        user_id,
        SUM(amount) AS total_spent,
        COUNT(*)    AS order_count
    FROM  orders
    GROUP BY user_id
),
top_spenders AS (
    -- Step 2: 取消费前3名
    SELECT * FROM user_spending
    ORDER BY total_spent DESC
    LIMIT 3
)
-- Step 3: 关联用户表取完整信息
SELECT
    u.username,
    u.email,
    ts.total_spent,
    ts.order_count
FROM       top_spenders ts
INNER JOIN users u ON u.id = ts.user_id;

-- 场景：组织架构表，查某人下面所有子级
-- 表结构：employees(id, name, manager_id)

WITH RECURSIVE subordinates AS (
    -- 锚点：起始节点（CEO，id=1）
    SELECT id, name, manager_id, 0 AS level
    FROM   employees
    WHERE  id = 1

    UNION ALL

    -- 递归：找所有下属
    SELECT e.id, e.name, e.manager_id, s.level + 1
    FROM   employees    e
    JOIN   subordinates s ON e.manager_id = s.id
)
SELECT printf('%s', substr('            ', 1, level*2) || name) AS org_tree
FROM  subordinates;

-- ROW_NUMBER: 按分数排名（组内排名）
SELECT
    username,
    score,
    ROW_NUMBER() OVER (ORDER BY score DESC) AS rank,
    RANK()       OVER (ORDER BY score DESC) AS rank_with_gap
-- RANK 遇到相同分数会跳号（1,2,2,4），ROW_NUMBER 不跳
FROM users;

-- PARTITION BY: 按年龄段分组后各自排名
SELECT
    username,
    age,
    score,
    DENSE_RANK() OVER (
        PARTITION BY (age / 10) -- 按10岁一组分区
        ORDER BY score DESC
    ) AS group_rank
FROM users;

-- 滑动窗口：累计平均 & 前后值对比
SELECT
    id,
    amount,
    AVG(amount) OVER (
        ORDER BY id
        ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
    ) AS moving_avg_3,           -- 3期移动平均
    LAG(amount, 1) OVER (ORDER BY id) AS prev_amount,  -- 上一期
    LEAD(amount, 1) OVER (ORDER BY id) AS next_amount  -- 下一期
FROM orders;

-- 表结构（JSON 存在 TEXT 列）
CREATE TABLE products (
    id    INTEGER PRIMARY KEY,
    name  TEXT,
    attrs TEXT  -- 存 JSON
);

INSERT INTO products VALUES (
    1, 'iPhone 16',
    '{"color":"black","storage":256,"tags":["phone","apple"]}'
);

-- json_extract: 提取 JSON 字段值
SELECT
    name,
    json_extract(attrs, '$.color')   AS color,
    json_extract(attrs, '$.storage') AS storage,
    json_extract(attrs, '$.tags[0]') AS first_tag
FROM products;

-- 按 JSON 字段过滤
SELECT * FROM products
WHERE json_extract(attrs, '$.storage') >= 128;

-- json_set: 更新 JSON 字段
UPDATE products
SET   attrs = json_set(attrs, '$.color', 'white')
WHERE id = 1;

-- json_array_length: 获取数组长度
SELECT json_array_length(json_extract(attrs, '$.tags')) FROM products;

-- 创建 FTS5 虚拟表
CREATE VIRTUAL TABLE articles_fts USING fts5(
    title,
    body,
    content='articles',  -- 内容表（可选）
    tokenize='unicode61' -- Unicode 分词
);

-- 插入并建索引
INSERT INTO articles_fts SELECT title, body FROM articles;

-- 全文搜索（比 LIKE 快 100x）
SELECT * FROM articles_fts
WHERE articles_fts MATCH 'sqlite AND database';  -- AND 运算

-- 短语搜索（引号包裹）
SELECT * FROM articles_fts
WHERE articles_fts MATCH '"full text search"';

-- 前缀搜索
SELECT * FROM articles_fts
WHERE articles_fts MATCH 'data*';  -- 匹配 data, database, datastore...

-- 按相关度排序（bm25 算法）
SELECT *, bm25(articles_fts) AS relevance
FROM  articles_fts
WHERE articles_fts MATCH 'sqlite'
ORDER BY relevance;

-- ✅ WAL 模式：写时允许并发读（最重要！）
PRAGMA journal_mode = WAL;

-- ✅ 同步模式：NORMAL 安全且快（比 FULL 快 3x）
PRAGMA synchronous = NORMAL;

-- ✅ 缓存大小：内存页数（负数=KB，正数=页数）
PRAGMA cache_size = -64000;  -- 64MB 缓存

-- ✅ mmap 内存映射：大文件读取加速
PRAGMA mmap_size = 268435456; -- 256MB

-- ✅ 外键约束开关（默认关闭！）
PRAGMA foreign_keys = ON;

-- ✅ temp_store：临时表存在内存
PRAGMA temp_store = MEMORY;

-- 查看当前配置
PRAGMA journal_mode;
PRAGMA page_size;
PRAGMA cache_size;

import sqlite3

def get_connection(db_path: str) -> sqlite3.Connection:
    """生产级 SQLite 连接配置"""
    conn = sqlite3.connect(
        db_path,
        timeout=30,                           # 等锁超时30秒
        isolation_level=None,                # 手动控制事务
        check_same_thread=False,             # 多线程可用
    )
    conn.row_factory = sqlite3.Row          # 结果可按列名访问
    conn.execute("PRAGMA journal_mode = WAL")
    conn.execute("PRAGMA synchronous = NORMAL")
    conn.execute("PRAGMA foreign_keys = ON")
    conn.execute("PRAGMA cache_size = -64000")
    return conn

# 使用示例：按列名访问结果
conn = get_connection('app.db')
cursor = conn.cursor()
cursor.execute("SELECT * FROM users WHERE id=?", (1,))
user = cursor.fetchone()
print(user['username'])  # 按列名访问