跳到主要内容

Python sqlite-utils 库完全使用指南

简介​

sqlite-utils 是由 Simon Willison 开发的一个强大的 Python 库，它提供了一套简洁优雅的 API 来操作 SQLite 数据库。相比原生的 sqlite3 模块，sqlite-utils 大大简化了常见的数据库操作，特别适合数据分析、ETL 任务和快速原型开发。

核心特性​

  • 简洁的 API 设计，减少样板代码
  • 自动创建表结构和索引
  • 支持批量插入和更新
  • 内置 CLI 工具，可直接在命令行操作数据库
  • 支持全文搜索（FTS）
  • 丰富的数据转换和查询功能

安装​

pip install sqlite-utils

基础操作​

1. 创建和连接数据库​

from sqlite_utils import Database

# 创建或连接到数据库文件
db = Database("mydata.db")

# 使用内存数据库（临时数据）
db = Database(memory=True)

2. 插入数据​

sqlite-utils 最大的优势之一就是插入数据时会自动创建表结构：

# 插入单条记录
db["users"].insert({
    "id": 1,
    "name": "张三",
    "email": "zhangsan@example.com",
    "age": 28
})

# 批量插入
users = [
    {"id": 2, "name": "李四", "email": "lisi@example.com", "age": 32},
    {"id": 3, "name": "王五", "email": "wangwu@example.com", "age": 25},
]
db["users"].insert_all(users)

3. 主键设置​

# 指定主键
db["users"].insert({
    "user_id": 1,
    "name": "张三"
}, pk="user_id")

# 复合主键
db["enrollments"].insert({
    "student_id": 1,
    "course_id": 101,
    "grade": "A"
}, pk=("student_id", "course_id"))

4. 查询数据​

# 获取所有记录
for user in db["users"].rows:
    print(user)

# 按条件查询
for user in db["users"].rows_where("age > ?", [25]):
    print(user["name"], user["age"])

# 使用字典参数
for user in db["users"].rows_where("age > :min_age", {"min_age": 25}):
    print(user)

# 获取单条记录
user = db["users"].get(1)  # 通过主键获取

5. 更新数据​

# 更新记录
db["users"].update(1, {"age": 29})

# 插入或更新（upsert）
db["users"].upsert({
    "id": 1,
    "name": "张三",
    "email": "new_email@example.com",
    "age": 30
}, pk="id")

# 批量 upsert
db["users"].upsert_all(users, pk="id")

6. 删除数据​

# 删除单条记录
db["users"].delete(1)

# 按条件删除
db["users"].delete_where("age < ?", [18])

# 删除整个表
db["users"].drop()

高级功能​

1. 表结构管理​

# 查看表结构
print(db["users"].columns)

# 添加列
db["users"].add_column("city", str)
db["users"].add_column("salary", float)

# 创建索引
db["users"].create_index(["email"])
db["users"].create_index(["name", "city"])

# 唯一索引
db["users"].create_index(["email"], unique=True)

2. 外键关系​

# 添加外键
db["posts"].insert({
    "id": 1,
    "title": "第一篇文章",
    "user_id": 1
})

db["posts"].add_foreign_key("user_id", "users", "id")

# 启用外键约束
db.enable_fkeys()

3. 全文搜索（FTS）​

# 启用全文搜索
db["articles"].enable_fts(["title", "content"], create_triggers=True)

# 搜索
results = db["articles"].search("Python 编程")
for article in results:
    print(article["title"])

4. 数据转换​

# 使用 transform 修改表结构
db["users"].transform(
    types={"age": int},  # 修改列类型
    rename={"email": "email_address"},  # 重命名列
    drop={"old_column"},  # 删除列
    pk="user_id"  # 更改主键
)

5. 聚合查询​

# 计数
count = db["users"].count

# 按条件计数
adult_count = db["users"].count_where("age >= 18")

# 使用原生 SQL
result = db.execute("SELECT AVG(age) as avg_age FROM users").fetchone()
print(f"平均年龄: {result['avg_age']}")

实用技巧​

1. 从 CSV/JSON 导入数据​

import csv
import json

# 从 CSV 导入
with open("data.csv") as f:
    reader = csv.DictReader(f)
    db["data"].insert_all(reader)

# 从 JSON 导入
with open("data.json") as f:
    data = json.load(f)
    db["data"].insert_all(data)

2. 导出数据​

import json

# 导出为字典列表
users = list(db["users"].rows)

# 导出为 JSON
with open("users.json", "w") as f:
    json.dump(users, f, ensure_ascii=False, indent=2)

3. 事务处理​

# 使用上下文管理器
with db.conn:
    db["users"].insert({"id": 10, "name": "用户10"})
    db["posts"].insert({"id": 100, "user_id": 10, "title": "文章100"})
# 自动提交或回滚

4. 查看表信息​

# 列出所有表
print(db.table_names())

# 检查表是否存在
if "users" in db.table_names():
    print("users 表存在")

# 查看表的行数
print(f"用户数: {db['users'].count}")

5. 处理大数据集​

# 使用生成器节省内存
def generate_records():
    for i in range(1000000):
        yield {"id": i, "value": f"record_{i}"}

# 分批插入
db["large_table"].insert_all(generate_records(), batch_size=1000)

CLI 命令行工具​

sqlite-utils 还提供了强大的命令行工具：

# 插入数据
sqlite-utils insert mydata.db users data.json

# 查询数据
sqlite-utils query mydata.db "SELECT * FROM users WHERE age > 25"

# 导出为 CSV
sqlite-utils query mydata.db "SELECT * FROM users" --csv > users.csv

# 创建索引
sqlite-utils create-index mydata.db users email

# 启用全文搜索
sqlite-utils enable-fts mydata.db articles title content

最佳实践​

  1. 使用批量操作：使用 insert_all() 而不是循环调用 insert()，性能提升显著。

  2. 合理设置主键：明确指定主键可以避免潜在问题，特别是在更新和查询时。

  3. 利用索引：对经常查询的字段创建索引，提升查询性能。

  4. 使用 upsert：当不确定记录是否存在时，使用 upsert() 避免重复插入错误。

  5. 启用外键约束：在需要数据完整性的场景中，使用 db.enable_fkeys() 启用外键检查。

实战示例：构建简单的博客数据库​

from sqlite_utils import Database
from datetime import datetime

db = Database("blog.db")

# 创建用户
db["users"].insert_all([
    {"id": 1, "username": "alice", "email": "alice@example.com"},
    {"id": 2, "username": "bob", "email": "bob@example.com"}
], pk="id")

# 创建文章
db["posts"].insert_all([
    {
        "id": 1,
        "title": "Python 入门",
        "content": "这是一篇关于 Python 的文章...",
        "user_id": 1,
        "created_at": datetime.now().isoformat()
    },
    {
        "id": 2,
        "title": "SQLite 技巧",
        "content": "SQLite 是一个轻量级数据库...",
        "user_id": 2,
        "created_at": datetime.now().isoformat()
    }
], pk="id")

# 添加外键
db["posts"].add_foreign_key("user_id", "users", "id")

# 启用全文搜索
db["posts"].enable_fts(["title", "content"])

# 创建索引
db["posts"].create_index(["user_id"])
db["posts"].create_index(["created_at"])

# 查询某用户的所有文章
for post in db["posts"].rows_where("user_id = ?", [1]):
    print(f"{post['title']} - {post['created_at']}")

# 全文搜索
for post in db["posts"].search("Python"):
    print(f"找到: {post['title']}")

总结​

sqlite-utils 是一个优秀的 SQLite 操作库，它通过简洁的 API 大大降低了数据库操作的复杂度。无论是数据分析、快速原型开发，还是构建小型应用，sqlite-utils 都是一个值得推荐的工具。它的自动化特性（如自动创建表、自动推断类型）让开发者可以更专注于业务逻辑，而不是繁琐的数据库管理工作。