资讯详情

python操作sqlLite数据库:从建表到查询的完整实践与TaoToken统一Key接入

📅 2026/10/16 0:18:04 | 华诺云谱 👁 阅读
python操作sqlLite数据库:从建表到查询的完整实践与TaoToken统一Key接入
1. 为什么脚本里总需要一个本地数据库从内存字典到 sqlite3 的落地场景写 Python 脚本的人大多经历过这个阶段一开始用字典、列表存数据跑完就丢后来数据要跨天保留于是写 JSON 文件结果并发写坏过一次或者查询时得把整个文件读进内存再遍历。再往后数据量到几万条JSON 的读写开始变慢你就需要一个真正的本地数据库了。Python 标准库里的 sqlite3 正好卡在这个位置。它不需要你装任何服务端一个.db文件就是一个完整数据库支持 SQL 查询、事务、索引、主键约束。适合谁用写爬虫做去重、做本地日志归档、给小型桌面工具做配置存储、给数据分析脚本做中间结果缓存这些场景都合适。它不适合高并发写入也不适合多进程同时写同一个文件但单机单进程的脚本工具它几乎是最省心的选择。这篇内容围绕一条主线展开用 sqlite3 从建库建表开始把增删改查、事务提交、异常回滚全部跑通最后给你一份可以直接复制进项目的模板代码。同时脚本里如果还要调用外部大模型接口凭据管理容易散落各处我会顺带说明怎么用 TaoToken 的统一 Key 通道把模型调用的 Base URL 和 Key 收拢到一处避免每个脚本里硬编码一堆密钥。先说清楚 sqlite3 的几个核心对象后面代码才不会晕。connect()返回连接对象 conn它负责事务层面的操作commit()提交、rollback()回滚、close()关闭。连接对象上再cursor()出一个游标 cu所有 SQL 语句的执行都通过游标完成execute()执行单条、executemany()批量执行、fetchone()取一条、fetchall()取全部。记住这个分工写代码时就不会把 commit 写到游标上。还有一个容易忽略的点sqlite3 默认是自动开启事务的也就是说你的 INSERT、UPDATE、DELETE 不会立刻落盘必须显式 commit。很多人第一次写的时候发现数据没存进去八成就是忘了提交。反过来如果一批操作中途出错你可以 rollback 把这一批全部撤销这正是事务的价值。下面从零开始一步步把环境、建表、CRUD、事务、排错全部走一遍。你可以在本地新建一个目录跟着敲也可以直接跳到第 3 节复制完整模板。2. 环境准备与 TaoToken 统一 Key 前置让脚本里的模型调用凭据不再散落sqlite3 是标准库不需要 pip 安装Python 3.6 以上直接import sqlite3就能用。你可以先确认一下版本python --version python -c import sqlite3; print(sqlite3.sqlite_version)第二条命令会打印底层 SQLite 引擎的版本号一般会在 3.30 以上支持UPSERT、窗口函数这些新语法。如果你的环境打印报错说明 Python 安装不完整重装一个官方发行版即可。数据库文件本身不需要提前创建。sqlite3.connect(demo.db)时如果文件不存在它会自动新建如果传:memory:就是在内存里建一个临时库进程结束就没了适合写单元测试。这一点比 MySQL 那种要先建库建用户的方式省事太多。接下来是外部模型调用的凭据问题。很多脚本除了本地存储还要调大模型做摘要、分类、翻译。常见做法是把 API Key 直接写在代码里或者散落在多个.env文件时间一长自己都记不清哪个 Key 对应哪个服务。我的做法是统一走 TaoToken 的 API 通道把 Base URL 和 Key 集中管理。TaoToken 的 API 入口是https://taotoken.net/api官网是https://taotoken.net/?utm_sourcetaotoken_aicg_blog_end。你可以在控制台里创建 Key然后所有脚本都读同一份配置。这样做的好处是换 Key 只改一处脚本里不再出现明文密钥也方便区分不同项目的用量。具体操作上先去控制台生成一个 API Key地址是https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewrite。生成后不要直接写进代码而是放到环境变量或者一个独立的配置文件里。比如建一个config.py# config.py import os TAOTOKEN_BASE_URL https://taotoken.net/api TAOTOKEN_API_KEY os.environ.get(TAOTOKEN_API_KEY, ) # 模型 ID 按你实际开通的填写 DEFAULT_MODEL_ID claude-sonnet-4-5然后在 shell 里设置环境变量export TAOTOKEN_API_KEY你的KeyWindows PowerShell 用$env:TAOTOKEN_API_KEY你的Key。这样脚本里import config就能拿到不会把密钥提交到 Git。如果你用的是 Claude Code 这类命令行工具或者 Cline、Codex 这类需要配置 Base URL 的客户端配置项就是三件套Base URL 填https://taotoken.net/apiKey 填你生成的Model ID 填你开通的模型。三者缺一不可只填 Key 不填 Base URL 会走到默认端点通常直接 401。需要说明的是TaoToken 在这里的角色是统一管理模型调用凭据的通道本地数据库部分完全由 sqlite3 自己完成两者互不依赖。你可以只用 sqlite3也可以只用 TaoToken组合起来只是让脚本更整洁。环境准备好之后下一节进入正题建表。3. 可复制的建表与 CRUD 配置一份能直接跑通的 sqlite3 模板这一节给你一份完整可复制的代码。我把它拆成几个函数每个函数职责单一方便你按需取用。先看建表部分。# db_demo.py # -*- coding: utf-8 -*- import sqlite3 import os DB_FILE_PATH demo.db SHOW_SQL True def get_conn(pathDB_FILE_PATH): 获取数据库连接path 为空时使用内存库 if path: conn sqlite3.connect(path) return conn return sqlite3.connect(:memory:) def get_cursor(conn): 获取游标 if conn is not None: return conn.cursor() return get_conn().cursor() def close_all(conn, cu): 关闭游标和连接注意顺序 try: if cu is not None: cu.close() finally: if conn is not None: conn.close() def create_table(conn, sql): 创建表 if not sql: print(SQL 为空跳过) return cu get_cursor(conn) if SHOW_SQL: print(f执行 SQL: {sql}) cu.execute(sql) conn.commit() close_all(conn, cu) print(建表成功)建表语句用CREATE TABLE IF NOT EXISTS避免重复执行报错CREATE_TABLE_SQL CREATE TABLE IF NOT EXISTS student ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, gender TEXT, age INTEGER, city TEXT, phone TEXT ) 这里我把主键设成自增整数比用业务字段做主键更省心。字段类型上 SQLite 是动态类型但写上 TEXT、INTEGER 有助于可读性和后续迁移。接下来是插入。重点看参数化查询的写法这是防 SQL 注入的关键def save(conn, sql, data): 批量插入data 是元组列表 if not sql or data is None: print(SQL 或数据为空) return cu get_cursor(conn) try: for d in data: if SHOW_SQL: print(f执行 SQL: {sql}, 参数: {d}) cu.execute(sql, d) conn.commit() print(f插入 {len(data)} 条成功) except Exception as e: conn.rollback() print(f插入失败已回滚: {e}) finally: close_all(conn, cu)注意cu.execute(sql, d)里的d是元组SQL 里用?占位。绝对不要用字符串拼接把用户输入拼进 SQL那是注入漏洞的源头。比如SELECT * FROM student WHERE name ?配(Tom,)是安全的而fWHERE name {name}是危险的。查询部分def fetch_all(conn, sql, paramsNone): 查询全部 cu get_cursor(conn) if SHOW_SQL: print(f执行 SQL: {sql}, 参数: {params}) cu.execute(sql, params or ()) rows cu.fetchall() close_all(conn, cu) return rows def fetch_one(conn, sql, paramsNone): 查询单条 cu get_cursor(conn) cu.execute(sql, params or ()) row cu.fetchone() close_all(conn, cu) return row更新和删除同样用参数化def update(conn, sql, data): 批量更新data 是元组列表 cu get_cursor(conn) try: for d in data: cu.execute(sql, d) conn.commit() print(f更新 {len(data)} 条成功) except Exception as e: conn.rollback() print(f更新失败已回滚: {e}) finally: close_all(conn, cu) def delete(conn, sql, data): 批量删除 cu get_cursor(conn) try: for d in data: cu.execute(sql, d) conn.commit() print(f删除 {len(data)} 条成功) except Exception as e: conn.rollback() print(f删除失败已回滚: {e}) finally: close_all(conn, cu)把上面拼起来加一个main()就能跑def main(): conn get_conn() create_table(conn, CREATE_TABLE_SQL) conn get_conn() insert_sql INSERT INTO student (name, gender, age, city, phone) VALUES (?, ?, ?, ?, ?) data [ (Hongten, 男, 20, 广东省广州市, 13423****62), (Tom, 男, 22, 美国旧金山, 15423****63), (Jake, 女, 18, 广东省广州市, 18823****87), (Cate, 女, 21, 广东省广州市, 14323****32), ] save(conn, insert_sql, data) conn get_conn() rows fetch_all(conn, SELECT * FROM student) for r in rows: print(r) conn get_conn() update(conn, UPDATE student SET name ? WHERE id ?, [(HongtenAA, 1)]) conn get_conn() delete(conn, DELETE FROM student WHERE id ?, [(3,)]) conn get_conn() print(最终结果, fetch_all(conn, SELECT * FROM student)) if __name__ __main__: main()这份模板里每个函数自己开连接、自己关连接虽然有点啰嗦但对新手最不容易出错。等你熟练了可以改成上下文管理器with sqlite3.connect(...) as conn:的写法代码会更短。如果你还要在同一个脚本里调模型配置部分就引用第 2 节的config.pyBase URL 用https://taotoken.net/apiKey 从环境变量读Model ID 按开通的填。这样数据库和模型调用两套配置互不干扰。4. 运行验证与结果核对从空文件到查询输出的完整链路代码写完了怎么确认它真的跑通了我习惯按下面的顺序验证每一步都有明确的预期输出。第一步确认数据库文件生成。在终端执行python db_demo.py跑完后ls一下当前目录应该能看到demo.db。如果没看到说明connect()那步出了问题或者你用的是:memory:。文件大小一开始是 0 字节插入数据后会变成几 KB。第二步核对建表结果。用 sqlite3 命令行工具查看表结构sqlite3 demo.db .schema student预期输出是完整的CREATE TABLE语句。如果提示no such table说明建表那步没执行成功回去检查CREATE_TABLE_SQL有没有拼错或者create_table里有没有 commit。第三步核对插入数据。执行sqlite3 demo.db SELECT COUNT(*) FROM student;预期输出是 4如果你按模板插了 4 条。如果输出 0八成是忘了 commit或者插入时抛异常被 rollback 了。这时候看脚本打印的日志插入失败已回滚后面会跟具体错误。第四步核对更新和删除。模板里更新了 id1 的名字删除了 id3 的记录。查询验证sqlite3 demo.db SELECT id, name FROM student ORDER BY id;预期看到 id1 的名字变成 HongtenAAid3 的记录消失剩下 1、2、4 三条。第五步验证参数化查询确实生效。手动构造一个带引号的输入比如查询名字里带单引号的情况conn get_conn() row fetch_one(conn, SELECT * FROM student WHERE name ?, (OBrien,)) print(row)如果用的是参数化这条查询不会报语法错误只会返回 None。如果你之前用的是字符串拼接这里就会抛sqlite3.OperationalError。这个对比能让你直观感受到参数化的价值。第六步验证事务回滚。故意在批量插入里放一条违反约束的数据比如主键重复conn get_conn() bad_data [(A, 男, 20, 北京, 111), (A, 男, 20, 北京, 111)] save(conn, INSERT INTO student (name, gender, age, city, phone) VALUES (?, ?, ?, ?, ?), bad_data)因为 id 是自增的name 重复不会报错所以这个例子不会触发回滚。要触发回滚可以给某个字段加 UNIQUE 约束或者手动在 SQL 里写错字段名。更直接的办法是插入时故意少传参数conn get_conn() save(conn, INSERT INTO student (name, gender) VALUES (?, ?), [(X,)])参数数量不匹配会抛ProgrammingError被 except 捕获后 rollback数据库里不会留下半条记录。你可以插入前后各查一次 COUNT确认数量没变。跑完这六步你对这套模板的行为就有底了。实际项目里我建议把验证脚本单独放一个文件每次改完数据库逻辑就跑一遍比手动点来点去可靠。5. 常见报错排查401、local proxy failed、reading choices 与 OAuth 的对照处理这一节把实际会撞到的报错集中列一下每个都给出定位思路。注意这些报错分两类一类来自 sqlite3 本身一类来自外部模型调用。分开看。先说 sqlite3 侧的。sqlite3.OperationalError: no such table: student。原因通常是建表语句没执行或者连接的是另一个数据库文件。检查DB_FILE_PATH是否一致以及create_table有没有真的被调用。如果你在:memory:上建表然后换了个连接去查那必然找不到表因为内存库是每个连接独立的。sqlite3.ProgrammingError: Incorrect number of bindings supplied。参数个数和占位符数量对不上。比如 SQL 里有 5 个?你只传了 4 个元素的元组。注意单个参数的写法cu.execute(sql, (value,))那个逗号不能省否则会被当成普通值而不是元组。sqlite3.IntegrityError: UNIQUE constraint failed。插入了违反唯一约束的数据。要么改数据要么用INSERT OR IGNORE或INSERT OR REPLACE。这个错误会触发事务回滚如果你在循环里批量插入一条失败会导致整批撤销看日志确认。database is locked。多进程或多线程同时写同一个文件时会遇到。sqlite3 默认的锁粒度比较粗解决办法是给连接设置超时sqlite3.connect(path, timeout10)或者把写操作集中到一个进程里串行执行。再说模型调用侧的这些和 sqlite3 无关但脚本里经常一起出现。401 Unauthorized。Key 不对、没传、或者传了但 Base URL 不对。检查三件套Base URL 是不是https://taotoken.net/apiKey 是不是从环境变量正确读取Model ID 是不是你开通的那个。只配 Key 不配 Base URL 是最常见的坑。local proxy failed或连接超时。这类报错通常是本地网络配置问题检查你的请求是否走了预期通道。脚本里如果用 requests 或 openai 库确认没有多余的代理设置干扰。Error reading choices或返回体里没有 choices 字段。说明请求发出去了但响应结构不符合预期。常见原因是 Model ID 填错或者请求体格式不对。打印完整响应体看返回的 JSON 里到底有什么字段。OAuth相关报错。如果你用的是需要 OAuth 授权的客户端token 过期会导致这类错误。重新走一遍授权流程或者换成 API Key 方式。用统一 Key 通道的好处就在这里不用维护多套 OAuth 凭据一个 Key 管所有脚本。排查顺序建议先确认 sqlite3 部分单独能跑通再接入模型调用。两者混在一起调试很容易把数据库错误误判成网络错误。分开验证定位快很多。6. 把模板用起来从单文件脚本到可复用模块的下一步到这里建表、CRUD、事务、验证、排错都过了一遍。你手上应该有一份能跑的db_demo.py以及一份config.py管模型凭据。接下来怎么把它变成项目里真正能用的东西我的做法是把数据库操作抽成一个类连接在初始化时建立方法里不再反复开关连接。比如class StudentDB: def __init__(self, pathDB_FILE_PATH): self.conn sqlite3.connect(path, timeout10) self.conn.row_factory sqlite3.Row def close(self): self.conn.close() def add(self, name, gender, age, city, phone): sql INSERT INTO student (name, gender, age, city, phone) VALUES (?, ?, ?, ?, ?) with self.conn: self.conn.execute(sql, (name, gender, age, city, phone)) def all(self): cur self.conn.execute(SELECT * FROM student) return cur.fetchall()with self.conn:会自动 commit出错自动 rollback比手动写 try/except 干净。row_factory sqlite3.Row让查询结果支持按列名访问row[name]比row[1]可读性好得多。如果你还要在同一个项目里调模型把config.py里的 Base URL 和 Key 读进来请求时带上就行。TaoToken 的模型对话入口在https://taotoken.net/models?utm_sourcetaotoken_aicg_blog_endutm_contentmodelsutm_campaignrewrite接入文档在https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite需要长期跑编码任务的可以看 Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite。API Key 管理在https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi-keysutm_campaignrewrite。最后给一个实用建议数据库文件不要提交到 Git在.gitignore里加上*.db。需要初始化数据的话写一个init_db.py专门负责建表和插入种子数据每次部署跑一次就行。这样你的脚本既有本地持久化能力又不会把二进制文件混进版本历史。
📝

华诺云谱内容团队

资深建站顾问 · 行业研究员

10年+企业数字化服务经验,专注智能建站、SEO优化与品牌营销,持续输出建站技巧、行业洞察与营销干货,已帮助5000+企业实现数字化增长。

你可能需要的服务

订阅华诺云谱资讯周报

每周一封,精选建站技巧、SEO与营销干货,直达邮箱。已有 8,000+ 企业主订阅,助你少走弯路。

↑