1. 为什么要在 pymysql 里区分单步连接和 with 方式连接如果你写过 Python 连 MySQL 的脚本大概率见过这两种写法一种是conn pymysql.connect(**cfg)然后手动close()另一种是with pymysql.connect(**cfg) as conn:让上下文管理器帮你收尾。它们看起来只是风格差异但在真实项目里选错写法会带来连接泄漏、事务没提交、游标提前失效这些很难查的问题。这篇聚焦的场景很具体本地 Python 项目里用 pymysql 分别跑通单步连接和 with 方式连接并且让两种方式在同一个配置骨架下行为一致。配置来源我统一放在settings.jsonKey 和 API 通道走 TaoToken 的统一入口这样数据库连接参数和模型调用参数可以放在同一份配置里管理不用在多个文件之间来回找。适合谁看刚接触 pymysql、分不清cursorclass和with关系的新手已经在用 pymysql 但连接偶尔报Too many connections的开发者想把数据库配置和 API Key 配置收敛到一份 JSON 的人。下面所有代码都可以直接复制到本地跑我会给出完整的settings.json骨架、两种连接方式的模板以及一个能验证两者结果一致的脚本。先说结论方向单步连接适合需要精细控制提交时机、或者一个连接里跑多段独立逻辑的场景with 方式适合「拿连接、执行、关闭」这种一次性的短操作。两者不是替代关系而是分工关系。2. TaoToken 前置统一 Key 与 settings.json 的定位在动手写连接代码之前先把配置层理清楚。很多教程把数据库密码硬编码在脚本里跑通没问题但一旦要换环境就到处改。我的做法是把数据库连接参数和 TaoToken 的 API Key 都放进settings.json脚本只负责读配置。TaoToken 在这里的角色是统一 Key 和 API 通道你可以在它的控制台里管理 Key模型对话、编码计划、API Keys 这些入口是分开的。对于本篇的 pymysql 场景数据库本身还是连你自己的 MySQLTaoToken 负责的是配置管理和后续可能用到的模型调用通道两者通过同一份settings.json组织在一起避免配置散落。需要提前准备的入口按用途分想先验证 Key 能不能用、模型通道是否正常模型对话入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel_chat要生成或查看 API KeyAPI Keys 入口https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi_keys长期做编码、跑 Agent 任务Coding Plan 入口https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding_plan查接入文档、确认参数格式文档入口https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc控制台总览https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentconsoleAPI 基础地址是https://taotoken.net/api注意这个地址不带 UTM 参数写进代码里就用这个。官网是https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content。注意数据库密码和 API Key 都属于敏感信息settings.json不要提交到公开仓库建议加进.gitignore本地用环境变量覆盖也行。3. 可复制配置settings.json 骨架与连接参数模板先给一份完整的settings.json骨架。它分成两块database放 pymysql 需要的连接参数taotoken放统一 Key 和 API 地址。这样一份文件同时服务数据库和模型通道。{ database: { host: 127.0.0.1, port: 3306, user: root, password: your_db_password, db: demo_db, charset: utf8mb4, cursorclass: DictCursor, connect_timeout: 10, autocommit: false }, taotoken: { api_base: https://taotoken.net/api, api_key: sk-your-taotoken-key, default_model: your-model-name } }几个参数值得单独说。cursorclass我写成字符串DictCursor因为 JSON 不能直接存类对象读出来之后要映射成pymysql.cursors.DictCursor。charset用utf8mb4而不是utf8避免存 emoji 或四字节字符时报错。autocommit默认false这样你能明确控制什么时候提交with 方式下这点尤其重要。读取配置的代码import json import pymysql CURSOR_MAP { Cursor: pymysql.cursors.Cursor, DictCursor: pymysql.cursors.DictCursor, SSCursor: pymysql.cursors.SSCursor, SSDictCursor: pymysql.cursors.SSDictCursor, } def load_db_config(pathsettings.json): with open(path, r, encodingutf-8) as f: raw json.load(f)[database] cfg dict(raw) cfg[cursorclass] CURSOR_MAP[raw.get(cursorclass, DictCursor)] return cfg这里把cursorclass从字符串转成真正的类后面pymysql.connect(**cfg)才能直接用。如果你用的是Cursor而不是DictCursorfetchone()返回的是元组取值要用下标row[0]而DictCursor返回字典可以用row[c]。这个差异在两种连接方式里都存在跟连接方式无关但很容易混。4. 单步连接与 with 方式连接的完整模板4.1 单步连接手动管理生命周期单步连接的核心是「谁创建、谁关闭」。连接和游标都要显式close()顺序是先关游标再关连接。def query_count_single(cfg, tb_name): conn pymysql.connect(**cfg) cur conn.cursor() try: cur.execute(select count(*) as c from tb_name) row cur.fetchone() return row[c] finally: cur.close() conn.close()注意finally块即使execute抛异常游标和连接也会被关闭。如果只写cur.close()和conn.close()在函数末尾一旦中间报错就会泄漏连接。表名用字符串拼接是因为 pymysql 的参数化占位符%s不能用于表名但生产环境一定要对tb_name做白名单校验别直接拼用户输入。单步连接适合这些情况一个连接里要跑多段逻辑、需要手动commit()或rollback()、需要复用同一个连接做多次查询。它的代价是你得自己保证关闭写漏一处就出问题。4.2 with 方式连接上下文管理器自动收尾with 方式分两种写法取决于你的cursorclass。如果cursorclass是DictCursorwith pymysql.connect(**cfg) as cur:直接给你一个游标不用再创建def query_count_with_dict(cfg, tb_name): with pymysql.connect(**cfg) as cur: cur.execute(select count(*) as c from tb_name) row cur.fetchone() return row[c]如果cursorclass是默认的Cursorwith ... as conn给的是连接需要自己再conn.cursor()def query_count_with_conn(cfg, tb_name): with pymysql.connect(**cfg) as conn: with conn.cursor() as cur: cur.execute(select count(*) as c from tb_name) row cur.fetchone() return row[0]这两种写法的区别来自 pymysql 对__enter__的实现连接对象的__enter__返回的是游标当cursorclass指定时或连接本身。很多人第一次用 with 会困惑「为什么我拿到的是游标不是连接」原因就在这里。注意with 方式在退出时pymysql 的连接__exit__会调用close()但不会自动 commit。如果你的操作是写库必须在 with 块内显式conn.commit()否则数据不会落库。这是最常见的坑之一。4.3 两种方式的行为对照维度单步连接with 方式连接创建conn pymysql.connect(**cfg)with pymysql.connect(**cfg) as x:游标手动conn.cursor()DictCursor 下直接是游标否则手动创建关闭手动cur.close()conn.close()退出 with 自动关闭提交手动conn.commit()仍需手动conn.commit()异常安全依赖 try/finally自动关闭但提交仍需自己处理适用多段逻辑、精细控制一次性短操作5. 验证请求与成功结果确认两种方式行为一致光看代码不够得跑一个脚本确认两种方式返回的结果一致。下面这个脚本读同一份settings.json分别用单步和 with 查同一张表的行数然后比对。import json import pymysql CURSOR_MAP { Cursor: pymysql.cursors.Cursor, DictCursor: pymysql.cursors.DictCursor, } def load_cfg(pathsettings.json): with open(path, r, encodingutf-8) as f: raw json.load(f)[database] cfg dict(raw) cfg[cursorclass] CURSOR_MAP[raw.get(cursorclass, DictCursor)] return cfg def count_single(cfg, tb): conn pymysql.connect(**cfg) cur conn.cursor() try: cur.execute(select count(*) as c from tb) return cur.fetchone()[c] finally: cur.close() conn.close() def count_with(cfg, tb): with pymysql.connect(**cfg) as cur: cur.execute(select count(*) as c from tb) return cur.fetchone()[c] if __name__ __main__: cfg load_cfg() tb your_table_name a count_single(cfg, tb) b count_with(cfg, tb) print(single:, a) print(with :, b) print(consistent:, a b)跑通后你会看到类似输出single: 128 with : 128 consistent: True如果consistent是True说明两种连接方式在读取行为上一致。如果两个数字不一样通常是连接参数不同比如一个连了测试库一个连了正式库或者表在两次查询之间被改了。先确认settings.json只有一份、两个函数读的是同一份配置。再补一个写操作的验证确认 with 方式下提交行为def insert_with_commit(cfg): with pymysql.connect(**cfg) as conn: with conn.cursor() as cur: cur.execute(insert into demo_table(name) values(%s), (test_row,)) conn.commit()这里conn.commit()在 with 块内、游标块外。如果你把它放到 with 块外面连接已经关闭会报错。这个顺序要记牢先提交再退出 with。6. 本篇常见错排查6.1 报错Cursor object has no attribute commit原因你把with pymysql.connect(**cfg) as cur拿到的当成连接用了但它其实是游标因为cursorclass指定了。游标没有commit方法。解决要么改成with pymysql.connect(**cfg) as conn:再conn.cursor()要么在 with 块内通过连接对象提交。如果你确实需要提交建议用连接形式的 with。6.2 报错(2006, MySQL server has gone away)原因连接空闲太久被服务端断开或者connect_timeout太短。单步连接如果创建后长时间不用再执行容易碰到。解决在settings.json里把connect_timeout调大一点或者用conn.ping(reconnectTrue)在每次执行前探活。with 方式因为生命周期短一般不容易遇到。6.3 数据没写进去但脚本没报错原因autocommit是false你没调commit()。with 方式退出时只关连接不提交。解决在 with 块内显式conn.commit()或者把settings.json里的autocommit改成true但这样会失去事务控制写操作要谨慎。6.4KeyError: c或TypeError: tuple indices must be integers原因cursorclass是Cursor而不是DictCursorfetchone()返回元组你用row[c]取值就报错。解决统一用DictCursor或者取值时改成row[0]。检查settings.json里的cursorclass字段和CURSOR_MAP映射是否对得上。6.5 连接数暴涨Too many connections原因单步连接里某条异常路径没走到close()连接泄漏。常见于try块里return了但finally没写全。解决所有单步连接都套try/finally或者干脆改用 with 方式。排查时可以在 MySQL 里执行show processlist;看有多少空闲连接。6.6 表名拼接导致 SQL 报错原因tb_name带了反引号或特殊字符拼接后语法错误。解决表名用反引号包裹比如select count(*) as c from tb 并且对tb_name做白名单校验只允许字母数字下划线。7. 下一步把配置和 Key 收敛到一处两种连接方式跑通之后建议把settings.json作为项目唯一的配置入口。数据库参数、TaoToken 的 API Key、默认模型都放这里脚本通过load_cfg()读取。这样换环境只改一个文件不用翻代码。如果你接下来要验证模型通道是否正常可以用模型对话入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentmodel_chat发一条测试消息。要生成新的 API Key 或查看已有 Key走 API Keys 入口https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentapi_keys。接入参数格式不确定时查文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentdoc。长期跑编码任务或 Agent用 Coding Plan 入口https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentcoding_plan更合适。控制台总览在https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentconsole。最后留一个实用习惯每次改完settings.json先跑一遍第 5 节的验证脚本确认consistent: True再继续写业务代码。这个动作花不了几秒但能挡掉大部分「配置改了没生效」的问题。