606 SQLite 嵌入式数据库使用¶

# Ubuntu 安装
sudo apt install sqlite3              # 安装 SQLite 命令行
sqlite3 --version                     # 查看版本

# macOS（自带）
sqlite3 --version                     # macOS 预装了 SQLite

# Python（自带，无需安装）
python3 -c "import sqlite3; print(sqlite3.sqlite_version)"  # 查看版本

# pip 安装增强版（可选）
pip install pysqlite3                 # Python 增强版 SQLite 绑定

# 白话：SQLite 数据库就是一个文件，创建数据库=创建文件
sqlite3 mydb.db                       # 打开/创建数据库文件
# 进入交互模式后就可以执行 SQL 了

# 常用 dot 命令（SQLite 特有的管理命令）
.help                                 # 查看所有命令
.databases                            # 列出已打开的数据库
.tables                               # 列出所有表
.schema users                         # 查看建表语句
.mode column                          # 以表格形式显示查询结果
.headers on                           # 显示列名
.quit                                 # 退出

-- 创建表
CREATE TABLE IF NOT EXISTS users (     -- IF NOT EXISTS 防止重复创建
    id INTEGER PRIMARY KEY AUTOINCREMENT,  -- 自增主键
    name TEXT NOT NULL,                -- 姓名，不允许为空
    email TEXT UNIQUE,                 -- 邮箱，唯一约束
    age INTEGER DEFAULT 0,             -- 年龄，默认值 0
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP  -- 创建时间，自动填当前时间
);

-- 插入数据
INSERT INTO users (name, email, age) VALUES ('张三', 'zhangsan@example.com', 25);
INSERT INTO users (name, email, age) VALUES ('李四', 'lisi@example.com', 30);

-- 查询数据
SELECT * FROM users;                   -- 查所有
SELECT name, age FROM users WHERE age > 25;  -- 条件查询
SELECT * FROM users ORDER BY age DESC LIMIT 10;  -- 按年龄降序，取前10条

-- 更新数据
UPDATE users SET age = 26 WHERE name = '张三';  -- 更新张三的年龄

-- 删除数据
DELETE FROM users WHERE id = 1;        -- 删除 id=1 的记录

import sqlite3  # Python 标准库自带，无需安装

# 连接数据库（文件不存在会自动创建）
conn = sqlite3.connect('myapp.db')     # 连接数据库文件
conn.row_factory = sqlite3.Row         # 让查询结果可以用列名访问
cursor = conn.cursor()                 # 创建游标

# 创建表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS tasks (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        title TEXT NOT NULL,
        done INTEGER DEFAULT 0,
        created_at DATETIME DEFAULT CURRENT_TIMESTAMP
    )
''')

# 插入数据（使用参数绑定防止 SQL 注入！）
cursor.execute(
    'INSERT INTO tasks (title) VALUES (?)',  # ? 是占位符
    ('学习 SQLite',)                         # 参数用元组传入
)

# 批量插入
tasks = [
    ('学习 Python',),
    ('学习 Redis',),
    ('写毕业论文',),
]
cursor.executemany(
    'INSERT INTO tasks (title) VALUES (?)',
    tasks                              # 批量数据
)

conn.commit()                          # 提交事务（重要！不提交数据不会保存）

# 查询数据
cursor.execute('SELECT * FROM tasks WHERE done = ?', (0,))
for row in cursor.fetchall():          # 遍历所有结果
    print(f"ID: {row['id']}, 标题: {row['title']}")  # 用列名访问

# 更新数据
cursor.execute('UPDATE tasks SET done = 1 WHERE id = ?', (1,))
conn.commit()

# 关闭连接
conn.close()                           # 用完记得关闭

import sqlite3

def get_db():
    """获取数据库连接的推荐写法"""
    conn = sqlite3.connect('myapp.db')
    conn.row_factory = sqlite3.Row
    conn.execute('PRAGMA foreign_keys = ON')  # 开启外键约束（默认关闭！）
    conn.execute('PRAGMA journal_mode = WAL')  # 开启 WAL 模式（提升并发性能）
    return conn

# 使用 with 语句自动管理事务
with get_db() as conn:                 # with 语句自动 commit 或 rollback
    conn.execute(
        'INSERT INTO tasks (title) VALUES (?)',
        ('新任务',)
    )
    # with 块结束时自动提交

-- PRAGMA 是 SQLite 特有的配置命令
-- 白话：调整 SQLite 的"内部开关"来优化性能

PRAGMA journal_mode = WAL;       -- 开启 WAL 模式（允许读写并发，性能大幅提升）
PRAGMA synchronous = NORMAL;     -- 同步模式（NORMAL 比 FULL 快，安全性足够）
PRAGMA cache_size = -64000;      -- 缓存 64MB（负数单位是 KB）
PRAGMA foreign_keys = ON;        -- 开启外键约束（SQLite 默认不检查外键！）
PRAGMA busy_timeout = 5000;      -- 写入锁等待 5 秒（避免 database is locked 错误）
PRAGMA temp_store = MEMORY;      -- 临时表存在内存里（更快）

-- SQLite 内置 JSON 函数，可以存储和查询 JSON 数据
CREATE TABLE configs (
    id INTEGER PRIMARY KEY,
    data TEXT                          -- JSON 数据存为 TEXT
);

INSERT INTO configs (data) VALUES 
('{"app": "myapp", "version": "1.0", "features": ["search", "export"]}');

-- 提取 JSON 字段
SELECT json_extract(data, '$.app') AS app_name FROM configs;  -- → "myapp"
SELECT json_extract(data, '$.features[0]') FROM configs;      -- → "search"

-- 修改 JSON
UPDATE configs 
SET data = json_set(data, '$.version', '2.0')  -- 修改 version 字段
WHERE id = 1;

-- FTS5 是 SQLite 的全文搜索扩展
-- 白话：可以像搜索引擎一样搜索文本内容

-- 创建全文搜索虚拟表
CREATE VIRTUAL TABLE articles_fts USING fts5(
    title,                             -- 标题
    content,                           -- 内容
    tokenize='unicode61'               -- 分词器
);

-- 插入数据
INSERT INTO articles_fts VALUES ('Python入门', 'Python是一种编程语言');
INSERT INTO articles_fts VALUES ('Redis缓存', 'Redis是内存数据库');

-- 全文搜索
SELECT * FROM articles_fts WHERE articles_fts MATCH 'Python';  -- 搜索包含 Python 的
SELECT * FROM articles_fts WHERE articles_fts MATCH 'Python OR Redis';  -- OR 查询
SELECT highlight(articles_fts, 0, '<b>', '</b>') FROM articles_fts  -- 高亮匹配词
WHERE articles_fts MATCH 'Python';

# 命令行快速导入 CSV 文件
sqlite3 mydb.db
.mode csv                              # 设置 CSV 模式
.import data.csv my_table              # 导入 CSV 到表（表不存在会自动创建）

# Python 导入 CSV
import csv
import sqlite3

conn = sqlite3.connect('mydb.db')
cursor = conn.cursor()

with open('data.csv', 'r') as f:
    reader = csv.DictReader(f)         # 用表头作为字段名
    for row in reader:
        cursor.execute(
            'INSERT INTO my_table (name, value) VALUES (?, ?)',
            (row['name'], row['value'])
        )

conn.commit()
conn.close()

sqlite3.OperationalError: database is locked

# 方法1：设置超时等待
conn = sqlite3.connect('mydb.db', timeout=10)  # 等待最多10秒

# 方法2：开启 WAL 模式（推荐）
conn.execute('PRAGMA journal_mode = WAL')  # 读写并发不冲突

外键约束未执行

SQLite 是动态类型，INTEGER 列也能存字符串

CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    age INTEGER NOT NULL
) STRICT;                              -- 严格模式，强制类型检查