跳到主要内容

sqlite_utils 详细使用指南

sqlite_utils 是一个强大的 Python 库,用于简化 SQLite 数据库的操作。它提供了比标准库 sqlite3 更简洁、更 Pythonic 的 API。

安装​

pip install sqlite-utils

核心功能​

1. 基本数据库操作​

from sqlite_utils import Database

# 创建/连接数据库
db = Database("my_database.db")

# 内存数据库
db = Database(memory=True)

2. 插入数据​

# 插入单条记录
db["dogs"].insert({
    "name": "Cleo",
    "age": 4,
    "breed": "Labrador"
})

# 插入多条记录
db["dogs"].insert_all([
    {"name": "Max", "age": 2, "breed": "Poodle"},
    {"name": "Buddy", "age": 5, "breed": "Golden Retriever"}
])

# 如果存在则替换
db["dogs"].insert({"id": 1, "name": "Cleo", "age": 5}, replace=True)

# 如果存在则忽略
db["dogs"].insert({"name": "Max"}, ignore=True)

3. 查询数据​

# 获取所有行
for row in db["dogs"].rows:
    print(row)

# 使用 SQL 查询
for row in db.query("SELECT * FROM dogs WHERE age > ?", [3]):
    print(row)

# 获取特定列
for row in db["dogs"].rows_where(select="name, age"):
    print(row)

# 带条件查询
for row in db["dogs"].rows_where("age > ?", [3]):
    print(row)

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

4. 更新数据​

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

# 批量更新
db["dogs"].update_where("breed = ?", ["Labrador"], {"age": 6})

# upsert (更新或插入)
db["dogs"].upsert({"id": 1, "name": "Cleo", "age": 6}, pk="id")

5. 删除数据​

# 删除特定记录
db["dogs"].delete(1)

# 批量删除
db["dogs"].delete_where("age < ?", [2])

# 删除表
db["dogs"].drop()

6. 表结构操作​

# 创建表并指定主键
db["users"].insert({"id": 1, "name": "Alice"}, pk="id")

# 添加列
db["dogs"].add_column("weight", float)

# 创建索引
db["dogs"].create_index(["breed"])

# 创建唯一索引
db["dogs"].create_index(["name"], unique=True)

# 添加外键
db["walks"].add_foreign_key("dog_id", "dogs", "id")

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

7. 主键和类型处理​

# 指定主键
db["users"].insert({"username": "alice", "email": "alice@example.com"}, pk="username")

# 复合主键
db["scores"].insert({"user_id": 1, "game_id": 2, "score": 100}, pk=("user_id", "game_id"))

# 自动检测类型
db["measurements"].insert({
    "temperature": 23.5,
    "humidity": 65,
    "recorded_at": "2024-01-07"
})

8. 批量操作和事务​

# 批量插入更高效
rows = [{"name": f"Dog{i}", "age": i} for i in range(1000)]
db["dogs"].insert_all(rows)

# 显式事务
with db.conn:
    db["dogs"].insert({"name": "Rex", "age": 3})
    db["cats"].insert({"name": "Whiskers", "age": 2})

9. 数据转换​

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

# 转换为 pandas DataFrame
import pandas as pd
df = pd.DataFrame(db["dogs"].rows)

# 从 DataFrame 导入
db["new_table"].insert_all(df.to_dict("records"))

10. 全文搜索​

# 启用 FTS (Full-Text Search)
db["articles"].enable_fts(["title", "content"])

# 搜索
for row in db["articles"].search("python"):
    print(row)

11. 实用方法​

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

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

# 获取行数
count = db["dogs"].count

# 获取表信息
print(db["dogs"].count_where("age > ?", [3]))

# 执行原始 SQL
db.execute("CREATE INDEX idx_age ON dogs(age)")

# 真空数据库(压缩)
db.vacuum()

12. 查找和替换​

# 使用 lookup 创建查找表
author_id = db["authors"].lookup({"name": "J.K. Rowling"})
# 如果不存在则创建,返回 ID

db["books"].insert({
    "title": "Harry Potter",
    "author_id": author_id
})

13. 高级查询​

# 使用 order_by
for row in db["dogs"].rows_where(order_by="age DESC"):
    print(row)

# 限制结果
for row in db["dogs"].rows_where(limit=5, offset=10):
    print(row)

# 分组统计
result = db.execute("""
    SELECT breed, COUNT(*) as count, AVG(age) as avg_age
    FROM dogs
    GROUP BY breed
""").fetchall()

实际应用示例​

from sqlite_utils import Database

# 创建数据库
db = Database("blog.db")

# 插入文章
db["posts"].insert_all([
    {"title": "First Post", "content": "Hello World", "author": "Alice"},
    {"title": "Second Post", "content": "Learning SQLite", "author": "Bob"}
], pk="title")

# 创建标签表
db["tags"].insert_all([
    {"post_title": "First Post", "tag": "introduction"},
    {"post_title": "First Post", "tag": "hello"},
    {"post_title": "Second Post", "tag": "tutorial"}
])

# 创建索引提高查询性能
db["posts"].create_index(["author"])
db["tags"].create_index(["tag"])

# 查询
for post in db["posts"].rows_where("author = ?", ["Alice"]):
    print(f"Title: {post['title']}")
    
# 统计每个作者的文章数
stats = db.execute("""
    SELECT author, COUNT(*) as post_count
    FROM posts
    GROUP BY author
""").fetchall()

print(stats)

命令行工具​

sqlite-utils 还提供了命令行工具:

# 插入 JSON 数据
echo '{"name": "Cleo", "age": 4}' | sqlite-utils insert dogs.db dogs -

# 查询并输出 JSON
sqlite-utils query dogs.db "SELECT * FROM dogs" --json

# 从 CSV 导入
sqlite-utils insert dogs.db dogs data.csv --csv

总结​

sqlite_utils 的主要优势:

  • 简洁的 API: 比原生 sqlite3 更易用
  • 自动类型检测: 无需手动定义表结构
  • 批量操作: 高效处理大量数据
  • 实用功能: FTS、lookup、upsert 等
  • 命令行工具: 快速数据操作

这个库特别适合数据处理、ETL 任务、快速原型开发等场景。