SQLite3 完全指南：从基础到进阶的实战教程

1# 安装 SQLite3 命令行工具
2sudo apt-get update
3sudo apt-get install sqlite3
4
5# 安装开发库（用于编程）
6sudo apt-get install libsqlite3-dev
7
8# 验证安装
9sqlite3 --version

1# macOS 自带 SQLite3
2sqlite3 --version
3
4# 如需更新版本，使用 Homebrew
5brew install sqlite

1# 验证安装
2sqlite3 --version

1# 检查 SQLite3 版本
2import sqlite3
3print(f"SQLite version: {sqlite3.sqlite_version}")
4print(f"Python sqlite3 module version: {sqlite3.version}")

1# 创建新数据库（如果不存在）
2sqlite3 my_database.db
3
4# 创建临时数据库（内存数据库）
5sqlite3 :memory:

 1import sqlite3
 2
 3# 连接到数据库（如果不存在会自动创建）
 4conn = sqlite3.connect('my_database.db')
 5
 6# 创建内存数据库
 7# conn = sqlite3.connect(':memory:')
 8
 9# 创建游标对象
10cursor = conn.cursor()
11
12# 关闭连接
13conn.close()

 1# 进入 SQLite 命令行
 2sqlite3 my_database.db
 3
 4# 查看数据库信息
 5.databases              # 列出所有数据库
 6.tables                 # 列出所有表
 7.schema                 # 显示所有表的创建语句
 8.schema table_name      # 显示指定表的创建语句
 9
10# 导入/导出数据
11.dump                    # 导出数据库为 SQL 脚本
12.read backup.sql         # 导入 SQL 脚本
13.mode list              # 设置输出模式（list, line, column, html, csv 等）
14.headers on             # 显示列名
15.output output.csv      # 将输出重定向到文件
16.import data.csv table_name  # 导入 CSV 文件
17
18# 其他有用命令
19.quit                   # 退出
20.help                   # 显示帮助
21.timer on               # 显示执行时间

 1-- 创建各种数据类型的表
 2CREATE TABLE products (
 3    id INTEGER PRIMARY KEY,
 4    name TEXT NOT NULL,
 5    price REAL DEFAULT 0.0,
 6    stock INTEGER DEFAULT 0,
 7    description TEXT,
 8    image BLOB,
 9    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
10);
11
12-- 插入数据
13INSERT INTO products (name, price, stock, description)
14VALUES ('iPhone 15', 7999.99, 100, '最新款智能手机');
15
16-- 插入不同类型的数据（SQLite 会自动转换）
17INSERT INTO products (id, name, price) VALUES (2, '数据转换', '100.50');
18-- 价格 '100.50' 会被转换为 REAL 类型

 1-- 查看列的类型信息
 2PRAGMA table_info(products);
 3
 4-- 输出示例：
 5-- cid | name        | type    | notnull | dflt_value | pk
 6-- ----|-------------|---------|---------|------------|---
 7-- 0   | id          | INTEGER | 0       | NULL       | 1
 8-- 1   | name        | TEXT    | 1       | NULL       | 0
 9-- 2   | price       | REAL    | 0       | 0.0        | 0
10-- 3   | stock       | INTEGER | 0       | 0          | 0
11-- 4   | description | TEXT    | 0       | NULL       | 0

 1-- 基本表创建
 2CREATE TABLE users (
 3    id INTEGER PRIMARY KEY AUTOINCREMENT,
 4    username TEXT NOT NULL UNIQUE,
 5    email TEXT NOT NULL,
 6    age INTEGER CHECK (age >= 0 AND age <= 150),
 7    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
 8);
 9
10-- 带外键约束的表
11CREATE TABLE orders (
12    id INTEGER PRIMARY KEY AUTOINCREMENT,
13    user_id INTEGER NOT NULL,
14    order_no TEXT NOT NULL UNIQUE,
15    total_amount REAL NOT NULL,
16    status TEXT DEFAULT 'pending',
17    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
18    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
19);
20
21-- 创建带索引的表
22CREATE TABLE articles (
23    id INTEGER PRIMARY KEY AUTOINCREMENT,
24    title TEXT NOT NULL,
25    content TEXT,
26    author TEXT,
27    views INTEGER DEFAULT 0,
28    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
29);
30
31-- 创建索引
32CREATE INDEX idx_articles_author ON articles(author);
33CREATE INDEX idx_articles_created ON articles(created_at DESC);

 1-- 添加列
 2ALTER TABLE users ADD COLUMN avatar BLOB;
 3ALTER TABLE users ADD COLUMN phone TEXT;
 4
 5-- 重命名表
 6ALTER TABLE old_table_name RENAME TO new_table_name;
 7
 8-- 注意：SQLite 不支持直接删除列或修改列类型
 9-- 需要通过重建表的方式实现：
10-- 1. 创建新表（包含需要的列）
11-- 2. 将数据从旧表复制到新表
12-- 3. 删除旧表
13-- 4. 重命名新表

1-- 删除表
2DROP TABLE users;
3
4-- 删除表（如果存在）
5DROP TABLE IF EXISTS users;

 1-- 单条插入
 2INSERT INTO users (username, email, age)
 3VALUES ('alice', 'alice@example.com', 25);
 4
 5-- 多条插入
 6INSERT INTO users (username, email, age) VALUES
 7    ('bob', 'bob@example.com', 30),
 8    ('charlie', 'charlie@example.com', 28),
 9    ('diana', 'diana@example.com', 32);
10
11-- 插入指定列
12INSERT INTO users (username, email) VALUES ('eve', 'eve@example.com');
13
14-- 使用默认值
15INSERT INTO users (username, email) VALUES ('frank', 'frank@example.com');
16-- age 和 created_at 会使用默认值
17
18-- 插入或更新（UPSERT）
19INSERT INTO users (username, email, age)
20VALUES ('alice', 'new_email@example.com', 26)
21ON CONFLICT(username) DO UPDATE SET
22    email = excluded.email,
23    age = excluded.age;

 1-- 基本查询
 2SELECT * FROM users;
 3
 4-- 查询指定列
 5SELECT username, email FROM users;
 6
 7-- 条件查询
 8SELECT * FROM users WHERE age > 25;
 9
10-- 多条件查询
11SELECT * FROM users WHERE age >= 25 AND age <= 35;
12
13-- 范围查询
14SELECT * FROM users WHERE age BETWEEN 25 AND 35;
15
16-- IN 查询
17SELECT * FROM users WHERE username IN ('alice', 'bob', 'charlie');
18
19-- 模糊查询
20SELECT * FROM users WHERE username LIKE 'a%';  -- 以 a 开头
21SELECT * FROM users WHERE email LIKE '%@example.com';
22
23-- 排序
24SELECT * FROM users ORDER BY age ASC;
25SELECT * FROM users ORDER BY age DESC, username ASC;
26
27-- 限制结果数量
28SELECT * FROM users LIMIT 5;
29SELECT * FROM users LIMIT 5 OFFSET 2;  -- 分页
30
31-- 去重
32SELECT DISTINCT age FROM users;
33
34-- 聚合查询
35SELECT COUNT(*) FROM users;
36SELECT COUNT(DISTINCT age) FROM users;
37SELECT AVG(age) FROM users;
38SELECT MAX(age), MIN(age) FROM users;
39SELECT SUM(views) FROM articles;
40
41-- 分组查询
42SELECT age, COUNT(*) as count
43FROM users
44GROUP BY age
45HAVING count > 1;

1-- 更新单条记录
2UPDATE users SET email = 'new_alice@example.com' WHERE username = 'alice';
3
4-- 更新多条记录
5UPDATE users SET age = age + 1 WHERE age < 30;
6
7-- 批量更新
8UPDATE articles SET views = views + 1 WHERE id = 1;

 1-- 删除单条记录
 2DELETE FROM users WHERE username = 'eve';
 3
 4-- 删除多条记录
 5DELETE FROM users WHERE age < 25;
 6
 7-- 删除所有数据（保留表结构）
 8DELETE FROM users;
 9
10-- 清空表并重置自增 ID
11DELETE FROM users;
12DELETE FROM sqlite_sequence WHERE name = 'users';
13
14-- 更高效的方式（SQLite 3.7.17+）
15TRUNCATE TABLE users;

 1-- 内连接
 2SELECT u.username, o.order_no, o.total_amount
 3FROM users u
 4INNER JOIN orders o ON u.id = o.user_id;
 5
 6-- 左连接
 7SELECT u.username, o.order_no
 8FROM users u
 9LEFT JOIN orders o ON u.id = o.user_id;
10
11-- 多表连接
12SELECT u.username, o.order_no, o.total_amount
13FROM users u
14JOIN orders o ON u.id = o.user_id
15JOIN order_items oi ON o.id = oi.order_id
16JOIN products p ON oi.product_id = p.id;

 1-- 标量子查询
 2SELECT username,
 3       (SELECT COUNT(*) FROM orders WHERE user_id = users.id) as order_count
 4FROM users;
 5
 6-- IN 子查询
 7SELECT * FROM users
 8WHERE id IN (SELECT DISTINCT user_id FROM orders WHERE total_amount > 1000);
 9
10-- EXISTS 子查询
11SELECT * FROM users u
12WHERE EXISTS (
13    SELECT 1 FROM orders o WHERE o.user_id = u.id AND o.status = 'completed'
14);
15
16-- FROM 子查询
17SELECT * FROM (
18    SELECT username, age,
19           RANK() OVER (ORDER BY age DESC) as age_rank
20    FROM users
21) ranked_users
22WHERE age_rank <= 10;

 1-- 排名函数
 2SELECT username, age,
 3       ROW_NUMBER() OVER (ORDER BY age DESC) as row_num,
 4       RANK() OVER (ORDER BY age DESC) as rank,
 5       DENSE_RANK() OVER (ORDER BY age DESC) as dense_rank
 6FROM users;
 7
 8-- 分组排名
 9SELECT username, age,
10       RANK() OVER (PARTITION BY age DIV 10 ORDER BY username) as group_rank
11FROM users;
12
13-- 累计求和
14SELECT order_no, total_amount,
15       SUM(total_amount) OVER (ORDER BY created_at) as running_total
16FROM orders;
17
18-- 移动平均
19SELECT order_no, total_amount,
20       AVG(total_amount) OVER (
21           ORDER BY created_at
22           ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
23       ) as moving_avg
24FROM orders;

 1import sqlite3
 2
 3conn = sqlite3.connect('my_database.db')
 4cursor = conn.cursor()
 5
 6try:
 7    # 开始事务
 8    conn.execute("BEGIN TRANSACTION")
 9
10    # 执行多个操作
11    cursor.execute("INSERT INTO users (username, email) VALUES (?, ?)", ('test', 'test@example.com'))
12    cursor.execute("UPDATE users SET age = age + 1 WHERE username = 'alice'")
13
14    # 提交事务
15    conn.commit()
16    print("事务提交成功")
17
18except Exception as e:
19    # 回滚事务
20    conn.rollback()
21    print(f"事务失败，已回滚: {e}")
22
23finally:
24    conn.close()

 1-- 查看表的索引
 2PRAGMA index_list(users);
 3
 4-- 查看索引详细信息
 5PRAGMA index_info(idx_articles_author);
 6
 7-- 分析索引使用情况
 8EXPLAIN QUERY PLAN SELECT * FROM users WHERE username = 'alice';
 9
10-- 创建复合索引
11CREATE INDEX idx_users_age_email ON users(age, email);
12
13-- 创建唯一索引
14CREATE UNIQUE INDEX idx_users_username ON users(username);
15
16-- 部分索引（只索引满足条件的行）
17CREATE INDEX idx_active_orders ON orders(user_id) WHERE status = 'active';
18
19-- 删除索引
20DROP INDEX IF EXISTS idx_articles_author;

 1-- 使用 EXPLAIN 分析查询计划
 2EXPLAIN SELECT * FROM users WHERE username = 'alice';
 3
 4-- 使用 EXPLAIN QUERY PLAN 获取更可读的输出
 5EXPLAIN QUERY PLAN SELECT * FROM users WHERE username = 'alice';
 6
 7-- 避免使用 SELECT *
 8SELECT id, username FROM users;  -- ✅ 推荐
 9SELECT * FROM users;              -- ❌ 不推荐
10
11-- 使用 LIMIT 限制结果数量
12SELECT * FROM users LIMIT 1000;
13
14-- 使用 UNION ALL 代替 UNION（如果不需要去重）
15SELECT username FROM users WHERE age > 25
16UNION ALL
17SELECT username FROM users WHERE age < 20;
18
19-- 使用 EXISTS 代替 IN（子查询）
20SELECT * FROM users u
21WHERE EXISTS (SELECT 1 FROM orders o WHERE o.user_id = u.id);

 1import sqlite3
 2
 3conn = sqlite3.connect('my_database.db')
 4
 5# 性能优化设置
 6
 7# 设置同步模式（牺牲数据安全性换取性能）
 8# NORMAL: 每1000毫秒同步一次
 9# OFF: 不同步（最不安全，最快）
10conn.execute("PRAGMA synchronous = NORMAL")
11
12# 设置日记模式
13# WAL: 写前日志（Write-Ahead Logging），支持并发读写
14conn.execute("PRAGMA journal_mode = WAL")
15
16# 设置缓存大小（KB）
17conn.execute("PRAGMA cache_size = -64000")  # 64MB
18
19# 设置临时存储位置（内存）
20conn.execute("PRAGMA temp_store = MEMORY")
21
22# 设置页面大小（字节）
23conn.execute("PRAGMA page_size = 4096")
24
25# 启用内存映射 I/O
26conn.execute("PRAGMA mmap_size = 268435456")  # 256MB
27
28# 设置繁忙超时（毫秒）
29conn.execute("PRAGMA busy_timeout = 5000")
30
31# 查看设置
32cursor = conn.cursor()
33cursor.execute("PRAGMA cache_size")
34print(f"Cache size: {cursor.fetchone()[0]}")

 1import sqlite3
 2
 3def insert_users_batch(users_data):
 4    conn = sqlite3.connect('my_database.db')
 5    cursor = conn.cursor()
 6
 7    # 开始事务
 8    conn.execute("BEGIN TRANSACTION")
 9
10    try:
11        # 使用 executemany 批量插入
12        cursor.executemany(
13            "INSERT INTO users (username, email, age) VALUES (?, ?, ?)",
14            users_data
15        )
16        conn.commit()
17        print(f"成功插入 {len(users_data)} 条记录")
18
19    except Exception as e:
20        conn.rollback()
21        print(f"插入失败: {e}")
22
23    finally:
24        conn.close()
25
26# 示例数据
27users = [
28    ('user1', 'user1@example.com', 25),
29    ('user2', 'user2@example.com', 30),
30    ('user3', 'user3@example.com', 28),
31    # ... 更多数据
32]
33
34insert_users_batch(users)

1# 方法1：使用 .dump 命令
2sqlite3 my_database.db .dump > backup.sql
3
4# 方法2：直接复制文件（最简单）
5cp my_database.db my_database_backup.db
6
7# 方法3：使用 Python 备份

 1import sqlite3
 2
 3def backup_database(source_db, backup_db):
 4    # 连接源数据库
 5    source = sqlite3.connect(source_db)
 6    # 创建备份
 7    backup = sqlite3.connect(backup_db)
 8
 9    # 执行备份
10    with backup:
11        source.backup(backup)
12
13    source.close()
14    backup.close()
15    print(f"备份成功: {backup_db}")
16
17backup_database('my_database.db', 'backup.db')

1# 从 SQL 备份恢复
2sqlite3 my_database.db < backup.sql
3
4# 从文件备份恢复
5cp my_database_backup.db my_database.db

 1import sqlite3
 2
 3def restore_database(backup_db, target_db):
 4    # 连接目标数据库
 5    target = sqlite3.connect(target_db)
 6    # 连接备份
 7    backup = sqlite3.connect(backup_db)
 8
 9    # 执行恢复
10    with target:
11        backup.backup(target)
12
13    target.close()
14    backup.close()
15    print(f"恢复成功: {target_db}")
16
17restore_database('backup.db', 'my_database.db')

1import sqlite3
2
3# 启用 WAL 模式以支持并发读写
4conn = sqlite3.connect('my_database.db')
5conn.execute("PRAGMA journal_mode = WAL")
6
7# 设置繁忙超时
8conn.execute("PRAGMA busy_timeout = 5000")

 1import sqlite3
 2
 3def try_recover_database(db_path):
 4    try:
 5        conn = sqlite3.connect(db_path)
 6        conn.execute("PRAGMA integrity_check")
 7        result = conn.fetchone()[0]
 8        if result == "ok":
 9            print("数据库完整")
10        else:
11            print(f"数据库可能损坏: {result}")
12        conn.close()
13    except sqlite3.DatabaseError as e:
14        print(f"数据库错误: {e}")
15
16try_recover_database('my_database.db')

 1# 使用 pysqlcipher3（需要单独安装）
 2# pip install pysqlcipher3
 3"""
 4from pysqlcipher3 import dbapi2 as sqlite
 5
 6# 创建加密数据库
 7conn = sqlite.connect('encrypted.db')
 8conn.execute("PRAGMA key = 'your_encryption_key'")
 9conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
10conn.commit()
11conn.close()
12
13# 打开加密数据库
14conn = sqlite.connect('encrypted.db')
15conn.execute("PRAGMA key = 'your_encryption_key'")
16"""

 1-- 创建适当的索引
 2CREATE INDEX idx_composite ON table_name(column1, column2);
 3
 4-- 使用部分索引
 5CREATE INDEX idx_active ON orders(user_id) WHERE status = 'active';
 6
 7-- 使用覆盖索引
 8CREATE INDEX idx_covering ON users(username, email, age);
 9
10-- 分析查询计划
11EXPLAIN QUERY PLAN SELECT * FROM users WHERE username = 'alice';
12
13-- 使用 LIMIT
14SELECT * FROM large_table LIMIT 1000;
15
16-- 使用分页
17SELECT * FROM large_table LIMIT 100 OFFSET 200;

 1# ❌ 不安全：SQL 注入风险
 2username = input("输入用户名: ")
 3cursor.execute(f"SELECT * FROM users WHERE username = '{username}'")
 4
 5# ✅ 安全：使用参数化查询
 6username = input("输入用户名: ")
 7cursor.execute("SELECT * FROM users WHERE username = ?", (username,))
 8
 9# ✅ 多个参数
10cursor.execute(
11    "INSERT INTO users (username, email, age) VALUES (?, ?, ?)",
12    ('alice', 'alice@example.com', 25)
13)

1# ✅ 使用 with 语句自动关闭连接
2with sqlite3.connect('my_database.db') as conn:
3    cursor = conn.cursor()
4    cursor.execute("SELECT * FROM users")
5    rows = cursor.fetchall()
6    # 连接会自动关闭

 1import sqlite3
 2
 3try:
 4    conn = sqlite3.connect('my_database.db')
 5    cursor = conn.cursor()
 6
 7    cursor.execute("INSERT INTO users (username, email) VALUES (?, ?)", ('test', 'test@example.com'))
 8    conn.commit()
 9
10except sqlite3.IntegrityError as e:
11    print(f"数据完整性错误: {e}")
12except sqlite3.OperationalError as e:
13    print(f"操作错误: {e}")
14except sqlite3.Error as e:
15    print(f"数据库错误: {e}")
16finally:
17    if conn:
18        conn.close()

1-- 分析表以优化查询计划
2ANALYZE;
3
4-- 重建数据库以减少碎片
5VACUUM;
6
7-- 清理临时数据
8PRAGMA incremental_vacuum;

 1import sqlite3
 2import time
 3
 4def execute_with_timing(conn, query, params=None):
 5    start_time = time.time()
 6    cursor = conn.cursor()
 7    if params:
 8        cursor.execute(query, params)
 9    else:
10        cursor.execute(query)
11    elapsed = time.time() - start_time
12    print(f"查询耗时: {elapsed:.4f} 秒")
13    return cursor
14
15# 使用示例
16conn = sqlite3.connect('my_database.db')
17cursor = execute_with_timing(conn, "SELECT * FROM users WHERE age > ?", (25,))
18results = cursor.fetchall()
19conn.close()

 1import sqlite3
 2from datetime import datetime
 3
 4class BlogDatabase:
 5    def __init__(self, db_path='blog.db'):
 6        self.conn = sqlite3.connect(db_path)
 7        self.create_tables()
 8
 9    def create_tables(self):
10        """创建博客相关表"""
11        cursor = self.conn.cursor()
12
13        cursor.execute("""
14            CREATE TABLE IF NOT EXISTS categories (
15                id INTEGER PRIMARY KEY AUTOINCREMENT,
16                name TEXT NOT NULL UNIQUE,
17                slug TEXT NOT NULL UNIQUE
18            )
19        """)
20
21        cursor.execute("""
22            CREATE TABLE IF NOT EXISTS posts (
23                id INTEGER PRIMARY KEY AUTOINCREMENT,
24                title TEXT NOT NULL,
25                slug TEXT NOT NULL UNIQUE,
26                content TEXT NOT NULL,
27                excerpt TEXT,
28                category_id INTEGER,
29                status TEXT DEFAULT 'draft',
30                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
31                updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
32                FOREIGN KEY (category_id) REFERENCES categories(id)
33            )
34        """)
35
36        cursor.execute("""
37            CREATE INDEX IF NOT EXISTS idx_posts_status
38            ON posts(status, created_at DESC)
39        """)
40
41        cursor.execute("""
42            CREATE INDEX IF NOT EXISTS idx_posts_category
43            ON posts(category_id)
44        """)
45
46        self.conn.commit()
47
48    def create_post(self, title, content, category_id=None, status='draft'):
49        """创建新文章"""
50        cursor = self.conn.cursor()
51        slug = title.lower().replace(' ', '-')
52        cursor.execute("""
53            INSERT INTO posts (title, slug, content, category_id, status)
54            VALUES (?, ?, ?, ?, ?)
55        """, (title, slug, content, category_id, status))
56        self.conn.commit()
57        return cursor.lastrowid
58
59    def get_published_posts(self, limit=10):
60        """获取已发布的文章"""
61        cursor = self.conn.cursor()
62        cursor.execute("""
63            SELECT p.*, c.name as category_name
64            FROM posts p
65            LEFT JOIN categories c ON p.category_id = c.id
66            WHERE p.status = 'published'
67            ORDER BY p.created_at DESC
68            LIMIT ?
69        """, (limit,))
70        return cursor.fetchall()
71
72    def get_post_by_slug(self, slug):
73        """根据 slug 获取文章"""
74        cursor = self.conn.cursor()
75        cursor.execute("""
76            SELECT p.*, c.name as category_name
77            FROM posts p
78            LEFT JOIN categories c ON p.category_id = c.id
79            WHERE p.slug = ?
80        """, (slug,))
81        return cursor.fetchone()
82
83    def close(self):
84        """关闭数据库连接"""
85        self.conn.close()
86
87# 使用示例
88blog = BlogDatabase()
89blog.create_post(
90    title="我的第一篇文章",
91    content="这是文章内容...",
92    category_id=1,
93    status='published'
94)
95posts = blog.get_published_posts()
96blog.close()

  1import sqlite3
  2import hashlib
  3import secrets
  4
  5class AuthSystem:
  6    def __init__(self, db_path='auth.db'):
  7        self.conn = sqlite3.connect(db_path)
  8        self.create_tables()
  9
 10    def create_tables(self):
 11        """创建认证相关表"""
 12        cursor = self.conn.cursor()
 13
 14        cursor.execute("""
 15            CREATE TABLE IF NOT EXISTS users (
 16                id INTEGER PRIMARY KEY AUTOINCREMENT,
 17                username TEXT NOT NULL UNIQUE,
 18                email TEXT NOT NULL UNIQUE,
 19                password_hash TEXT NOT NULL,
 20                salt TEXT NOT NULL,
 21                is_active INTEGER DEFAULT 1,
 22                created_at DATETIME DEFAULT CURRENT_TIMESTAMP
 23            )
 24        """)
 25
 26        cursor.execute("""
 27            CREATE TABLE IF NOT EXISTS sessions (
 28                id INTEGER PRIMARY KEY AUTOINCREMENT,
 29                user_id INTEGER NOT NULL,
 30                token TEXT NOT NULL UNIQUE,
 31                expires_at DATETIME NOT NULL,
 32                created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
 33                FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE
 34            )
 35        """)
 36
 37        cursor.execute("""
 38            CREATE INDEX IF NOT EXISTS idx_sessions_token
 39            ON sessions(token)
 40        """)
 41
 42        self.conn.commit()
 43
 44    def _generate_salt(self):
 45        """生成随机 salt"""
 46        return secrets.token_hex(16)
 47
 48    def _hash_password(self, password, salt):
 49        """密码哈希"""
 50        return hashlib.sha256((password + salt).encode()).hexdigest()
 51
 52    def register_user(self, username, email, password):
 53        """注册用户"""
 54        cursor = self.conn.cursor()
 55        salt = self._generate_salt()
 56        password_hash = self._hash_password(password, salt)
 57
 58        try:
 59            cursor.execute("""
 60                INSERT INTO users (username, email, password_hash, salt)
 61                VALUES (?, ?, ?, ?)
 62            """, (username, email, password_hash, salt))
 63            self.conn.commit()
 64            return True
 65        except sqlite3.IntegrityError:
 66            return False
 67
 68    def verify_user(self, username, password):
 69        """验证用户"""
 70        cursor = self.conn.cursor()
 71        cursor.execute("""
 72            SELECT id, password_hash, salt, is_active
 73            FROM users
 74            WHERE username = ?
 75        """, (username,))
 76        user = cursor.fetchone()
 77
 78        if not user:
 79            return False
 80
 81        user_id, stored_hash, salt, is_active = user
 82
 83        if not is_active:
 84            return False
 85
 86        password_hash = self._hash_password(password, salt)
 87        return secrets.compare_digest(password_hash, stored_hash)
 88
 89    def create_session(self, user_id, expires_hours=24):
 90        """创建会话"""
 91        cursor = self.conn.cursor()
 92        token = secrets.token_urlsafe(32)
 93        from datetime import timedelta
 94        expires_at = datetime.now() + timedelta(hours=expires_hours)
 95
 96        cursor.execute("""
 97            INSERT INTO sessions (user_id, token, expires_at)
 98            VALUES (?, ?, ?)
 99        """, (user_id, token, expires_at))
100        self.conn.commit()
101        return token
102
103    def verify_session(self, token):
104        """验证会话"""
105        cursor = self.conn.cursor()
106        cursor.execute("""
107            SELECT s.user_id, s.expires_at
108            FROM sessions s
109            JOIN users u ON s.user_id = u.id
110            WHERE s.token = ? AND u.is_active = 1 AND s.expires_at > datetime('now')
111        """, (token,))
112        return cursor.fetchone()
113
114    def close(self):
115        """关闭数据库连接"""
116        self.conn.close()
117
118# 使用示例
119auth = AuthSystem()
120auth.register_user('alice', 'alice@example.com', 'secure_password')
121if auth.verify_user('alice', 'secure_password'):
122    print("登录成功！")
123auth.close()