Python SQLite3 使用教程

本教程将带你从零掌握 Python 内置 sqlite3 模块的核心用法——从无配置的快速上手，到事务、行工厂这类高级特性的灵活应用，最后还有完整可运行的示例和练习题巩固。

1. SQLite 简介

SQLite 是目前全球部署最广泛的嵌入式数据库引擎，它没有传统数据库的“服务端-客户端”架构，直接把整个数据库系统（包括定义、数据、索引）压缩进单一跨平台文件，是轻量级项目的首选存储方案。

它的核心优势：

  • 🚫 零服务器/零配置：无需安装额外软件，创建文件即可用
  • 🌐 跨全平台：支持从手机端到服务器的所有主流系统
  • 📝 兼容标准 SQL：学习成本极低，语法可复用
  • 🐍 Python 原生内置：Python 2.5+ 直接引入 sqlite3 模块

2. 前置核心概念

正式动手前，先理清两个必用的核心对象和通用操作流程：

2.1 核心对象

对象名作用
Connection代表与单一数据库文件的物理连接，负责提交/回滚事务、关闭连接
Cursor依附于 Connection，用于执行 SQL 语句和获取查询结果（相当于 SQL 的“控制台”）

2.2 通用操作流程

所有 SQLite 操作都遵循这 6 步（但可以用上下文管理器简化）：

  1. 建立数据库文件连接
  2. 从连接获取游标对象
  3. 通过游标执行 SQL
  4. 如果是写操作（增删改），提交事务
  5. 关闭游标（可选但推荐）
  6. 关闭连接（必须，防止资源泄漏）

3. 基础 CRUD 操作

我们以「用户管理」为例，一步步演示核心功能。

3.1 连接数据库

import sqlite3

# 连接到 example.db 文件（不存在则自动在当前目录创建）
conn = sqlite3.connect('example.db')

如果只是临时测试，不想生成持久化文件，可以用 :memory: 参数创建内存数据库：

# 程序结束后自动清空的内存数据库
conn = sqlite3.connect(':memory:')

3.2 创建表

# 第一步从连接获取游标
cursor = conn.cursor()

# 创建 users 表（IF NOT EXISTS 避免重复创建报错）
cursor.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
''')

# 建表是写操作，必须提交
conn.commit()

3.3 插入数据

⚠️ 敲黑板！绝对不要用字符串拼接参数！ 用 ? 占位符+元组的参数化查询，防止 SQL 注入。

单条插入

# 用 ? 占位参数，第二个参数必须是元组（哪怕只有一个）
cursor.execute(
    "INSERT INTO users (name, email) VALUES (?, ?)", 
    ('Alice', 'alice@example.com')
)
conn.commit()

批量插入

用 executemany() 方法，比循环调用 execute() 快几十倍：

# 准备批量数据（必须是元组/列表的可迭代对象）
new_users = [
    ('Bob', 'bob@example.com'),
    ('Charlie', 'charlie@example.com'),
    ('Diana', 'diana@example.com')
]

# 执行批量插入
cursor.executemany(
    "INSERT INTO users (name, email) VALUES (?, ?)", 
    new_users
)
conn.commit()

3.4 查询数据

查询有 3 种获取结果的方式，按需选择：

全量获取：fetchall()

cursor.execute("SELECT * FROM users")
all_users = cursor.fetchall()  # 返回一个元组的列表

# 遍历打印
for user in all_users:
    print(user)

单条获取：fetchone()

适合查询唯一结果的场景（比如按 ID/唯一键查）：

cursor.execute("SELECT name, email FROM users WHERE id = ?", (1,))
alice = cursor.fetchone()  # 返回一个元组，没有结果返回 None
print(alice)

分批获取：fetchmany(n)

适合大数据量查询，避免一次性加载内存溢出：

cursor.execute("SELECT * FROM users")
while True:
    batch = cursor.fetchmany(2)  # 每次取 2 条
    if not batch:
        break
    for user in batch:
        print(user)

3.5 更新数据

cursor.execute(
    "UPDATE users SET email = ? WHERE name = ?", 
    ('alice.updated@example.com', 'Alice')
)
conn.commit()

3.6 删除数据

cursor.execute("DELETE FROM users WHERE name = ?", ('Bob',))
conn.commit()

4. 提升开发效率的高级特性

4.1 用上下文管理器自动管理连接/事务

Python 3.10+ 开始，Connection 本身支持 with 语句：

  • 成功退出 with 块时自动提交事务
  • 抛出异常时自动回滚事务
  • 块结束后自动关闭连接
with sqlite3.connect('example.db') as conn:
    # 这里可以直接创建/使用游标
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users")
    print(cursor.fetchall())

4.2 用行工厂返回字典（而非默认元组）

默认 fetchall()/fetchone() 返回的是不带字段名的元组，用 row_factory 可以改成更易读的字典：

自定义字典工厂

def dict_factory(cursor, row):
    return {col[0]: row[idx] for idx, col in enumerate(cursor.description)}

# 给连接设置行工厂
conn.row_factory = dict_factory
cursor = conn.cursor()

cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
alice_dict = cursor.fetchone()
print(alice_dict["email"])  # 直接用字段名访问

用内置的 sqlite3.Row

如果不想自己写工厂，Python 内置了更轻量的 sqlite3.Row：

conn.row_factory = sqlite3.Row
cursor = conn.cursor()

cursor.execute("SELECT * FROM users WHERE id = ?", (1,))
alice_row = cursor.fetchone()
print(alice_row["email"])  # 支持字段名访问
print(tuple(alice_row))    # 也支持转成元组

4.3 显式控制事务

虽然 with 语句已经自动处理了，但复杂场景下显式控制更灵活：

try:
    # 执行多个写操作
    cursor.execute("UPDATE users SET stock = stock - 1 WHERE id = ?", (1,))
    cursor.execute("INSERT INTO orders (user_id, product_id) VALUES (?, ?)", (1, 1))
    
    # 全部成功后提交
    conn.commit()
except sqlite3.Error as e:
    # 任意失败回滚
    print(f"操作失败，已回滚：{e}")
    conn.rollback()

5. 最佳实践清单

  1. ✅ 永远用参数化查询：不要碰字符串拼接 SQL
  2. ✅ 用 with 语句：简化连接/事务管理，避免资源泄漏
  3. ✅ 批量操作优先 executemany()：性能提升显著
  4. ✅ 合理使用内存数据库：临时测试/缓存用 :memory:
  5. ✅ 捕获 sqlite3.Error：处理可能的 SQL 语法/约束错误
  6. ✅ 复杂应用考虑 ORM：比如 SQLAlchemy，把表映射成 Python 类

6. 完整可运行示例：简易商品库存系统

import sqlite3
from contextlib import closing  # 可选，但更规范地管理游标

def init_inventory_db():
    """初始化商品库存表"""
    with sqlite3.connect('inventory.db') as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute('''
                CREATE TABLE IF NOT EXISTS products (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    name TEXT NOT NULL UNIQUE,
                    price REAL NOT NULL CHECK (price > 0),
                    stock INTEGER DEFAULT 0 CHECK (stock >= 0)
                )
            ''')

def add_product(name: str, price: float, stock: int = 0) -> int | None:
    """添加商品，返回商品ID，失败返回None"""
    try:
        with sqlite3.connect('inventory.db') as conn:
            with closing(conn.cursor()) as cursor:
                cursor.execute(
                    "INSERT INTO products (name, price, stock) VALUES (?, ?, ?)",
                    (name, price, stock)
                )
                return cursor.lastrowid  # 获取刚插入的自增ID
    except sqlite3.IntegrityError as e:
        print(f"商品名重复或价格/库存非法：{e}")
        return None
    except sqlite3.Error as e:
        print(f"数据库异常：{e}")
        return None

def get_products_by_min_price(min_price: float = 0) -> list[dict]:
    """查询价格≥min_price的商品，返回字典列表，按价格升序"""
    try:
        with sqlite3.connect('inventory.db') as conn:
            conn.row_factory = sqlite3.Row
            with closing(conn.cursor()) as cursor:
                cursor.execute(
                    "SELECT * FROM products WHERE price >= ? ORDER BY price ASC",
                    (min_price,)
                )
                return [dict(row) for row in cursor.fetchall()]
    except sqlite3.Error as e:
        print(f"查询失败：{e}")
        return []

# 主程序示例
if __name__ == '__main__':
    init_inventory_db()
    
    # 添加商品
    add_product("机械键盘", 299.99, 50)
    add_product("无线鼠标", 89.99, 120)
    add_product("游戏耳机", 499.99, 30)
    
    # 查询价格≥200的商品
    premium_products = get_products_by_min_price(200)
    print("高端外设商品：")
    for p in premium_products:
        print(f"- {p['name']}: ¥{p['price']:.2f} (库存: {p['stock']})")

7. 配套练习与解决方案

练习：查询指定分数区间的学生

要求：实现 get_score_in(low, high) 函数，返回分数在 [low, high] 之间的学生姓名，按分数从低到高排序。

完整解决方案

import os
import sqlite3
from contextlib import closing

def get_score_in(low: int, high: int, db_path: str = 'test_scores.db') -> list[str]:
    with sqlite3.connect(db_path) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute(
                "SELECT name FROM user WHERE score BETWEEN ? AND ? ORDER BY score ASC",
                (low, high)
            )
            return [row[0] for row in cursor.fetchall()]

# 测试代码
if __name__ == '__main__':
    TEST_DB = 'test_scores.db'
    
    # 清理旧测试数据库
    if os.path.exists(TEST_DB):
        os.remove(TEST_DB)
    
    # 初始化测试数据
    with sqlite3.connect(TEST_DB) as conn:
        with closing(conn.cursor()) as cursor:
            cursor.execute('''
                CREATE TABLE user (
                    id VARCHAR(20) PRIMARY KEY,
                    name VARCHAR(20) NOT NULL,
                    score INT NOT NULL CHECK (score BETWEEN 0 AND 100)
                )
            ''')
            cursor.executemany(
                "INSERT INTO user VALUES (?, ?, ?)",
                [
                    ('A-001', 'Adam', 95),
                    ('A-002', 'Bart', 62),
                    ('A-003', 'Lisa', 78)
                ]
            )
    
    # 执行测试
    assert get_score_in(80, 95) == ['Adam'], "80-95分测试失败"
    assert get_score_in(60, 80) == ['Bart', 'Lisa'], "60-80分测试失败"
    assert get_score_in(60, 100) == ['Bart', 'Lisa', 'Adam'], "60-100分测试失败"
    print("🎉 所有测试通过！")