资讯动态

用类封装 SQLite3 增删改查:从建表到 CRUD 的完整实现

发布时间:2026/10/3 6:47:57 来源:尧图企业网站定制
1. 为什么零散 SQL 语句迟早要收进一个类里如果你写过几个小工具大概率经历过这个阶段脚本开头sqlite3.connect(data.db)中间到处cursor.execute(INSERT ...)结尾忘了commit下次跑的时候发现数据没进去。更麻烦的是同一个表在三个文件里被建了三次字段类型还不一样。这种写法在一次性脚本里能跑但只要需求稍微变一点——加个字段、换个查询条件、把内存库换成文件库——就得满文件找 SQL 字符串改。我试过把这类脚本直接复制到第二个项目里复用结果因为连接没关、游标复用、SQL 拼接引号错位调了半小时。后来我把所有数据库操作收进一个Database类连接、建表、增删改查各管各的调用方只关心「存一条用户」和「按名字查用户」不再碰 SQL 字符串。这篇文章就交付这个类的完整实现你可以直接复制到自己的脚本里用。核心检索词先明确Python 用类封装 SQLite3 数据库增删改查指的是把sqlite3模块的连接管理、建表语句、INSERT/SELECT/UPDATE/DELETE 四类操作包装成一个可实例化的类对外暴露insert、get、update、delete这样的方法。它适合谁适合正在写命令行小工具、爬虫落库脚本、本地配置管理、桌面小应用的开发者尤其是那些「不想引入 ORM 框架但又不想手写散落 SQL」的场景。SQLite3 本身是 Python 标准库自带的不需要pip install一个文件就是一个数据库非常适合本地小工具。但标准库的 API 是过程式的你要自己管连接、自己管游标、自己拼 SQL、自己 commit。类封装的价值就在于把这些重复动作固定下来同时把「表结构」和「操作入口」集中到一个地方改起来只改一处。下面我会先讲清楚这个类要解决哪些具体问题再给出可直接运行的完整代码然后演示调用和结果验证最后把常见的报错逐个拆开。全程用文件数据库demo.db演示你也可以改成:memory:做测试。2. 前置准备TaoToken 与运行环境怎么配这一节说两件事一是 Python 和 SQLite3 的环境确认二是如果你打算把这个数据库类接到大模型生成的代码流程里怎么用 TaoToken 拿到可用的 API 配置。先说环境因为这是跑通代码的前提。Python 3.6 以上都自带sqlite3你可以用下面命令确认版本和模块可用性python --version python -c import sqlite3; print(sqlite3.sqlite_version)如果第二条能打印出版本号比如3.39.5说明标准库没问题。不需要额外安装任何东西。数据库文件会在你运行脚本的目录下生成建议单独建一个data目录存放避免和代码混在一起。接下来说 TaoToken。它的定位是给开发者提供模型调用的统一入口你可以把它理解成「一个 Base URL 一个 Key就能调用多种模型」的接入层。官网地址是 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 根地址是 https://taotoken.net/api 。如果你只是本地跑 SQLite 封装其实不需要它但如果你想让模型帮你生成或补全这类数据库代码或者把「自然语言转 SQL」接进你的工具就需要先拿到 Key。拿 Key 的路径是进入控制台在 API Keys 页面创建一个新 Key。控制台入口在 https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite API Keys 页面在 https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite 。创建后把 Key 复制出来注意它只显示一次丢了只能重建。拿到 Key 之后你需要知道三个东西才能发起调用Base URL、Key、Model ID。Base URL 填https://taotoken.net/apiKey 填你刚创建的那串Model ID 填你要用的模型名在模型列表里能看到。这三个要素在后面配置 Claude Code、Cline 或 Codex 时都会反复出现先记牢。如果你用的是 Claude Code 这类编码工具想让它走 TaoToken 的接口需要配置环境变量或配置文件。Claude Code 的接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有 Base URL 和 Key 的填写位置说明。Coding Plan 适合长期编码和 Agent 场景入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。如果你只是想验证模型能不能正常返回用模型对话页面最快https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite 。注意Key 不要硬编码进提交到 Git 的脚本里。本地测试可以用环境变量TAOTOKEN_API_KEY或者放在不纳入版本管理的.env文件中。环境准备好之后我们就可以进入正题写这个Database类了。3. 可复制配置Database 类的完整实现这一节给出完整代码你可以直接保存为db.py。我把它拆成几个部分讲但代码是连续的复制整段即可运行。设计上遵循几个原则连接在__init__里建立建表用CREATE TABLE IF NOT EXISTS保证幂等增删改查各一个方法参数用字典传入避免 SQL 拼接引号出错所有写操作后 commit。先看完整代码# coding: utf-8 import sqlite3 import os class Database: 封装 SQLite3 的增删改查表结构固定为 user(name, age, job)。 def __init__(self, db_pathdemo.db): self.db_path db_path self.conn sqlite3.connect(db_path) # 让查询结果支持按列名取值 self.conn.row_factory sqlite3.Row self.cursor self.conn.cursor() self._create_table() def _create_table(self): 建表IF NOT EXISTS 保证重复调用不报错。 self.cursor.execute( CREATE TABLE IF NOT EXISTS user ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, age INTEGER NOT NULL, job TEXT NOT NULL ) ) self.conn.commit() def insert(self, name, age, job): 插入一条记录name 唯一重复插入会抛 IntegrityError。 try: self.cursor.execute( INSERT INTO user (name, age, job) VALUES (?, ?, ?), (name, age, job), ) self.conn.commit() return self.cursor.lastrowid except sqlite3.IntegrityError as e: print(f插入失败name 可能已存在: {e}) return None def get(self, name): 按 name 查询单条返回字典查不到返回空字典。 self.cursor.execute( SELECT id, name, age, job FROM user WHERE name ?, (name,) ) row self.cursor.fetchone() if row is None: return {} return {id: row[id], name: row[name], age: row[age], job: row[job]} def get_all(self): 查询全部记录返回字典列表。 self.cursor.execute(SELECT id, name, age, job FROM user ORDER BY id) rows self.cursor.fetchall() return [ {id: r[id], name: r[name], age: r[age], job: r[job]} for r in rows ] def update(self, name, **fields): 按 name 更新指定字段fields 只允许 age 和 job。 allowed {age, job} updates {k: v for k, v in fields.items() if k in allowed} if not updates: print(没有可更新的字段) return 0 if not self.get(name): print(f记录不存在: {name}) return 0 set_clause , .join(f{k} ? for k in updates) values list(updates.values()) [name] self.cursor.execute( fUPDATE user SET {set_clause} WHERE name ?, values ) self.conn.commit() return self.cursor.rowcount def delete(self, name): 按 name 删除返回受影响行数。 if not self.get(name): print(f记录不存在: {name}) return 0 self.cursor.execute(DELETE FROM user WHERE name ?, (name,)) self.conn.commit() return self.cursor.rowcount def close(self): 关闭连接脚本结束前调用。 self.cursor.close() self.conn.close() if __name__ __main__: db Database(demo.db) db.insert(zhangsan, 18, student) db.insert(lisi, 25, engineer) print(插入后查询 zhangsan:, db.get(zhangsan)) db.update(zhangsan, age78, jobEnglish) print(更新后查询 zhangsan:, db.get(zhangsan)) db.delete(wangwu) print(全部记录:, db.get_all()) db.close()几个关键设计点值得展开。第一row_factory sqlite3.Row让查询结果可以按列名访问row[name]比row[0]可读得多也不怕字段顺序变化。第二所有参数化查询都用?占位符把值作为元组传入这样既避免了 SQL 注入也省去了手动处理引号的麻烦——原 excerpt 里用{}拼接字符串遇到名字里带单引号就会炸。第三update方法用**fields接收关键字参数内部用白名单{age, job}过滤防止调用方传入name或id导致主键被改。第四CREATE TABLE IF NOT EXISTS保证脚本重复运行不会因为表已存在而报错。如果你要把这个类接到模型调用流程里比如让模型根据自然语言生成查询条件可以在get外面再包一层。但数据库类本身不需要知道模型的存在保持单一职责。关于配置文件的写法如果你用 Cline 或 Claude Code 这类工具它们的 MCP 或 settings 配置里需要填 Base URL、Key、Model ID 三件套。以 JSON 形式举例结构大致如下具体字段名以对应工具文档为准{ mcpServers: { taotoken: { baseUrl: https://taotoken.net/api, apiKey: 你的_TAOTOKEN_KEY, model: 你的_MODEL_ID } } }这段 JSON 里的三个值Base URL 固定是https://taotoken.net/apiKey 从 API Keys 页面拿Model ID 从模型列表里选。填错任何一个都会导致 401 或模型不存在。Cline MCP 的配置路径和 Claude Code 的 settings 文件位置不同接入文档里有对照说明。4. 验证请求跑一遍看结果对不对代码写完了得实际跑一遍确认行为符合预期。这一节给出运行命令、预期输出以及怎么用 SQLite 命令行工具二次验证数据真的落库了。先运行脚本python db.py预期输出类似插入后查询 zhangsan: {id: 1, name: zhangsan, age: 18, job: student} 更新后查询 zhangsan: {id: 1, name: zhangsan, age: 78, job: English} 记录不存在: wangwu 全部记录: [{id: 1, name: zhangsan, age: 78, job: English}, {id: 2, name: lisi, age: 25, job: engineer}]注意delete(wangwu)打印了「记录不存在」因为 wangwu 从没插入过这是预期行为不是报错。get_all返回两条记录zhangsan 的 age 和 job 已经被 update 改过。再用 SQLite 命令行确认数据确实写进了demo.db文件sqlite3 demo.db SELECT * FROM user;输出应该是1|zhangsan|78|English 2|lisi|25|engineer如果你机器上没有sqlite3命令行工具也可以用 Python 快速验证python -c import sqlite3; csqlite3.connect(demo.db); print(c.execute(SELECT * FROM user).fetchall())这一步的意义在于区分「类的内存状态」和「磁盘状态」。有时候你忘了 commit类里查询能查到因为同一个连接但换一个连接就查不到。用独立连接验证能确认 commit 真的生效了。再测一个边界情况重复插入同名用户。把db.insert(zhangsan, 18, student)再执行一次会看到插入失败name 可能已存在: UNIQUE constraint failed: user.name这是name TEXT NOT NULL UNIQUE约束在起作用insert方法捕获了IntegrityError并返回None不会让整个脚本崩掉。如果你希望重复插入时更新而不是报错可以把INSERT改成INSERT OR REPLACE但那样会改变 id需要根据业务决定。还有一个常见验证点更新不存在的记录。调用db.update(nobody, age30)会打印「记录不存在: nobody」并返回 0不会执行 UPDATE 语句。这个前置检查避免了「更新了 0 行但以为成功了」的困惑。如果你把数据库类接到模型调用上验证方式类似先确认模型能返回内容再确认返回的内容被正确解析并写进数据库。模型对话页面可以用来快速测一条请求确认 Base URL 和 Key 没问题。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth这一节把跑这个数据库类时可能遇到的报错以及接 TaoToken 时常见的接入错误逐个对照说明。数据库本身的报错和 API 接入的报错性质不同分开看。先说数据库侧的。第一个高频错误是sqlite3.OperationalError: no such table: user。原因通常是_create_table没被调用或者你连的数据库文件路径和建表时不一致。检查__init__里是否调用了self._create_table()以及db_path是否前后一致。如果你用相对路径注意脚本的工作目录——在 IDE 里运行和在终端里运行工作目录可能不同建议用绝对路径或os.path.join(os.path.dirname(__file__), demo.db)。第二个是sqlite3.ProgrammingError: You can only execute one statement at a time。这通常是因为你把多条 SQL 用分号连在一起传给execute。SQLite3 的execute一次只接受一条语句建表和多条插入要分开调用。我的_create_table里只有一条 CREATE所以没问题。第三个是sqlite3.IntegrityError: NOT NULL constraint failed。说明你插入时某个 NOT NULL 字段传了None。检查insert调用时三个参数是否都传了且不是None。第四个是忘记close()导致文件被占用。在 Windows 上如果连接没关删除demo.db会提示文件被占用。养成在脚本末尾调用db.close()的习惯或者用with上下文管理器需要给类加__enter__和__exit__。再说接 TaoToken 时的报错。第一个是401 Unauthorized。这几乎总是 Key 的问题Key 没填、填错、或者复制时带了空格。检查 API Keys 页面里的 Key 是否完整以及配置文件里有没有多余引号。注意 Base URL 是https://taotoken.net/api不要漏掉/api也不要多加斜杠。第二个是local proxy failed。这个报错通常出现在你本地配了代理但代理没启动或端口不对。检查你的环境变量HTTP_PROXY/HTTPS_PROXY是否指向了一个不可用的地址。如果你不需要代理把它们清空再试。第三个是reading choices 相关报错比如解析响应时读不到choices字段。这通常说明返回的不是标准对话结构可能是 Key 无权限、模型名写错、或者请求体格式不对。先确认 Model ID 和模型列表里的一致再检查请求 JSON 的字段名。第四个是OAuth 相关报错。如果你用 Claude Code 走 OAuth 登录而不是 API Key可能会遇到 token 过期或回调失败。这种情况下改用 API Key 方式接入更直接配置里填 Base URL、Key、Model ID 三件套即可。Claude Code 的接入文档里有两种方式的对照。提示排查接入问题时先用模型对话页面发一条最简单的请求确认 Base URL 和 Key 能通再回到代码里调。这样能把「网络/鉴权问题」和「代码问题」分开。把数据库报错和接入报错分开定位能省很多时间。数据库问题看 SQLite 的异常类型接入问题先看 HTTP 状态码。6. 把这个类用起来从脚本到小工具的落地建议代码跑通之后真正的问题是怎么把它用进你的项目。这一节给几个落地建议都是我在小工具里实际用过的。第一把表结构参数化。现在user表的字段是写死的如果你有多个表可以做一个基类BaseDB管连接和建表子类各自定义TABLE_SQL和字段白名单。这样insert、get、update、delete可以复用只是表名和字段不同。不过对于只有一两张表的小工具像现在这样单类写死反而更直观别过度设计。第二连接的生命周期要明确。命令行工具通常是「启动时建连接结束时关连接」用一次就退出。如果是常驻脚本比如定时任务可以在每次操作后 commit 但保持连接任务结束时再 close。不要在每次增删改查里都connect和close那样开销大且容易漏关。第三参数化查询是底线。原 excerpt 里用format拼 SQL遇到名字带单引号比如OBrien就会语法错误更严重的是有注入风险。我改成?占位符之后这类问题全部消失。你如果要在update里动态拼SET子句字段名必须来自白名单不能直接拼用户输入。第四给写操作加返回值。insert返回lastrowidupdate和delete返回rowcount调用方可以根据返回值判断是否成功而不是靠打印日志猜。这在批量操作时特别有用。第五测试用内存库。把db_path传:memory:每次运行都是干净的库适合写单元测试。生产用文件库测试用内存库切换只改一个参数。如果你想让模型帮你扩展这个类比如加一个「按 job 模糊查询」的方法可以把现有代码贴给模型让它按同样风格补一个方法。这时候 TaoToken 的模型对话入口就能派上用场。长期做这类编码辅助的话Coding Plan 比按次调用更划算。最后说一个我踩过的坑早期我把commit放在close里统一做结果脚本中途异常退出数据全丢了。后来改成每个写方法内部 commit虽然多几次磁盘写入但数据安全性高得多。对于本地小工具这个取舍是值得的。代码到这里就完整了。你可以直接把db.py复制走改表名和字段套进自己的脚本。遇到报错先对照第 5 节数据库问题和接入问题分开查基本都能定位。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑