1. 从一段能跑但不敢改的脚本说起Python 存储数据到 MySQL是很多人写脚本时绕不开的一步爬虫抓完数据要落库、定时任务跑完要写结果、AI 辅助生成的代码要接上真实数据库。但真正落到工程里问题往往不在INSERT语句本身而在配置散落各处、密钥硬编码、换台机器就跑不起来。我见过太多项目pymysql.connect(hostlocalhost, userroot, password123456)这行代码在十几个文件里复制粘贴改一次密码要全局搜索替换稍不留神就漏掉一处。这篇内容聚焦 Python 脚本向 MySQL 写入数据的工程化落地同时把 AI 辅助开发工具的接入配置一起收进来。核心交付物是一份可复制的settings.json配置骨架配合 TaoToken 统一 Key 通道把数据库连接参数和模型调用凭证集中管理。适合谁看正在写数据存储脚本、希望把配置从代码里剥离出来的 Python 开发者以及想用 AI 工具辅助写 SQL、生成建表语句、排查连接报错但不想在每个工具里重复填 Key 的人。整篇会按先搭配置骨架、再写可运行脚本、最后验证链路的顺序推进每一步都有完整命令和参数说明你可以直接跟着敲。数据库部分用本地 MySQL 演示模型通道部分用 TaoToken 的 API 地址两者在settings.json里各占一块互不干扰又统一管理。2. 前置准备TaoToken 统一 Key 与本地 MySQL 环境2.1 为什么要把 Key 和数据库配置放一起管工程化落地的一个基本原则是配置与代码分离。数据库的 host、port、user、password 是配置模型通道的 API Key、base_url 也是配置。它们都属于环境相关、不该进版本库的信息。把这两类配置统一收进一个settings.json好处很直接本地开发、测试、生产各用一份文件代码一行不改密钥泄露风险集中可控新同事拉下代码照着模板填一份配置就能跑。TaoToken 在这里扮演的是统一 Key 通道的角色。它提供一个兼容常见模型调用格式的 API 入口你申请一次 Key就能在多个 AI 辅助开发工具里复用不用每个工具单独注册、单独管密钥。对写数据存储脚本的人来说实际价值是当你让 AI 帮你生成建表 SQL、优化批量插入语句、解释pymysql报错时工具背后走的是同一个通道配置一次即可。2.2 获取 Key 与确认接入信息先到 TaoToken 控制台创建 API Key。入口在控制台的 API Keys 页面创建后复制那串以sk-开头的字符串注意它只完整显示一次关掉页面就看不到了建议先粘到临时文本里。控制台与 Key 管理https://taotoken.net/console?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteAPI Keys 页面https://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteAPI 的基础地址是https://taotoken.net/api这个地址在配置里会作为base_url使用。如果你用的是 Claude Code 这类编码工具对应的接入说明在文档里有单独章节配置方式略有不同但 Key 是同一个。2.3 本地 MySQL 与 Python 依赖本地 MySQL 假设运行在localhost:3306用户root密码按你自己的实际设置来。Python 侧只需要一个驱动包pip install pymysql如果你打算用 SQLAlchemy 做 ORM 层再补一个pip install sqlalchemy验证 MySQL 是否正常mysql -h 127.0.0.1 -P 3306 -u root -p -e SELECT VERSION();能打印出版本号就说明数据库通了。这一步别跳过后面脚本连不上时先确认是数据库问题还是配置问题能省很多排查时间。3. 可复制的 settings.json 配置骨架3.1 文件结构与字段说明在项目根目录建一个config文件夹里面放settings.json。骨架长这样{ database: { host: 127.0.0.1, port: 3306, user: root, password: your_mysql_password, charset: utf8mb4, database: test_db }, taotoken: { base_url: https://taotoken.net/api, api_key: sk-your-key-here, model: claude-sonnet-4-20250514, timeout: 60 }, app: { batch_size: 500, log_level: INFO } }字段含义对照字段作用注意事项database.hostMySQL 地址本地用 127.0.0.1别用 localhost 避免 socket 歧义database.charset字符集统一 utf8mb4中文和 emoji 都不会乱码database.database默认库名脚本启动时若库不存在可自动创建taotoken.base_urlAPI 基础地址固定为 https://taotoken.net/apitaotoken.api_key统一 Key从控制台复制勿提交到 gittaotoken.model默认模型按文档里可用模型名填写app.batch_size批量写入条数影响 executemany 的分批大小注意settings.json一定要写进.gitignore。仓库里只保留一份settings.example.json把敏感值留空别人克隆后复制改名再填。3.2 配置加载模块写一个config_loader.py把 JSON 读进来并做基本校验import json import os CONFIG_PATH os.path.join(os.path.dirname(__file__), config, settings.json) def load_settings(pathCONFIG_PATH): if not os.path.exists(path): raise FileNotFoundError(f配置文件不存在: {path}) with open(path, r, encodingutf-8) as f: cfg json.load(f) required [database, taotoken] for key in required: if key not in cfg: raise KeyError(f配置缺少必要节点: {key}) db cfg[database] for field in [host, port, user, password]: if field not in db: raise KeyError(fdatabase 缺少字段: {field}) return cfg if __name__ __main__: settings load_settings() print(配置加载成功目标库:, settings[database][database])运行python config_loader.py打印出目标库名就说明配置读取链路通了。这一步单独验证是为了把配置问题和数据库问题提前隔离开。3.3 用配置建立数据库连接把连接逻辑封装成函数后续所有脚本复用import pymysql from config_loader import load_settings def get_connection(with_dbTrue): cfg load_settings()[database] kwargs dict( hostcfg[host], portcfg[port], usercfg[user], passwordcfg[password], charsetcfg[charset], ) if with_db: kwargs[database] cfg[database] return pymysql.connect(**kwargs) def ensure_database(): cfg load_settings()[database] conn get_connection(with_dbFalse) try: with conn.cursor() as cur: cur.execute( fCREATE DATABASE IF NOT EXISTS {cfg[database]} fDEFAULT CHARACTER SET {cfg[charset]} ) conn.commit() finally: conn.close()ensure_database()做的是幂等创建库已存在就跳过不存在才建。这样脚本在任何一台新机器上首次运行都不会因为库不存在直接崩掉。4. 建表、插入与批量写入的完整脚本4.1 建表语句与字段设计接着写storage.py先建一张测试表from config_loader import load_settings from db import get_connection, ensure_database def create_table(): ensure_database() conn get_connection() sql CREATE TABLE IF NOT EXISTS test_table ( id VARCHAR(64) NOT NULL, name VARCHAR(128) NOT NULL, age INT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; try: with conn.cursor() as cur: cur.execute(sql) conn.commit() print(建表完成: test_table) finally: conn.close()相比最简版本这里加了created_at时间戳字段和ENGINEInnoDB。InnoDB 支持事务后面批量插入失败回滚才有意义时间戳字段在排查数据什么时候写进来的时特别有用。4.2 单条插入与事务处理单条插入封装成函数把异常处理写清楚def insert_one(record): conn get_connection() sql INSERT INTO test_table (id, name, age) VALUES (%s, %s, %s) try: with conn.cursor() as cur: cur.execute(sql, record) conn.commit() print(f插入成功: {record}) return True except pymysql.err.IntegrityError as e: conn.rollback() print(f主键冲突跳过: {record}, 原因: {e}) return False except Exception as e: conn.rollback() print(f插入失败: {record}, 原因: {e}) return False finally: conn.close()这里把IntegrityError单独拎出来处理是因为批量写数据时主键重复是最常见的场景直接rollback并跳过比整个脚本崩掉更合理。4.3 批量写入与分批控制数据量一上来逐条execute会慢得离谱。用executemany配合batch_size分批def insert_batch(records): cfg load_settings() batch_size cfg[app][batch_size] conn get_connection() sql INSERT INTO test_table (id, name, age) VALUES (%s, %s, %s) total 0 try: with conn.cursor() as cur: for i in range(0, len(records), batch_size): chunk records[i:i batch_size] cur.executemany(sql, chunk) conn.commit() total len(chunk) print(f已写入 {total}/{len(records)} 条) except Exception as e: conn.rollback() print(f批量写入中断已回滚当前批次: {e}) finally: conn.close() return total分批提交的好处是单批失败只回滚那一批前面成功的批次不受影响。batch_size放在配置里不同环境可以调本地测试设小一点方便观察生产环境设大一点减少提交次数。4.4 组装运行入口if __name__ __main__: create_table() insert_one((20120001, Bob, 20)) data [(f2024{i:04d}, fuser_{i}, 18 i % 30) for i in range(1, 2001)] written insert_batch(data) print(f批量写入完成共 {written} 条)跑一遍python storage.py你会看到建表、单条插入、分批进度依次打印。2000 条数据在本地通常一两秒内完成。5. 连通性验证与成功结果确认5.1 数据库侧验证脚本跑完后直接查库确认mysql -h 127.0.0.1 -P 3306 -u root -p test_db -e SELECT COUNT(*) AS total FROM test_table;预期输出类似------- | total | ------- | 2001 | -------2001 1 条单插 2000 条批量。数字对得上说明写入链路完整。5.2 模型通道连通性验证配置里的taotoken节点是否可用单独验证一次。用 curl 发一个最小请求curl -X POST https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer sk-your-key-here \ -H Content-Type: application/json \ -d { model: claude-sonnet-4-20250514, messages: [{role: user, content: 用一句话说明 MySQL 批量插入为什么比逐条快}] }返回体里带choices字段和正常文本内容就说明 Key 和通道都通了。如果返回 401检查 Key 是否复制完整返回 404检查base_url有没有多写或少写路径段。你也可以直接在模型对话页面手动测一句确认账号状态正常模型对话https://taotoken.net/chat?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite5.3 把 AI 辅助接进开发流程通道验证通过后实际开发中可以让 AI 帮你做几件事根据表结构生成CREATE TABLE语句、把慢查询改写成批量版本、解释pymysql的报错码。比如你拿到pymysql.err.OperationalError: (2003, Cant connect to MySQL server)把完整报错贴给模型它能直接指出是端口不通还是 host 写错。如果你长期用编码类工具建议走 Coding Plan配置一次就能在编辑器里持续用Coding Planhttps://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite6. 本篇常见报错排查6.1 连接类报错Cant connect to MySQL server on 127.0.0.1通常三种原因MySQL 服务没启动、端口不是 3306、防火墙拦截。先systemctl status mysql看服务状态再netstat -tlnp | grep 3306确认端口监听。Access denied for user rootlocalhost是密码或权限问题。注意settings.json里的密码和 MySQL 实际密码要一致改过密码后记得同步配置文件。6.2 编码类报错Incorrect string value基本是字符集不统一。检查三处建表语句的CHARSETutf8mb4、连接参数里的charset、以及 MySQL 服务端的character_set_server。三处都对齐到 utf8mb4 就不会再出现。6.3 配置读取类报错FileNotFoundError: 配置文件不存在说明路径不对。config_loader.py里用的是相对脚本文件的路径如果你从别的目录运行脚本要么用绝对路径要么确保工作目录正确。KeyError: database 缺少字段是 JSON 里漏了必填项。对照第 3.1 节的字段表逐个核对注意 JSON 不允许尾随逗号多一个逗号就会解析失败。6.4 批量写入性能问题如果 2000 条写了十几秒检查是不是每条都单独commit。正确做法是每批executemany后提交一次。另外确认表引擎是 InnoDB 且主键有索引主键冲突检查走索引会快很多。6.5 Key 相关报错401 Unauthorized优先检查 Key 是否完整、有没有多余空格。403一般是权限或额度问题到控制台确认账号状态。404检查base_url是否为https://taotoken.net/api不要自己拼接多余路径。排查接入问题时接入文档里的错误码对照表最省事接入文档https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite7. 把配置骨架用起来这套骨架的价值在于可复用。换项目时config_loader.py和db.py直接拷过去只改settings.json里的库名和表结构相关代码。密钥管理上本地开发用文件CI 环境用环境变量覆盖config_loader.py里加一层os.environ.get优先级判断就能兼顾。如果你还没创建 Key从 API Keys 页面开始API Keyshttps://taotoken.net/api-keys?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewrite实际用下来把数据库配置和模型通道配置收进同一个文件最大的收益不是省了几行代码而是排查问题时只需要看一个地方。连接失败、Key 失效、字符集不对全都在settings.json里有迹可循。
