1. 为什么本地脚本也要认真对待 SQLite 连接
如果你写过 Python 小工具,大概率用过sqlite3。它是标准库自带的模块,不用装任何东西,一个.db文件就能当数据库用,拷走就能跑,特别适合本地数据脚本、桌面小工具、轻量服务这类场景。但很多人对它的用法停留在「能跑就行」:连接、建表、插入、查询,五步走完就完事。等到脚本变多、表结构变复杂、需要多个脚本共享同一份数据时,问题就来了——连接到处散落、SQL 拼字符串、异常没人管、配置写死在代码里。
这篇要解决的就是这个工程化落地的问题。我会从sqlite3的基础五步讲起,然后把它封装成可复用的函数,再引入一份config.toml配置骨架,把数据库路径、超时、日志这些参数从代码里抽出来。同时,如果你在脚本里需要调用大模型做数据处理(比如把查询结果丢给模型总结、或者让模型生成 SQL),我会用 TaoToken 的统一 Key 和 API 通道把这块接进来,让本地数据库读写和模型调用走同一套配置。
适合谁看:写过一点 Python、用过sqlite3但想让代码更规整的人;想把本地脚本升级成小服务、又不想上重型数据库的人;以及需要在脚本里接入模型能力、但不想每个项目都重新配一遍 Key 的人。
先说清楚sqlite3的定位。它是文件型数据库,数据存在单个文件里,没有独立服务进程,读写直接操作文件。这带来两个特点:一是移植性极好,.db文件复制到哪都能用;二是并发写能力有限,多个进程同时写同一个文件会锁。所以它适合读多写少、单机运行的场景,不适合高并发写入的网络服务。理解这一点,后面的配置和封装才有方向。
2. TaoToken 前置:统一 Key 与 API 通道准备
在讲配置骨架之前,先把模型调用这条线准备好。TaoToken 的作用是提供一个统一的 API 入口和 Key 管理,你不用为每个模型或每个项目单独记一套地址和密钥。对于本地脚本来说,这意味着你可以在config.toml里只写一个 Key,脚本里通过统一的 base URL 去请求。
你需要先拿到 API Key。打开控制台,在 API Keys 页面创建一个新的 Key,复制保存好。这个 Key 就是后面配置里要填的值。
- 控制台入口:https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite
- API Keys 页面:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite
API 的基础地址是https://taotoken.net/api,这个地址在代码里作为 base URL 使用,注意它不带任何查询参数。如果你用的是 OpenAI 兼容的 SDK,把base_url指向它即可;如果是直接发 HTTP 请求,就拼上具体的路径。
这里要提醒一句:Key 属于敏感信息,不要硬编码在脚本里,也不要提交到 Git。后面我会把它放进config.toml,并且用环境变量兜底,这样本地开发和部署都能兼顾。
如果你只是想先验证模型通道是否通,可以打开模型对话页面直接试一句:
- 模型对话:https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite
确认能正常返回之后,再往下做配置。如果你打算长期在编码或 Agent 场景里用,可以了解一下 Coding Plan,它更适合高频调用:
- Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite
接入文档在这里,遇到参数问题可以对照查:
- 接入文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
3. 可复制配置:config.toml 骨架与 sqlite3 封装
3.1 config.toml 配置骨架
先建一个项目目录,比如sqlite_demo,在里面创建config.toml。这份配置把数据库参数和模型参数分开,结构清晰,后续加字段也方便。
# config.toml [database] # 数据库文件路径,相对路径以脚本运行目录为基准 path = "data/lite.db" # 连接超时,单位秒,避免锁等待时直接报错 timeout = 5.0 # 是否开启外键约束 foreign_keys = true [llm] # TaoToken 统一 API 地址,不带查询参数 base_url = "https://taotoken.net/api" # 建议通过环境变量 TAOTOKEN_API_KEY 注入,这里留空 api_key = "" # 默认使用的模型名称,按需替换 model = "gpt-4o-mini" # 请求超时,单位秒 timeout = 30 [log] level = "INFO" file = "logs/app.log"读取配置用 Python 3.11 起标准库自带的tomllib,不用额外装包。如果你用的是更早的版本,可以装tomli,用法一样。
# config_loader.py import os import tomllib from pathlib import Path def load_config(path: str = "config.toml") -> dict: with open(path, "rb") as f: cfg = tomllib.load(f) # 环境变量优先,避免把 Key 写进文件 env_key = os.environ.get("TAOTOKEN_API_KEY") if env_key: cfg["llm"]["api_key"] = env_key return cfg if __name__ == "__main__": cfg = load_config() print(cfg["database"]["path"]) print(cfg["llm"]["base_url"])运行python config_loader.py,应该能看到数据库路径和 API 地址被正确打印出来。这一步先确认配置读取没问题,再往下接数据库。
3.2 sqlite3 基础五步与封装
原始的五步是:创建连接、得到 cursor、执行 SQL、commit、关闭连接。直接写没问题,但每个函数都重复一遍就很啰嗦。我用一个上下文管理器把连接生命周期管起来,这样既保证关闭,又能统一处理异常。
# db.py import sqlite3 from contextlib import contextmanager from config_loader import load_config CFG = load_config() DB_PATH = CFG["database"]["path"] TIMEOUT = CFG["database"]["timeout"] FOREIGN_KEYS = CFG["database"]["foreign_keys"] @contextmanager def get_conn(): conn = sqlite3.connect(DB_PATH, timeout=TIMEOUT) if FOREIGN_KEYS: conn.execute("PRAGMA foreign_keys = ON") try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close()这里有几个细节值得说。timeout参数控制的是遇到锁时的等待时间,默认 5 秒,写多读少的场景可以调大。PRAGMA foreign_keys = ON是每个连接都要单独开的,SQLite 默认不启用外键约束,这点容易踩坑。commit放在yield之后,意味着只要没抛异常就自动提交,查询类操作提交也无害。
建表、插入、查询、删除、更新,全部基于这个上下文管理器来写:
# crud.py from db import get_conn def create_table(): with get_conn() as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS store ( item TEXT PRIMARY KEY, quantity INTEGER NOT NULL, price REAL NOT NULL ) """) def insert(item: str, quantity: int, price: float): with get_conn() as conn: conn.execute( "INSERT INTO store (item, quantity, price) VALUES (?, ?, ?)", (item, quantity, price), ) def view(): with get_conn() as conn: cur = conn.execute("SELECT item, quantity, price FROM store") return cur.fetchall() def delete(item: str): with get_conn() as conn: conn.execute("DELETE FROM store WHERE item = ?", (item,)) def update(quantity: int, price: float, item: str): with get_conn() as conn: conn.execute( "UPDATE store SET quantity = ?, price = ? WHERE item = ?", (quantity, price, item), )注意view里没有显式commit,因为查询不需要提交,fetchall拿到列表返回就行。另外我把item设成了主键,这样重复插入会报错,比默默插入重复数据更安全。参数全部用?占位,不要用字符串拼接,这是防注入的基本功。
3.3 接入模型调用
如果需要在脚本里调用模型,比如让模型根据自然语言生成 SQL,或者对查询结果做总结,可以用 OpenAI 兼容的方式请求 TaoToken。下面是一个最小封装:
# llm_client.py import httpx from config_loader import load_config CFG = load_config() LLM = CFG["llm"] def chat(prompt: str) -> str: url = f"{LLM['base_url']}/v1/chat/completions" headers = { "Authorization": f"Bearer {LLM['api_key']}", "Content-Type": "application/json", } payload = { "model": LLM["model"], "messages": [{"role": "user", "content": prompt}], } resp = httpx.post(url, json=payload, headers=headers, timeout=LLM["timeout"]) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"]调用前确保环境变量TAOTOKEN_API_KEY已经设置。这样数据库和模型两条线共用一份config.toml,改配置只改一个地方。
4. 验证请求:一次跑通读写与模型调用
配置和代码都齐了,现在做端到端验证。先建目录和文件:
mkdir -p sqlite_demo/data sqlite_demo/logs cd sqlite_demo touch config.toml config_loader.py db.py crud.py llm_client.py把前面的代码分别填进去,然后跑一个验证脚本:
# verify.py from crud import create_table, insert, view, update, delete create_table() insert("apple", 10, 3.5) insert("banana", 20, 1.2) print("插入后:", view()) update(15, 3.8, "apple") print("更新后:", view()) delete("banana") print("删除后:", view())运行python verify.py,预期输出类似:
插入后: [('apple', 10, 3.5), ('banana', 20, 1.2)] 更新后: [('apple', 15, 3.8), ('banana', 20, 1.2)] 删除后: [('apple', 15, 3.8)]如果能看到这三行,说明本地数据库读写链路已经通了。接着验证模型通道:
# verify_llm.py from llm_client import chat reply = chat("用一句话说明 SQLite 适合什么场景") print(reply)运行前先设置环境变量:
export TAOTOKEN_API_KEY="你的Key" python verify_llm.py能正常打印出模型回复,就说明统一 Key 和 API 通道也通了。到这里,数据库和模型两条线都验证完毕,可以开始写你自己的业务逻辑了。
5. 本篇常见错排查
5.1 建表报错 table already exists
这是最常见的。CREATE TABLE不带IF NOT EXISTS时,表已存在就会抛sqlite3.OperationalError。解决办法就是加上IF NOT EXISTS,我在create_table里已经这么写了。如果你改了表结构想重建,先DROP TABLE IF EXISTS store再建,但注意这会清空数据。
5.2 database is locked
多个进程或线程同时写同一个.db文件时会遇到。SQLite 同一时刻只允许一个写操作。排查方向:确认是不是有脚本没关连接,或者有长事务没提交。缓解办法是调大timeout,让后来的写操作多等一会儿;更彻底的做法是把写操作集中到一个进程里,或者改用 WAL 模式:
conn.execute("PRAGMA journal_mode = WAL")WAL 模式下读写可以并发,适合读多写少的场景。注意这个 PRAGMA 只需要设置一次,会持久化到数据库文件。
5.3 外键约束不生效
前面提过,SQLite 默认关闭外键约束,而且这个设置是连接级别的。如果你建了带外键的表却发现插入非法数据不报错,检查是不是每个连接都执行了PRAGMA foreign_keys = ON。我把它放在get_conn里,就是为了避免漏掉。
5.4 模型调用返回 401
先检查环境变量TAOTOKEN_API_KEY是否设置成功,可以用echo $TAOTOKEN_API_KEY确认。如果 Key 没问题,检查请求头里的Authorization格式是不是Bearer加 Key,中间有一个空格。另外确认 base URL 是https://taotoken.net/api,不要多加斜杠或路径。
5.5 配置读取报 FileNotFoundError
load_config默认从当前工作目录找config.toml。如果你在别的目录运行脚本,就会找不到。解决办法是用绝对路径,或者基于脚本所在目录拼接:
from pathlib import Path BASE_DIR = Path(__file__).resolve().parent cfg = load_config(BASE_DIR / "config.toml")这样无论从哪个目录运行,都能定位到配置文件。
6. 把两条线收进同一份配置
回头看,这篇做的事情其实就一件:把散落的连接和密钥收进一份config.toml,让数据库和模型调用共用一套配置骨架。sqlite3本身很简单,五步就能跑,但工程化落地靠的是连接管理、参数占位、异常回滚这些细节。模型调用也一样,统一 Key 和 base URL 之后,换模型、换项目都不用改代码结构。
如果你后面要把这套东西用到长期编码或 Agent 场景,可以看看 Coding Plan,它更适合高频调用:
- Coding Plan:https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite
需要新建或管理 Key 的时候,去 API Keys 页面:
- API Keys:https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite
参数细节对照接入文档:
- 接入文档:https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite
我自己的习惯是,每开一个新脚本项目,先把config.toml和db.py这两个文件复制过去,改一下数据库路径就能用。模型那条线如果暂时用不上,llm段留着不调用也不影响。等哪天需要了,填个 Key 就能接上,不用回头重构。