1. 从一次真实报错说起JSON 根节点为什么会被“跟了尾巴”Invalid JSON text: The document root must not be followed by other values这个报错字面意思是“JSON 文档根节点后面不能再跟其他值”。翻译成人话MySQL 在解析你写进去的那一列 JSON 时发现本该只有一个根对象或根数组的地方后面又冒出了第二个值。比如{a:1}{b:2}、{a:1} {b:2}、{a:1}\n{b:2}甚至{a:1}null都会触发它。这个错在 Python 调用 API 拿数据再写 MySQL 的场景里特别常见因为链路是“API 返回 JSON → Python 解析 → 拼装 → 写库”中间任何一步把多个 JSON 值拼在一起或者把响应包裹层没剥干净就会在写库那一刻炸掉。它跟 SQL 语法无关跟字段类型有关目标列是JSON类型MySQL 就会严格校验根节点唯一性。适合谁看正在用 Python 做数据管道、把第三方 API 的 JSON 落到 MySQL JSON 列、并且被这个报错卡住的同学。下面我会先讲清楚报错根因再给一套可复制的配置骨架最后用最小请求验证“根节点唯一性”把排查动作固定下来。2. 先定位根因多值拼接与响应包裹是两大元凶2.1 多值拼接循环里把多个 JSON 塞进一个字段最常见的写法是这样API 分页返回你在循环里把每次的response.json()直接或append到同一个字符串最后一次性写库。如果中间没有做数组包裹字符串就变成了{...}{...}根节点后面跟了第二个值。# 错误示范多个 JSON 对象直接拼接 raw for page in range(1, 4): resp requests.get(API_URL, params{page: page}, headersheaders) raw resp.text # 这里埋雷{...}{...}{...} cursor.execute( INSERT INTO api_raw (payload) VALUES (%s), (raw,) ) # 触发 Invalid JSON text: The document root must not be followed by other values正确做法是先把每个对象收进 Python 列表再json.dumps一次让根节点变成唯一的数组import json items [] for page in range(1, 4): resp requests.get(API_URL, params{page: page}, headersheaders) items.append(resp.json()) payload json.dumps(items, ensure_asciiFalse) # 根节点是 [ ... ]唯一 cursor.execute( INSERT INTO api_raw (payload) VALUES (%s), (payload,) )2.2 响应包裹把整个响应体连同外层一起写进去有些 API 返回的是{code:0,data:{...},msg:ok}你只想存data结果把整个响应体写进去本身没问题但如果你的代码里又手动拼了一层比如json.dumps(resp.json()) json.dumps(extra)就会变成两个根值。还有一种隐蔽情况响应是流式的resp.text里带了 SSE 的data:前缀或多行事件直接写库也会报同样的错。排查时先打印repr(payload[:200])看根节点后面有没有多余的{、[、null或换行后的第二个值。这一步比盯着报错猜要快得多。3. TaoToken 前置统一 Key 通道让请求侧先干净在排查写库问题之前我习惯先把“请求侧”固定下来避免变量太多。TaoToken 在这里的作用是提供一个统一的 API Key 通道把模型对话、编码计划、控制台管理这些入口收敛到一套凭证上这样你在 Python 里调 API 时header 和 base_url 是稳定的不会因为换了个服务就改一堆配置。官网入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 基地址https://taotoken.net/api几个常用 deep link按需取用模型对话https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewriteCoding Planhttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite控制台https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewriteAPI Keyshttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite接入文档https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewriteClaudeCodeAnthropichttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecodeutm_campaignrewrite注意Key 只放在环境变量或本地配置文件里不要硬编码进提交到仓库的脚本。下面给的config.toml和settings.json骨架都走“读环境变量”的方式。4. 可复制配置骨架config.toml 与 settings.json4.1 config.toml请求侧与数据库侧分离# config.toml [api] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY # 从环境变量读取不写明文 timeout 30 max_retries 3 [mysql] host 127.0.0.1 port 3306 user app_writer password_env MYSQL_PASSWORD database pipeline charset utf8mb4 json_column payload [pipeline] # 分页抓取后统一 json.dumps根节点唯一 wrap_as_array true strip_response_envelope true # 只取 data 字段4.2 settings.json给不读 toml 的组件用{ api: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, timeout: 30 }, mysql: { host: 127.0.0.1, port: 3306, user: app_writer, password_env: MYSQL_PASSWORD, database: pipeline, charset: utf8mb4 }, pipeline: { wrap_as_array: true, strip_response_envelope: true } }4.3 读取配置并组装请求import os import json import tomllib import requests import pymysql with open(config.toml, rb) as f: cfg tomllib.load(f) API_KEY os.environ[cfg[api][api_key_env]] HEADERS { Authorization: fBearer {API_KEY}, Content-Type: application/json, } def fetch_page(page: int) - dict: resp requests.get( f{cfg[api][base_url]}/v1/chat/completions, headersHEADERS, json{model: gpt-4o-mini, messages: [{role: user, content: fpage {page}}]}, timeoutcfg[api][timeout], ) resp.raise_for_status() return resp.json()这里base_url指向 TaoToken 的统一通道Key 从环境变量注入。请求侧干净了接下来只盯“写库前的 payload 是不是唯一根节点”。5. 验证请求用最小动作确认 JSON 根节点唯一5.1 写库前先做一次“根节点唯一性”校验不要等 MySQL 报错先在 Python 侧用json.loads做一次严格解析。json.loads对多值拼接会直接抛JSONDecodeError比数据库报错更早、更明确。def assert_single_root(payload: str) - None: 确认 payload 只有一个 JSON 根节点 try: json.loads(payload) except json.JSONDecodeError as e: raise ValueError(fpayload 不是唯一根节点: {e} | 前 200 字符: {payload[:200]!r})5.2 最小请求抓一页、剥包裹、写库def build_payload(pages: int 2) - str: items [] for p in range(1, pages 1): body fetch_page(p) if cfg[pipeline][strip_response_envelope]: body body.get(data, body) # 剥掉外层包裹 items.append(body) if cfg[pipeline][wrap_as_array]: payload json.dumps(items, ensure_asciiFalse) # 根节点唯一[...] else: payload json.dumps(items[0], ensure_asciiFalse) assert_single_root(payload) return payload def write_to_mysql(payload: str) - None: conn pymysql.connect( hostcfg[mysql][host], portcfg[mysql][port], usercfg[mysql][user], passwordos.environ[cfg[mysql][password_env]], databasecfg[mysql][database], charsetcfg[mysql][charset], ) with conn.cursor() as cur: cur.execute( INSERT INTO api_raw (payload) VALUES (%s), (payload,), ) conn.commit() conn.close() if __name__ __main__: write_to_mysql(build_payload())5.3 成功结果长什么样跑通后SELECT JSON_VALID(payload), JSON_TYPE(payload) FROM api_raw ORDER BY id DESC LIMIT 1;应该返回1和ARRAY或OBJECT。如果JSON_VALID返回0说明写进去的仍然不是合法 JSON回到第 5.1 步看assert_single_root有没有被绕过。提示MySQL 的 JSON 列在插入时会自动校验JSON_VALID只是事后复核。真正的拦截点应该放在 Python 侧越早越好。6. 本篇常见错排查清单6.1 报错依旧检查是不是绕过了校验有些人把assert_single_root写在build_payload里但实际写库用的是另一个函数校验没走到。排查方法在cursor.execute前打印type(payload)和payload[:120]确认它确实是str且只有一个根。6.2 参数化写法踩坑(code)不是元组excerpt 里提到的那个经典坑值得单独说cursor.execute(... where code%s, (code))里的(code)是普通括号不是元组PyMySQL 会把它当成单个值而不是参数序列轻则报参数数量不匹配重则把值拼进 SQL。正确写法是(code,)注意那个逗号。# 错误 cursor.execute(SELECT * FROM table_code WHERE code%s, (code)) # 正确 cursor.execute(SELECT * FROM table_code WHERE code%s, (code,))6.3 响应里带 BOM 或前后空白有些 API 返回的resp.text开头带\ufeff或者结尾有换行。json.loads能容忍部分空白但 MySQL 的 JSON 解析更严格。写库前用payload.strip().lstrip(\ufeff)清一遍。6.4 字段类型不是 JSON 却报 JSON 错如果目标列是TEXT而不是JSONMySQL 不会做 JSON 校验这个错就不会出现。反过来说一旦看到这个错先确认列类型SHOW COLUMNS FROM api_raw LIKE payload;Type应该是json。6.5 分页循环里resp.json()被调用两次resp.json()在某些实现里会消耗流第二次调用可能拿到空或异常。养成习惯body resp.json()只调一次后面都用body。7. 把通道和校验固定下来下次直接复用这套排查动作的核心就两件事请求侧用 TaoToken 统一 Key 通道保证 header 和 base_url 稳定写库侧用assert_single_root做根节点唯一性校验把 MySQL 的报错提前到 Python 侧。配置骨架可以直接抄config.toml和settings.json把api_key_env和password_env换成你自己的环境变量名即可。如果你还在接模型对话或编码计划建议从 API Keys 页面拿 Key再对照接入文档确认 base_url 和路径长期做编码和 Agent 的可以看 Coding Plan 的入口。排障和接入相关的入口统一放在 API Keys 和接入文档验证模型效果走模型对话别只记首页。最后留一个我常用的自检命令写库前跑一次比事后查日志快python -c import json,sys; json.loads(open(payload.json,encodingutf-8).read()); print(single root ok)根节点唯一了Invalid JSON text自然就不会再来找你。
Python API 写入 MySQL 报 Invalid JSON text？用 TaoToken 统一 Key 通道排查配置骨架
1. 从一次真实报错说起JSON 根节点为什么会被“跟了尾巴”Invalid JSON text: The document root must not be followed by other values这个报错字面意思是“JSON 文档根节点后面不能再跟其他值”。翻译成人话MySQL 在解析你写进去的那一列 JSON 时发现本该只有一个根对象或根数组的地方后面又冒出了第二个值。比如{a:1}{b:2}、{a:1} {b:2}、{a:1}\n{b:2}甚至{a:1}null都会触发它。这个错在 Python 调用 API 拿数据再写 MySQL 的场景里特别常见因为链路是“API 返回 JSON → Python 解析 → 拼装 → 写库”中间任何一步把多个 JSON 值拼在一起或者把响应包裹层没剥干净就会在写库那一刻炸掉。它跟 SQL 语法无关跟字段类型有关目标列是JSON类型MySQL 就会严格校验根节点唯一性。适合谁看正在用 Python 做数据管道、把第三方 API 的 JSON 落到 MySQL JSON 列、并且被这个报错卡住的同学。下面我会先讲清楚报错根因再给一套可复制的配置骨架最后用最小请求验证“根节点唯一性”把排查动作固定下来。2. 先定位根因多值拼接与响应包裹是两大元凶2.1 多值拼接循环里把多个 JSON 塞进一个字段最常见的写法是这样API 分页返回你在循环里把每次的response.json()直接或append到同一个字符串最后一次性写库。如果中间没有做数组包裹字符串就变成了{...}{...}根节点后面跟了第二个值。# 错误示范多个 JSON 对象直接拼接 raw for page in range(1, 4): resp requests.get(API_URL, params{page: page}, headersheaders) raw resp.text # 这里埋雷{...}{...}{...} cursor.execute( INSERT INTO api_raw (payload) VALUES (%s), (raw,) ) # 触发 Invalid JSON text: The document root must not be followed by other values正确做法是先把每个对象收进 Python 列表再json.dumps一次让根节点变成唯一的数组import json items [] for page in range(1, 4): resp requests.get(API_URL, params{page: page}, headersheaders) items.append(resp.json()) payload json.dumps(items, ensure_asciiFalse) # 根节点是 [ ... ]唯一 cursor.execute( INSERT INTO api_raw (payload) VALUES (%s), (payload,) )2.2 响应包裹把整个响应体连同外层一起写进去有些 API 返回的是{code:0,data:{...},msg:ok}你只想存data结果把整个响应体写进去本身没问题但如果你的代码里又手动拼了一层比如json.dumps(resp.json()) json.dumps(extra)就会变成两个根值。还有一种隐蔽情况响应是流式的resp.text里带了 SSE 的data:前缀或多行事件直接写库也会报同样的错。排查时先打印repr(payload[:200])看根节点后面有没有多余的{、[、null或换行后的第二个值。这一步比盯着报错猜要快得多。3. TaoToken 前置统一 Key 通道让请求侧先干净在排查写库问题之前我习惯先把“请求侧”固定下来避免变量太多。TaoToken 在这里的作用是提供一个统一的 API Key 通道把模型对话、编码计划、控制台管理这些入口收敛到一套凭证上这样你在 Python 里调 API 时header 和 base_url 是稳定的不会因为换了个服务就改一堆配置。官网入口https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_contentAPI 基地址https://taotoken.net/api几个常用 deep link按需取用模型对话https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentmodel_chatutm_campaignrewriteCoding Planhttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentcoding_planutm_campaignrewrite控制台https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentconsoleutm_campaignrewriteAPI Keyshttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite接入文档https://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewriteClaudeCodeAnthropichttps://taotoken.net/api?utm_sourcetaotoken_aicg_blog_endutm_contentclaudecodeutm_campaignrewrite注意Key 只放在环境变量或本地配置文件里不要硬编码进提交到仓库的脚本。下面给的config.toml和settings.json骨架都走“读环境变量”的方式。4. 可复制配置骨架config.toml 与 settings.json4.1 config.toml请求侧与数据库侧分离# config.toml [api] base_url https://taotoken.net/api api_key_env TAOTOKEN_API_KEY # 从环境变量读取不写明文 timeout 30 max_retries 3 [mysql] host 127.0.0.1 port 3306 user app_writer password_env MYSQL_PASSWORD database pipeline charset utf8mb4 json_column payload [pipeline] # 分页抓取后统一 json.dumps根节点唯一 wrap_as_array true strip_response_envelope true # 只取 data 字段4.2 settings.json给不读 toml 的组件用{ api: { base_url: https://taotoken.net/api, api_key_env: TAOTOKEN_API_KEY, timeout: 30 }, mysql: { host: 127.0.0.1, port: 3306, user: app_writer, password_env: MYSQL_PASSWORD, database: pipeline, charset: utf8mb4 }, pipeline: { wrap_as_array: true, strip_response_envelope: true } }4.3 读取配置并组装请求import os import json import tomllib import requests import pymysql with open(config.toml, rb) as f: cfg tomllib.load(f) API_KEY os.environ[cfg[api][api_key_env]] HEADERS { Authorization: fBearer {API_KEY}, Content-Type: application/json, } def fetch_page(page: int) - dict: resp requests.get( f{cfg[api][base_url]}/v1/chat/completions, headersHEADERS, json{model: gpt-4o-mini, messages: [{role: user, content: fpage {page}}]}, timeoutcfg[api][timeout], ) resp.raise_for_status() return resp.json()这里base_url指向 TaoToken 的统一通道Key 从环境变量注入。请求侧干净了接下来只盯“写库前的 payload 是不是唯一根节点”。5. 验证请求用最小动作确认 JSON 根节点唯一5.1 写库前先做一次“根节点唯一性”校验不要等 MySQL 报错先在 Python 侧用json.loads做一次严格解析。json.loads对多值拼接会直接抛JSONDecodeError比数据库报错更早、更明确。def assert_single_root(payload: str) - None: 确认 payload 只有一个 JSON 根节点 try: json.loads(payload) except json.JSONDecodeError as e: raise ValueError(fpayload 不是唯一根节点: {e} | 前 200 字符: {payload[:200]!r})5.2 最小请求抓一页、剥包裹、写库def build_payload(pages: int 2) - str: items [] for p in range(1, pages 1): body fetch_page(p) if cfg[pipeline][strip_response_envelope]: body body.get(data, body) # 剥掉外层包裹 items.append(body) if cfg[pipeline][wrap_as_array]: payload json.dumps(items, ensure_asciiFalse) # 根节点唯一[...] else: payload json.dumps(items[0], ensure_asciiFalse) assert_single_root(payload) return payload def write_to_mysql(payload: str) - None: conn pymysql.connect( hostcfg[mysql][host], portcfg[mysql][port], usercfg[mysql][user], passwordos.environ[cfg[mysql][password_env]], databasecfg[mysql][database], charsetcfg[mysql][charset], ) with conn.cursor() as cur: cur.execute( INSERT INTO api_raw (payload) VALUES (%s), (payload,), ) conn.commit() conn.close() if __name__ __main__: write_to_mysql(build_payload())5.3 成功结果长什么样跑通后SELECT JSON_VALID(payload), JSON_TYPE(payload) FROM api_raw ORDER BY id DESC LIMIT 1;应该返回1和ARRAY或OBJECT。如果JSON_VALID返回0说明写进去的仍然不是合法 JSON回到第 5.1 步看assert_single_root有没有被绕过。提示MySQL 的 JSON 列在插入时会自动校验JSON_VALID只是事后复核。真正的拦截点应该放在 Python 侧越早越好。6. 本篇常见错排查清单6.1 报错依旧检查是不是绕过了校验有些人把assert_single_root写在build_payload里但实际写库用的是另一个函数校验没走到。排查方法在cursor.execute前打印type(payload)和payload[:120]确认它确实是str且只有一个根。6.2 参数化写法踩坑(code)不是元组excerpt 里提到的那个经典坑值得单独说cursor.execute(... where code%s, (code))里的(code)是普通括号不是元组PyMySQL 会把它当成单个值而不是参数序列轻则报参数数量不匹配重则把值拼进 SQL。正确写法是(code,)注意那个逗号。# 错误 cursor.execute(SELECT * FROM table_code WHERE code%s, (code)) # 正确 cursor.execute(SELECT * FROM table_code WHERE code%s, (code,))6.3 响应里带 BOM 或前后空白有些 API 返回的resp.text开头带\ufeff或者结尾有换行。json.loads能容忍部分空白但 MySQL 的 JSON 解析更严格。写库前用payload.strip().lstrip(\ufeff)清一遍。6.4 字段类型不是 JSON 却报 JSON 错如果目标列是TEXT而不是JSONMySQL 不会做 JSON 校验这个错就不会出现。反过来说一旦看到这个错先确认列类型SHOW COLUMNS FROM api_raw LIKE payload;Type应该是json。6.5 分页循环里resp.json()被调用两次resp.json()在某些实现里会消耗流第二次调用可能拿到空或异常。养成习惯body resp.json()只调一次后面都用body。7. 把通道和校验固定下来下次直接复用这套排查动作的核心就两件事请求侧用 TaoToken 统一 Key 通道保证 header 和 base_url 稳定写库侧用assert_single_root做根节点唯一性校验把 MySQL 的报错提前到 Python 侧。配置骨架可以直接抄config.toml和settings.json把api_key_env和password_env换成你自己的环境变量名即可。如果你还在接模型对话或编码计划建议从 API Keys 页面拿 Key再对照接入文档确认 base_url 和路径长期做编码和 Agent 的可以看 Coding Plan 的入口。排障和接入相关的入口统一放在 API Keys 和接入文档验证模型效果走模型对话别只记首页。最后留一个我常用的自检命令写库前跑一次比事后查日志快python -c import json,sys; json.loads(open(payload.json,encodingutf-8).read()); print(single root ok)根节点唯一了Invalid JSON text自然就不会再来找你。
