1. 为什么本地脚本连 MySQL 总在配置上翻车如果你用 python3 写过数据同步、定时报表、爬虫落库这类小工具大概率绕不开 pymysql。它足够轻纯 Python 实现装完就能连 MySQL不像某些驱动还要编译环境。但真正让脚本从「能跑」变成「敢长期跑」的往往不是 SQL 本身而是配置管理。我见过太多脚本把 host、user、password 直接写死在代码里换台机器就得改一遍也见过把数据库密码和模型 API Key 混在同一个 .env 里结果 git 一提交全泄露。更麻烦的是现在很多脚本除了写库还要调用大模型做摘要、分类、生成于是又多了一套 API Key 要管。数据库凭证和模型凭证散落各处排查问题时根本不知道哪个 Key 失效了。这篇就聚焦一个具体场景python3 用 pymysql 操作 MySQL 的工程化落地。我会给出一份可复制的 config.toml 配置骨架把数据库连接参数和模型调用凭证统一收口再用 TaoToken 的统一 Key 通道管理调用凭证最后通过一次查询加一次写入验证整套配置真的生效。适合本地脚本、小型服务、个人项目这类不需要上重型配置中心的场景。核心检索词先摆出来python3 怎么用 pymysql 连 mysql、config.toml 怎么写数据库配置、TaoToken 统一 Key 怎么接入。下面按可跟做的步骤走。2. 前置准备pymysql 安装与 TaoToken 统一 Key先说依赖。pymysql 的安装没什么坑一条命令pip install pymysql如果你要用 TOML 解析python3.11 及以上自带 tomllib低版本装一个pip install tomli读取时做个兼容即可后面代码里会体现。接下来是凭证管理这块。数据库这边我建议单独建一个只对业务库有权限的账号别用 root 跑脚本。模型调用这边如果你只是偶尔调一下直接用官方 Key 也行但如果脚本里要频繁调用、或者多个脚本共用一套凭证用 TaoToken 的统一 Key 会省事很多。它的思路是把调用凭证集中管理脚本里只认一个 Key换模型、换通道都不用改业务代码。TaoToken 的 API 地址是https://taotoken.net/api控制台在https://taotoken.net/consoleAPI Key 在https://taotoken.net/api-keys生成。注意 API 地址不要加多余参数直接用它给的 base 就行。生成 Key 之后把它和数据库密码一起放进 config.toml但 config.toml 本身要加进 .gitignore这点后面会强调。提示数据库账号和模型 Key 都属于敏感信息配置文件不要提交到公开仓库。可以放一份 config.example.toml 做模板真实配置本地保留。3. 可复制的 config.toml 配置骨架下面这份骨架是我在多个小脚本里沉淀下来的分三块database、llm、app。database 放 pymysql 需要的连接参数llm 放 TaoToken 的 Key 和 base_urlapp 放一些业务开关。# config.toml [app] name trade-sync debug true log_level INFO [database] host 127.0.0.1 port 3306 user app_user password your_db_password db test charset utf8mb4 connect_timeout 5 read_timeout 10 write_timeout 10 [llm] provider taotoken base_url https://taotoken.net/api api_key sk-your-taotoken-key model gpt-4o-mini timeout 30几个参数值得说明。charset 用 utf8mb4 而不是 utf8因为 MySQL 的 utf8 其实是阉割版存 emoji 或某些生僻字会出问题utf8mb4 才是完整的四字节 UTF-8。connect_timeout 和读写超时建议显式设置默认值在弱网或数据库压力大时容易让脚本卡死。llm 段的 base_url 指向 TaoToken 的 API 入口api_key 就是你在控制台生成的那串。读取配置的代码要兼容 tomllib 和 tomliimport sys if sys.version_info (3, 11): import tomllib else: import tomli as tomllib def load_config(pathconfig.toml): with open(path, rb) as f: return tomllib.load(f) CONFIG load_config()这样一份配置数据库和模型凭证都在同一个文件里但逻辑上分开了。换数据库只动 database 段换模型只动 llm 段互不影响。4. pymysql 连接封装与增删改查实操有了配置接下来封装连接。直接每次 Connect 也行但小服务里更推荐用一个获取连接的函数方便统一加参数。import pymysql from pymysql.cursors import DictCursor def get_connection(): db CONFIG[database] return pymysql.connect( hostdb[host], portdb[port], userdb[user], passworddb[password], databasedb[db], charsetdb[charset], connect_timeoutdb[connect_timeout], read_timeoutdb[read_timeout], write_timeoutdb[write_timeout], cursorclassDictCursor, autocommitFalse, )注意这里用了 DictCursor查询结果直接是字典比默认的元组好用。autocommit 设为 False写操作手动 commit事务边界更清晰。先建一张测试表方便后面验证CREATE TABLE IF NOT EXISTS trade ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(64) NOT NULL, account VARCHAR(32) NOT NULL, saving DECIMAL(12,2) DEFAULT 0, expend DECIMAL(12,2) DEFAULT 0, income DECIMAL(12,2) DEFAULT 0, UNIQUE KEY uk_account (account) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;插入数据用参数化别用字符串拼接。原始 excerpt 里那种sql % data的写法有 SQL 注入风险工程里要避免def insert_trade(name, account, saving): conn get_connection() try: with conn.cursor() as cursor: sql INSERT INTO trade (name, account, saving) VALUES (%s, %s, %s) cursor.execute(sql, (name, account, saving)) conn.commit() print(成功插入, cursor.rowcount, 条数据) except Exception as e: conn.rollback() print(插入失败, e) finally: conn.close()查询和更新同理参数化占位符统一用 %spymysql 会自动处理类型转义。事务处理那段把多条 SQL 放在 try 里异常时 rollback正常时 commitdef transfer(account): conn get_connection() try: with conn.cursor() as cursor: cursor.execute(UPDATE trade SET saving saving 1000 WHERE account %s, (account,)) cursor.execute(UPDATE trade SET expend expend 1000 WHERE account %s, (account,)) cursor.execute(UPDATE trade SET income income 2000 WHERE account %s, (account,)) conn.commit() print(事务处理成功) except Exception as e: conn.rollback() print(事务处理失败, e) finally: conn.close()这里有个细节cursor.rowcount在 with 块外访问时游标可能已经关闭所以打印行数最好在 with 内部完成或者提前存到变量里。5. 用 TaoToken 统一 Key 调用模型做数据摘要数据库写完了现在把模型调用接进来。场景很实际比如 trade 表里存了一堆交易记录你想让模型生成一段自然语言摘要或者对某条记录做分类。用 TaoToken 的 base_url 加统一 Key代码里只认配置不写死任何厂商细节。import json import urllib.request def summarize_trade(rows): llm CONFIG[llm] prompt 请用一句话总结以下交易记录\n json.dumps(rows, ensure_asciiFalse) payload { model: llm[model], messages: [{role: user, content: prompt}], temperature: 0.3, } req urllib.request.Request( llm[base_url].rstrip(/) /v1/chat/completions, datajson.dumps(payload).encode(utf-8), headers{ Content-Type: application/json, Authorization: Bearer llm[api_key], }, methodPOST, ) with urllib.request.urlopen(req, timeoutllm[timeout]) as resp: result json.loads(resp.read().decode(utf-8)) return result[choices][0][message][content]这里用标准库 urllib 是为了少装依赖你换成 requests 或 openai SDK 都行关键是 base_url 和 api_key 从配置读。TaoToken 的 API 入口是https://taotoken.net/api拼接路径时注意别重复斜杠。如果你更习惯用 OpenAI 兼容的 SDK可以这样from openai import OpenAI client OpenAI( base_urlCONFIG[llm][base_url] /v1, api_keyCONFIG[llm][api_key], ) resp client.chat.completions.create( modelCONFIG[llm][model], messages[{role: user, content: 你好}], ) print(resp.choices[0].message.content)两种方式都指向同一个统一 Key换模型只改 config.toml 里的 model 字段业务代码不动。6. 验证请求一次查询加一次写入确认配置生效配置写完不验证等于没写。下面这段脚本把数据库和模型两条链路都跑一遍。def verify(): # 1. 写入一条测试数据 insert_trade(测试用户, 13900000000, 5000.00) # 2. 查询刚写入的数据 conn get_connection() try: with conn.cursor() as cursor: cursor.execute( SELECT name, saving FROM trade WHERE account %s, (13900000000,), ) rows cursor.fetchall() for row in rows: print(Name:%s Saving:%.2f % (row[name], row[saving])) print(共查找出, cursor.rowcount, 条数据) finally: conn.close() # 3. 调用模型做摘要 summary summarize_trade(rows) print(模型摘要:, summary) if __name__ __main__: verify()预期输出类似先打印「成功插入 1 条数据」再打印查询到的用户名和金额最后打印模型返回的一句话摘要。如果数据库段报连接错误检查 host、port、user、password如果模型段报 401检查 api_key 是否复制完整如果报 404检查 base_url 拼接路径。实测下来这套结构跑通之后后面加表、加模型调用都只是往 config.toml 里加字段的事业务代码几乎不用动。7. 本篇常见错排查报错 1pymysql.err.OperationalError: (2003, Cant connect to MySQL server)先确认 MySQL 服务在跑再确认 host 和 port。如果数据库在容器里127.0.0.1 可能不通要换成容器网络里的地址。connect_timeout 设太小也会误报弱网环境调到 10 秒试试。报错 2pymysql.err.OperationalError: (1045, Access denied for user)账号密码不对或者该账号没有从当前 host 连接的权限。MySQL 的 user 表里 host 字段如果是 localhost用 127.0.0.1 连可能被拒需要单独授权。报错 3UnicodeEncodeError或中文乱码charset 没设成 utf8mb4或者建表时用了 utf8。两边都要统一。另外连接参数里 charset 和建表语句的字符集要一致。报错 4模型调用返回 401api_key 没填对或者复制时带了空格。TaoToken 的 Key 在控制台生成后只显示一次丢了就重新生成。注意 Authorization 头是Bearer加 Key中间有空格。报错 5模型调用返回 404base_url 拼接错了。TaoToken 的 API 入口是https://taotoken.net/api如果 SDK 内部会自动加/v1你就不要再手动加一遍。用 urllib 手写时路径是/api/v1/chat/completions。报错 6cursor.rowcount打印为 -1某些查询或游标类型下 rowcount 不可靠尤其是 SELECT 之后。要拿准确行数用len(cursor.fetchall())更稳。报错 7事务没生效数据没写进去检查 autocommit 是否为 False以及有没有在异常分支里 rollback 后忘了重新 commit。另外 InnoDB 才支持事务MyISAM 不支持。8. 凭证收口之后脚本才真正可维护把数据库连接和模型调用都收进 config.toml再用 TaoToken 的统一 Key 管住模型凭证最大的好处是排查问题时路径清晰。数据库连不上就看 database 段模型报错就看 llm 段不用在几百行代码里翻 Key。如果你后面要把脚本跑成长期服务或者多个脚本共用一套凭证建议去控制台把 Key 按用途分开管理再配合 Coding Plan 这类长期方案做额度规划。接入文档里有更细的参数说明遇到拼接路径或鉴权问题时对着查一遍比盲试快得多。 python3 使用 pymysql 操作 mysql：TaoToken 统一 Key 接入与 config.toml 配置骨架
1. 为什么本地脚本连 MySQL 总在配置上翻车如果你用 python3 写过数据同步、定时报表、爬虫落库这类小工具大概率绕不开 pymysql。它足够轻纯 Python 实现装完就能连 MySQL不像某些驱动还要编译环境。但真正让脚本从「能跑」变成「敢长期跑」的往往不是 SQL 本身而是配置管理。我见过太多脚本把 host、user、password 直接写死在代码里换台机器就得改一遍也见过把数据库密码和模型 API Key 混在同一个 .env 里结果 git 一提交全泄露。更麻烦的是现在很多脚本除了写库还要调用大模型做摘要、分类、生成于是又多了一套 API Key 要管。数据库凭证和模型凭证散落各处排查问题时根本不知道哪个 Key 失效了。这篇就聚焦一个具体场景python3 用 pymysql 操作 MySQL 的工程化落地。我会给出一份可复制的 config.toml 配置骨架把数据库连接参数和模型调用凭证统一收口再用 TaoToken 的统一 Key 通道管理调用凭证最后通过一次查询加一次写入验证整套配置真的生效。适合本地脚本、小型服务、个人项目这类不需要上重型配置中心的场景。核心检索词先摆出来python3 怎么用 pymysql 连 mysql、config.toml 怎么写数据库配置、TaoToken 统一 Key 怎么接入。下面按可跟做的步骤走。2. 前置准备pymysql 安装与 TaoToken 统一 Key先说依赖。pymysql 的安装没什么坑一条命令pip install pymysql如果你要用 TOML 解析python3.11 及以上自带 tomllib低版本装一个pip install tomli读取时做个兼容即可后面代码里会体现。接下来是凭证管理这块。数据库这边我建议单独建一个只对业务库有权限的账号别用 root 跑脚本。模型调用这边如果你只是偶尔调一下直接用官方 Key 也行但如果脚本里要频繁调用、或者多个脚本共用一套凭证用 TaoToken 的统一 Key 会省事很多。它的思路是把调用凭证集中管理脚本里只认一个 Key换模型、换通道都不用改业务代码。TaoToken 的 API 地址是https://taotoken.net/api控制台在https://taotoken.net/consoleAPI Key 在https://taotoken.net/api-keys生成。注意 API 地址不要加多余参数直接用它给的 base 就行。生成 Key 之后把它和数据库密码一起放进 config.toml但 config.toml 本身要加进 .gitignore这点后面会强调。提示数据库账号和模型 Key 都属于敏感信息配置文件不要提交到公开仓库。可以放一份 config.example.toml 做模板真实配置本地保留。3. 可复制的 config.toml 配置骨架下面这份骨架是我在多个小脚本里沉淀下来的分三块database、llm、app。database 放 pymysql 需要的连接参数llm 放 TaoToken 的 Key 和 base_urlapp 放一些业务开关。# config.toml [app] name trade-sync debug true log_level INFO [database] host 127.0.0.1 port 3306 user app_user password your_db_password db test charset utf8mb4 connect_timeout 5 read_timeout 10 write_timeout 10 [llm] provider taotoken base_url https://taotoken.net/api api_key sk-your-taotoken-key model gpt-4o-mini timeout 30几个参数值得说明。charset 用 utf8mb4 而不是 utf8因为 MySQL 的 utf8 其实是阉割版存 emoji 或某些生僻字会出问题utf8mb4 才是完整的四字节 UTF-8。connect_timeout 和读写超时建议显式设置默认值在弱网或数据库压力大时容易让脚本卡死。llm 段的 base_url 指向 TaoToken 的 API 入口api_key 就是你在控制台生成的那串。读取配置的代码要兼容 tomllib 和 tomliimport sys if sys.version_info (3, 11): import tomllib else: import tomli as tomllib def load_config(pathconfig.toml): with open(path, rb) as f: return tomllib.load(f) CONFIG load_config()这样一份配置数据库和模型凭证都在同一个文件里但逻辑上分开了。换数据库只动 database 段换模型只动 llm 段互不影响。4. pymysql 连接封装与增删改查实操有了配置接下来封装连接。直接每次 Connect 也行但小服务里更推荐用一个获取连接的函数方便统一加参数。import pymysql from pymysql.cursors import DictCursor def get_connection(): db CONFIG[database] return pymysql.connect( hostdb[host], portdb[port], userdb[user], passworddb[password], databasedb[db], charsetdb[charset], connect_timeoutdb[connect_timeout], read_timeoutdb[read_timeout], write_timeoutdb[write_timeout], cursorclassDictCursor, autocommitFalse, )注意这里用了 DictCursor查询结果直接是字典比默认的元组好用。autocommit 设为 False写操作手动 commit事务边界更清晰。先建一张测试表方便后面验证CREATE TABLE IF NOT EXISTS trade ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(64) NOT NULL, account VARCHAR(32) NOT NULL, saving DECIMAL(12,2) DEFAULT 0, expend DECIMAL(12,2) DEFAULT 0, income DECIMAL(12,2) DEFAULT 0, UNIQUE KEY uk_account (account) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;插入数据用参数化别用字符串拼接。原始 excerpt 里那种sql % data的写法有 SQL 注入风险工程里要避免def insert_trade(name, account, saving): conn get_connection() try: with conn.cursor() as cursor: sql INSERT INTO trade (name, account, saving) VALUES (%s, %s, %s) cursor.execute(sql, (name, account, saving)) conn.commit() print(成功插入, cursor.rowcount, 条数据) except Exception as e: conn.rollback() print(插入失败, e) finally: conn.close()查询和更新同理参数化占位符统一用 %spymysql 会自动处理类型转义。事务处理那段把多条 SQL 放在 try 里异常时 rollback正常时 commitdef transfer(account): conn get_connection() try: with conn.cursor() as cursor: cursor.execute(UPDATE trade SET saving saving 1000 WHERE account %s, (account,)) cursor.execute(UPDATE trade SET expend expend 1000 WHERE account %s, (account,)) cursor.execute(UPDATE trade SET income income 2000 WHERE account %s, (account,)) conn.commit() print(事务处理成功) except Exception as e: conn.rollback() print(事务处理失败, e) finally: conn.close()这里有个细节cursor.rowcount在 with 块外访问时游标可能已经关闭所以打印行数最好在 with 内部完成或者提前存到变量里。5. 用 TaoToken 统一 Key 调用模型做数据摘要数据库写完了现在把模型调用接进来。场景很实际比如 trade 表里存了一堆交易记录你想让模型生成一段自然语言摘要或者对某条记录做分类。用 TaoToken 的 base_url 加统一 Key代码里只认配置不写死任何厂商细节。import json import urllib.request def summarize_trade(rows): llm CONFIG[llm] prompt 请用一句话总结以下交易记录\n json.dumps(rows, ensure_asciiFalse) payload { model: llm[model], messages: [{role: user, content: prompt}], temperature: 0.3, } req urllib.request.Request( llm[base_url].rstrip(/) /v1/chat/completions, datajson.dumps(payload).encode(utf-8), headers{ Content-Type: application/json, Authorization: Bearer llm[api_key], }, methodPOST, ) with urllib.request.urlopen(req, timeoutllm[timeout]) as resp: result json.loads(resp.read().decode(utf-8)) return result[choices][0][message][content]这里用标准库 urllib 是为了少装依赖你换成 requests 或 openai SDK 都行关键是 base_url 和 api_key 从配置读。TaoToken 的 API 入口是https://taotoken.net/api拼接路径时注意别重复斜杠。如果你更习惯用 OpenAI 兼容的 SDK可以这样from openai import OpenAI client OpenAI( base_urlCONFIG[llm][base_url] /v1, api_keyCONFIG[llm][api_key], ) resp client.chat.completions.create( modelCONFIG[llm][model], messages[{role: user, content: 你好}], ) print(resp.choices[0].message.content)两种方式都指向同一个统一 Key换模型只改 config.toml 里的 model 字段业务代码不动。6. 验证请求一次查询加一次写入确认配置生效配置写完不验证等于没写。下面这段脚本把数据库和模型两条链路都跑一遍。def verify(): # 1. 写入一条测试数据 insert_trade(测试用户, 13900000000, 5000.00) # 2. 查询刚写入的数据 conn get_connection() try: with conn.cursor() as cursor: cursor.execute( SELECT name, saving FROM trade WHERE account %s, (13900000000,), ) rows cursor.fetchall() for row in rows: print(Name:%s Saving:%.2f % (row[name], row[saving])) print(共查找出, cursor.rowcount, 条数据) finally: conn.close() # 3. 调用模型做摘要 summary summarize_trade(rows) print(模型摘要:, summary) if __name__ __main__: verify()预期输出类似先打印「成功插入 1 条数据」再打印查询到的用户名和金额最后打印模型返回的一句话摘要。如果数据库段报连接错误检查 host、port、user、password如果模型段报 401检查 api_key 是否复制完整如果报 404检查 base_url 拼接路径。实测下来这套结构跑通之后后面加表、加模型调用都只是往 config.toml 里加字段的事业务代码几乎不用动。7. 本篇常见错排查报错 1pymysql.err.OperationalError: (2003, Cant connect to MySQL server)先确认 MySQL 服务在跑再确认 host 和 port。如果数据库在容器里127.0.0.1 可能不通要换成容器网络里的地址。connect_timeout 设太小也会误报弱网环境调到 10 秒试试。报错 2pymysql.err.OperationalError: (1045, Access denied for user)账号密码不对或者该账号没有从当前 host 连接的权限。MySQL 的 user 表里 host 字段如果是 localhost用 127.0.0.1 连可能被拒需要单独授权。报错 3UnicodeEncodeError或中文乱码charset 没设成 utf8mb4或者建表时用了 utf8。两边都要统一。另外连接参数里 charset 和建表语句的字符集要一致。报错 4模型调用返回 401api_key 没填对或者复制时带了空格。TaoToken 的 Key 在控制台生成后只显示一次丢了就重新生成。注意 Authorization 头是Bearer加 Key中间有空格。报错 5模型调用返回 404base_url 拼接错了。TaoToken 的 API 入口是https://taotoken.net/api如果 SDK 内部会自动加/v1你就不要再手动加一遍。用 urllib 手写时路径是/api/v1/chat/completions。报错 6cursor.rowcount打印为 -1某些查询或游标类型下 rowcount 不可靠尤其是 SELECT 之后。要拿准确行数用len(cursor.fetchall())更稳。报错 7事务没生效数据没写进去检查 autocommit 是否为 False以及有没有在异常分支里 rollback 后忘了重新 commit。另外 InnoDB 才支持事务MyISAM 不支持。8. 凭证收口之后脚本才真正可维护把数据库连接和模型调用都收进 config.toml再用 TaoToken 的统一 Key 管住模型凭证最大的好处是排查问题时路径清晰。数据库连不上就看 database 段模型报错就看 llm 段不用在几百行代码里翻 Key。如果你后面要把脚本跑成长期服务或者多个脚本共用一套凭证建议去控制台把 Key 按用途分开管理再配合 Coding Plan 这类长期方案做额度规划。接入文档里有更细的参数说明遇到拼接路径或鉴权问题时对着查一遍比盲试快得多。