企业官网建设流程全解析

1. 为什么 pymysql 的 SQL 日志总是打不全

pymysql 输出 SQL 日志这件事，看起来只是加一行 print，实际做起来坑不少。我在几个数据同步项目里都遇到过类似情况：本地调试时 SQL 能打出来，一上测试环境就只剩个空壳；或者日志里只有 SELECT，INSERT 和 UPDATE 全丢了；再或者参数化查询打出来是一堆 %s，根本看不出真实值。这些问题的根源，通常不在 pymysql 本身，而在于日志挂载的位置和参数拼装的方式。

先说清楚 pymysql 是什么、能做什么、适合谁。pymysql 是纯 Python 实现的 MySQL 客户端库，不依赖 C 扩展，安装即用，适合中小型项目、脚本任务、数据管道这类需要快速连 MySQL 的场景。它本身不提供 SQL 日志能力，但它的 cursor 类是可继承的，这给了我们注入日志的入口。你要做的，是选一个合适的切面，把 execute 和 executemany 两个方法包住，同时把参数拼成可读形式。

常见的错误做法有三种。第一种是直接在业务代码里到处 print，日志散落各处，改起来痛苦。第二种是只重写 execute，忘了 executemany，批量插入的 SQL 全漏掉。第三种是只打 query 字符串，不打 args，遇到参数化查询就抓瞎。我试过在线上排查一个慢查询，日志里只有SELECT * FROM orders WHERE id = %s，完全不知道是哪个 id，最后只能加临时日志重跑，浪费了半天。

正确的思路是：写一个 LoggingCursor，继承 pymysql 的 Cursor 或 DictCursor，在 execute 和 executemany 里先格式化再调用父类方法。格式化时用 pymysql 自带的 mogrify，它能把 query 和 args 拼成最终 SQL，既保留参数化查询的安全性，又能看到真实值。日志输出目标可以是控制台，也可以是文件，甚至同时输出，用 Python 标准 logging 模块管理级别和格式。

还有一个容易被忽略的点：连接池。如果你用 DBUtils 或 SQLAlchemy 的连接池，cursor 的创建方式会变，LoggingCursor 要挂到连接工厂上，否则日志不生效。这部分我在第 3 节会给完整配置。

另外，既然项目里已经在调模型做 SQL 生成或分析，把模型调用也统一到一条 Key 通道上，能省掉多套凭证管理的麻烦。TaoToken 提供的就是这样一个统一入口，兼容 OpenAI 风格的接口，Base URL 换成https://taotoken.net/api即可，模型 ID 按需选。这样你的 SQL 日志和模型调用日志可以放在同一套 logging 配置里，排查问题时时间线对得上。

本节的核心检索词是「pymysql 输出 SQL 日志」，你要记住三件事：日志要挂在 cursor 层，参数要用 mogrify 还原，输出要走 logging 而不是 print。做到这三点，后面配置就是填空。

2. TaoToken 统一 Key 接入的前置准备

在写 pymysql 日志代码之前，先把模型调用的通道理清楚，这样后面 config.toml 里两套配置能放在一起管理。TaoToken 的定位是统一 API 入口，你不需要为每个模型单独申请 Key，一个 Key 走通对话、代码补全、Agent 调用等场景。对 Python 项目来说，最直接的价值是：环境变量少一个，配置文件少一段，切换模型只改一个 Model ID。

前置准备分三步。第一步是拿 Key。访问官网 https://taotoken.net/?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ，注册后在控制台创建 API Key。控制台地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_content=console&utm_campaign=rewrite ，Key 只在创建时显示一次，复制后存到安全的地方。如果你还没决定用哪个模型，可以先到模型对话页面 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_content=models&utm_campaign=rewrite 试一下，确认模型 ID 再写进配置。

第二步是确认 Base URL。TaoToken 的 API 端点是https://taotoken.net/api，注意这个地址不带 UTM 参数，直接写进代码或配置文件即可。如果你用 OpenAI SDK，把base_url指向它；如果你用 requests 裸调，拼接/v1/chat/completions这类路径。Key 放在请求头Authorization: Bearer <你的Key>。

第三步是规划配置结构。我建议把数据库连接和模型调用都放进 config.toml，用两个 section 分开：[database]放 host、port、user、password、db，[llm]放 base_url、api_key、model。这样一份配置文件同时管两件事，部署时只改环境相关的值。api_key 不要硬编码在代码里，用环境变量覆盖，config.toml 里写占位符。

这里要提醒一个安全边界：TaoToken 是合规的 API 聚合入口，不是让你绕过任何限制的工具。你用它来统一管理模型调用凭证，和用其他云服务的 API 网关是一个性质。配置文件里的 Key 要按密钥管理，别提交到 Git。

如果你做的是长期编码或 Agent 类项目，可以了解一下 Coding Plan https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_content=coding-plan&utm_campaign=rewrite ，它针对持续调用场景做了额度规划。接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_content=doc&utm_campaign=rewrite ，里面有各语言的调用示例，Python 部分可以直接参考。

前置准备做完，你手里应该有三样东西：一个可用的 API Key、确认过的 Base URL、一份 config.toml 草稿。下一节开始写可复制的代码。

3. 可复制的 pymysql 日志与 config.toml 配置骨架

这一节是全文的核心，给你一份能直接跑的配置和代码。先看 config.toml 骨架，路径放在项目根目录，文件名就叫 config.toml。

# config.toml [database] host = "127.0.0.1" port = 3306 user = "app_user" password = "your_db_password" db = "app_db" charset = "utf8mb4" # 日志输出：console / file / both log_output = "both" log_file = "logs/sql.log" log_level = "DEBUG" [llm] base_url = "https://taotoken.net/api" api_key = "sk-替换成你的Key" model = "gpt-4o-mini" timeout = 30

注意[llm]里的base_url就是 TaoToken 的 API 地址，api_key从控制台拿，model按你实际用的填。如果你用 Claude Code 或 Cline 这类工具，Base URL、Key、Model ID 三件套的填法是一样的，只是入口在工具的设置里。

接下来是 pymysql 日志的核心代码。新建db_logger.py：

import logging import pymysql from pymysql.cursors import DictCursor logger = logging.getLogger("pymysql.sql") class LoggingCursor(DictCursor): def _log(self, sql, args=None): if args: try: final_sql = self.mogrify(sql, args) except Exception: final_sql = f"{sql} | args={args}" else: final_sql = sql logger.debug("SQL: %s", final_sql) def execute(self, query, args=None): self._log(query, args) return super().execute(query, args) def executemany(self, query, args): self._log(query, args) return super().executemany(query, args)

这段代码的关键点：继承 DictCursor 而不是 Cursor，这样查询结果直接是字典，日志和业务都好用；mogrify是 pymysql 自带方法，能把参数安全地拼进 SQL，不会引入注入风险；execute 和 executemany 都覆盖，批量操作不漏。

然后是连接工厂和 logging 初始化：

import tomllib import logging import pymysql from db_logger import LoggingCursor def load_config(path="config.toml"): with open(path, "rb") as f: return tomllib.load(f) def setup_logging(cfg): handlers = [] if cfg["log_output"] in ("console", "both"): handlers.append(logging.StreamHandler()) if cfg["log_output"] in ("file", "both"): handlers.append(logging.FileHandler(cfg["log_file"], encoding="utf-8")) logging.basicConfig( level=cfg["log_level"], format="%(asctime)s [%(levelname)s] %(name)s - %(message)s", handlers=handlers, ) def get_conn(cfg): db = cfg["database"] return pymysql.connect( host=db["host"], port=db["port"], user=db["user"], password=db["password"], database=db["db"], charset=db["charset"], cursorclass=LoggingCursor, )

注意cursorclass=LoggingCursor这一行，它让所有通过这个连接创建的 cursor 都自动带日志。如果你用连接池，比如 DBUtils 的 PooledDB，把cursorclass传给 PooledDB 的构造参数即可，效果一样。

模型调用部分，用 OpenAI SDK 指向 TaoToken：

from openai import OpenAI def get_llm_client(cfg): llm = cfg["llm"] return OpenAI(base_url=llm["base_url"], api_key=llm["api_key"])

这样数据库和模型两套配置都在 config.toml 里，代码里只读配置，不硬编码。部署到不同环境时，用环境变量覆盖敏感值，比如export DB_PASSWORD=xxx，然后在 load_config 后做一次覆盖。

配置骨架到这里就完整了。下一节验证它是否真的能打出 SQL 日志。

4. 验证请求与成功结果

配置写完了，得跑一次真实查询确认日志生效。我准备了一个最小验证脚本verify.py：

from db_logger import load_config, setup_logging, get_conn, get_llm_client cfg = load_config() setup_logging(cfg) conn = get_conn(cfg) with conn.cursor() as cur: cur.execute("SELECT id, name FROM users WHERE id = %s", (1,)) row = cur.fetchone() print("查询结果:", row) cur.executemany( "INSERT INTO users (name) VALUES (%s)", [("alice",), ("bob",)], ) conn.commit() conn.close() client = get_llm_client(cfg) resp = client.chat.completions.create( model=cfg["llm"]["model"], messages=[{"role": "user", "content": "用一句话说明这条SQL在做什么：SELECT id, name FROM users WHERE id = 1"}], ) print("模型回复:", resp.choices[0].message.content)

运行python verify.py，控制台应该出现类似输出：

2025-01-15 10:23:41,102 [DEBUG] pymysql.sql - SQL: SELECT id, name FROM users WHERE id = 1 查询结果: {'id': 1, 'name': 'alice'} 2025-01-15 10:23:41,105 [DEBUG] pymysql.sql - SQL: INSERT INTO users (name) VALUES ('alice'),('bob') 模型回复: 这条 SQL 从 users 表中查询 id 等于 1 的记录的 id 和 name 字段。

看到SQL:开头的行，说明日志生效了。注意 SELECT 那条，参数%s已经被替换成1，这是 mogrify 的功劳。INSERT 那条，executemany 把两条数据拼成了一条多值插入，日志里能看清实际写入内容。模型回复正常返回，说明 TaoToken 通道也通了。

如果你把log_output改成file，日志会写到logs/sql.log，控制台不再输出 SQL，但查询结果和模型回复照常打印。改成both则两边都有。日志级别设成DEBUG才能看到 SQL，设成INFO会被过滤掉，这是 logging 模块的标准行为。

验证时还要注意一个细节：conn.commit()必须在 executemany 之后调用，否则插入不生效，但日志已经打了，容易误判。我建议在验证脚本里加一句print("受影响行数:", cur.rowcount)，确认写入真的成功。

模型调用这边，如果返回 401，说明 Key 不对或没带上；如果返回 model not found，说明 Model ID 写错了。这两个错误在第 5 节详细说。

验证通过后，你可以把 verify.py 删掉，或者保留作为冒烟测试。生产代码里，把 get_conn 和 get_llm_client 封装成单例或依赖注入，避免每次请求都新建连接。

5. 本篇常见错误排查

这一节对照真实报错，把 pymysql 日志和 TaoToken 接入过程中最容易踩的坑列出来。

报错一：pymysql.err.OperationalError: (2003, "Can't connect to MySQL server on '127.0.0.1'")

这是数据库连不上，和日志无关。检查 config.toml 里的 host、port 是否正确，MySQL 服务是否启动，防火墙是否放行。如果你用 Docker，注意容器内的 127.0.0.1 指向容器本身，要用宿主机的 IP 或服务名。

报错二：日志里只有SQL: SELECT ... %s，参数没替换

说明 mogrify 没生效，通常是 args 传成了 None 或者格式不对。检查你的 execute 调用，参数要用元组或字典，比如cur.execute(sql, (1,))，不能写成cur.execute(sql, 1)。另外，如果你用的是自己写的 Cursor 子类但没继承 DictCursor，mogrify 可能不可用，确认继承链正确。

报错三：AttributeError: 'LoggingCursor' object has no attribute 'mogrify'

mogrify 是 pymysql cursor 的方法，如果你继承的是pymysql.cursors.Cursor但版本太老，可能没有。升级 pymysql 到最新版即可：pip install --upgrade pymysql。另外，mogrify 在连接未建立时调用会报错，确保 cursor 是从已连接的 connection 创建的。

报错四：openai.AuthenticationError: Error code: 401

这是 TaoToken 的 Key 问题。检查 config.toml 里[llm]的 api_key 是否填对，有没有多余空格。如果你用环境变量覆盖，确认变量名和读取逻辑一致。401 也可能是 Key 被禁用或额度用完，到控制台 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_content=api-keys&utm_campaign=rewrite 看一下 Key 状态。

报错五：openai.NotFoundError: Error code: 404或model not found

Model ID 写错了。TaoToken 的模型 ID 和官方一致，比如gpt-4o-mini、claude-3-5-sonnet-20241022。到模型对话页面确认可用 ID，别自己拼。另外，Base URL 要写https://taotoken.net/api，不要多加/v1，SDK 会自己拼路径。

报错六：local proxy failed或连接超时

如果你本地配了网络代理，OpenAI SDK 可能走了代理导致连不上。检查环境变量HTTP_PROXY、HTTPS_PROXY，临时 unset 掉再试。TaoToken 的 API 是直连的，不需要额外代理配置。

报错七：日志文件不生成

检查log_file的目录是否存在，logging.FileHandler 不会自动创建目录。在 setup_logging 里加os.makedirs(os.path.dirname(cfg["log_file"]), exist_ok=True)。另外，文件权限要可写，容器部署时注意挂载卷的权限。

报错八：RuntimeError: Working outside of application context

如果你在 Flask 或 Django 里用这套配置，logging 初始化要放在应用启动时，不能放在请求处理函数里。连接也要用连接池管理，别每个请求新建。

排查顺序建议：先确认数据库能连上，再确认日志级别是 DEBUG，再确认 cursorclass 挂对了，最后确认模型 Key 和 Model ID。大部分问题出在前两步。

6. 把 SQL 日志和模型调用收进一条通道

走到这里，你应该已经有一份能跑的 config.toml、一个带日志的 LoggingCursor、一次成功的验证输出。剩下的就是把它们放进真实项目。

我的做法是：在项目入口处调用 setup_logging 和 load_config，把 cfg 传给各个模块。数据库模块用 get_conn 拿连接，模型模块用 get_llm_client 拿客户端。日志格式统一成时间 [级别] 名称 - 消息，这样 SQL 日志和模型调用日志在同一个文件里按时间排列，排查问题时能看清「先查了什么、再问了模型什么」。

如果你用 Claude Code 做开发，可以把 TaoToken 的 Base URL、Key、Model ID 填进它的配置，这样代码补全和对话都走同一条通道。接入文档里有具体步骤。长期跑 Agent 任务的话，Coding Plan 的额度规划比按次调用更划算。

最后给一个实用技巧：在日志里加一个请求 ID。每次业务请求生成一个 uuid，通过 logging 的 extra 参数带进每条 SQL 日志和模型调用日志。这样即使并发很高，你也能把一次请求涉及的所有 SQL 和模型交互串起来。实现方式是在 logger.debug 时传extra={"req_id": rid}，format 里加%(req_id)s。这个改动很小，但排查线上问题时价值很大。

配置即得可观测 SQL 日志，这句话的前提是配置写对、切面挂对、验证跑通。你现在手里这三样都有了，剩下的就是复制到项目里改改路径和凭证。