Python 操作 MySQL 数据库：从连接池到 ORM 的完整实践指南

1. 为什么裸写 PyMySQL 迟早会踩坑：连接池与 ORM 的真实场景

如果你写过pymysql.connect()然后cursor.execute()，大概率经历过这几件事：脚本跑几分钟后报Lost connection to MySQL server during query；并发一上来连接数直接打满Too many connections；插入中文变成问号；拼接 SQL 被人塞了个' OR '1'='1。这些不是 MySQL 的问题，是数据访问层没搭好。

Python 操作 MySQL 的主流方案其实就三层：最底层是驱动（PyMySQL、mysqlclient），中间是连接池（DBUtils、SQLAlchemy Pool），最上层是 ORM（SQLAlchemy ORM、Peewee）。很多人直接从驱动跳到 ORM，跳过了连接池这一层，结果 ORM 用起来也不稳。这篇就按「驱动 → 连接池 → ORM」的完整链路走一遍，每一层都给可复制的配置，最后用本地 MySQL 容器做读写验证。

适合谁看：写过一点 Python、能跑通pip install、但数据访问层总是出问题的后端同学；或者正在把脚本改造成常驻服务、需要连接复用的开发者。核心检索词就是 Python 操作 MySQL 的连接池配置和 SQLAlchemy ORM 映射，全文围绕这两个点展开。

先说清楚一个概念：连接池不是「优化」，是「必需」。MySQL 建立一条 TCP 连接 + 认证 + 初始化会话，成本在毫秒级，单机 QPS 上千时，每次请求都新建连接，光握手就吃掉大半性能。连接池做的事就是维护一批长连接，用完还回去，下次直接取。ORM 则解决另一件事：把表映射成类，把行映射成对象，让你少写 SQL、少拼字符串。

下面按顺序来：先装环境、起容器，再讲 PyMySQL 裸连的坑，然后上连接池，最后用 SQLAlchemy 把 ORM 和连接池串起来。

2. 环境准备与 TaoToken 前置：本地 MySQL 容器 + 依赖安装

2.1 用 Docker 起一个本地 MySQL

不折腾本机安装，直接用容器，删了重来也干净。下面这条命令起一个 8.0 的 MySQL，端口映射到 3306，字符集设成 utf8mb4：

docker run -d \ --name mysql-dev \ -p 3306:3306 \ -e MYSQL_ROOT_PASSWORD=root123 \ -e MYSQL_DATABASE=test_db \ -e MYSQL_USER=dev \ -e MYSQL_PASSWORD=dev123 \ mysql:8.0 \ --character-set-server=utf8mb4 \ --collation-server=utf8mb4_unicode_ci

等十几秒让容器初始化完，进去验证一下：

docker exec -it mysql-dev mysql -udev -pdev123 -e "SHOW DATABASES;"

能看到test_db就说明起来了。注意--character-set-server=utf8mb4这个参数必须加，否则默认字符集存 emoji 会报Incorrect string value。

2.2 安装 Python 依赖

pip install pymysql dbutils sqlalchemy cryptography

cryptography是 MySQL 8.0 默认的caching_sha2_password认证插件需要的，不装会报RuntimeError: 'cryptography' package is required。这个坑很常见，先装上省事。

2.3 关于 TaoToken 的接入位置

如果你的项目里还要调大模型做数据处理、SQL 生成或者结果摘要，可以把模型调用统一走 TaoToken 的 API 网关，和数据库访问层解耦。它的 Base URL 是https://taotoken.net/api，在代码里配置成 OpenAI 兼容的客户端即可。API Key 在控制台生成：https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=console

需要说明的是，TaoToken 在这里的角色是模型调用的统一入口，不碰你的数据库连接，两者是独立的。数据库这块还是老老实实用 PyMySQL + SQLAlchemy。模型对话调试可以用 https://taotoken.net/models?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=models 先验证 Key 是否可用。

2.4 建一张测试表

CREATE TABLE user ( id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(64) NOT NULL, age INT DEFAULT 0, email VARCHAR(128) UNIQUE, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

ENGINE=InnoDB是必须的，MyISAM 不支持事务，后面讲事务回滚会失效。

3. 可复制配置：PyMySQL 连接池参数模板与 SQLAlchemy 会话配置

3.1 先看裸连的问题

裸连的写法长这样：

import pymysql conn = pymysql.connect(host='localhost', port=3306, user='dev', password='dev123', database='test_db', charset='utf8mb4') cursor = conn.cursor() cursor.execute("SELECT * FROM user") print(cursor.fetchall()) cursor.close() conn.close()

单次脚本没问题，但放到 Web 服务里，每个请求都connect()再close()，连接数会瞬间飙高。而且close()只是把连接还给操作系统，不是真的复用。这就是为什么需要连接池。

3.2 DBUtils 连接池配置模板

DBUtils 提供两种池：PooledDB（连接池）和PersistentDB（每线程持久连接）。Web 服务用PooledDB，多线程脚本用PersistentDB。下面是PooledDB的完整配置：

import pymysql from dbutils.pooled_db import PooledDB POOL = PooledDB( creator=pymysql, maxconnections=20, # 池中最大连接数 mincached=5, # 启动时预建的空闲连接 maxcached=10, # 池中最多空闲连接 maxshared=0, # 共享连接数，0 表示不共享 blocking=True, # 连接耗尽时阻塞等待而非报错 maxusage=None, # 单连接最大复用次数，None 不限 setsession=['SET AUTOCOMMIT = 0'], # 关闭自动提交 ping=1, # 每次取连接前 ping 一次，防断连 host='localhost', port=3306, user='dev', password='dev123', database='test_db', charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor, )

几个参数值得单独说。ping=1是关键，MySQL 默认wait_timeout是 8 小时，空闲连接会被服务端断开，ping=1会在取连接时检测并重连，避免Lost connection。blocking=True让连接耗尽时排队等待，而不是直接抛异常。setsession=['SET AUTOCOMMIT = 0']把自动提交关掉，事务控制权交回代码。

用的时候：

def query_users(): conn = POOL.connection() try: with conn.cursor() as cursor: cursor.execute("SELECT id, name, age FROM user WHERE age > %s", (18,)) return cursor.fetchall() finally: conn.close() # 注意：这里是还给池，不是真关闭

conn.close()在 DBUtils 里是归还连接，不是断开。这点和裸连语义不同，别搞混。

3.3 SQLAlchemy 引擎与会话配置

SQLAlchemy 自带连接池，不用再套 DBUtils。核心是create_engine和sessionmaker：

from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, declarative_base DATABASE_URL = "mysql+pymysql://dev:dev123@localhost:3306/test_db?charset=utf8mb4" engine = create_engine( DATABASE_URL, pool_size=10, # 常驻连接数 max_overflow=20, # 超出 pool_size 后临时创建的连接上限 pool_recycle=3600, # 连接回收时间（秒），小于 MySQL wait_timeout pool_pre_ping=True, # 取连接前 ping，等价于 DBUtils 的 ping=1 echo=False, # True 会打印所有 SQL，调试用 ) SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False) Base = declarative_base()

pool_recycle=3600和pool_pre_ping=True是防断连的双保险。pool_recycle让连接在 1 小时后强制重建，pool_pre_ping在取连接时检测可用性。生产环境这两个都建议开。

3.4 ORM 模型映射

把user表映射成类：

from sqlalchemy import Column, Integer, String, DateTime, func class User(Base): __tablename__ = 'user' id = Column(Integer, primary_key=True, autoincrement=True) name = Column(String(64), nullable=False) age = Column(Integer, default=0) email = Column(String(128), unique=True) created_at = Column(DateTime, server_default=func.now()) def __repr__(self): return f"<User(id={self.id}, name={self.name}, age={self.age})>"

server_default=func.now()让时间戳由数据库生成，避免应用服务器时区不一致的问题。

3.5 会话依赖注入模板

FastAPI 或 Flask 里，会话要按请求创建、按请求关闭：

def get_db(): db = SessionLocal() try: yield db finally: db.close()

这个get_db生成器保证每个请求一个会话，请求结束自动归还连接。别用全局单例 Session，多线程下会串数据。

4. 验证请求：本地容器下的读写与事务回滚实测

4.1 插入与查询验证

先跑一段插入 + 查询，确认 ORM 映射和连接池都正常：

from sqlalchemy import select def create_and_query(): db = SessionLocal() try: new_user = User(name="张三", age=20, email="zhangsan@example.com") db.add(new_user) db.commit() db.refresh(new_user) print("插入成功，ID:", new_user.id) stmt = select(User).where(User.age > 18) for u in db.scalars(stmt): print(u) finally: db.close() create_and_query()

预期输出：

插入成功，ID: 1 <User(id=1, name=张三, age=20)>

如果name显示成乱码，检查连接串里的charset=utf8mb4和建表时的字符集是否一致。

4.2 事务回滚验证

事务是数据访问层的核心。下面这段故意在插入后抛异常，验证回滚：

def transaction_test(): db = SessionLocal() try: db.add(User(name="李四", age=25, email="lisi@example.com")) db.add(User(name="王五", age=30, email="lisi@example.com")) # email 重复 db.commit() except Exception as e: db.rollback() print("事务回滚:", type(e).__name__) finally: db.close() transaction_test()

因为email有唯一约束，第二条插入会触发IntegrityError，整个事务回滚，李四也不会被写入。验证一下：

db = SessionLocal() print(db.query(User).filter_by(name="李四").count()) # 输出 0 db.close()

输出 0 说明回滚生效。如果输出 1，说明autocommit没关，或者用了 MyISAM 引擎。

4.3 连接池复用验证

验证连接确实在复用，可以打印连接 ID：

def pool_check(): ids = set() for _ in range(5): db = SessionLocal() conn = db.connection().connection ids.add(id(conn)) db.close() print("不同连接数:", len(ids)) pool_check()

pool_size=10的情况下，5 次请求大概率复用同一条连接，输出不同连接数: 1。如果每次都是新连接，检查pool_size是否被设成了 0。

4.4 参数化查询防注入验证

用 ORM 或参数化查询，注入字符串会被当普通值处理：

db = SessionLocal() result = db.query(User).filter(User.name == "张三' OR '1'='1").all() print("匹配条数:", len(result)) # 输出 0，而不是全表 db.close()

输出 0 说明注入被挡住了。如果输出全表，说明你用了字符串拼接，赶紧改。

5. 本篇常见错排查：401、Lost connection、reading choices 与 OAuth 报错

5.1 认证失败：Access denied 与 401

报错：

pymysql.err.OperationalError: (1045, "Access denied for user 'dev'@'localhost' (using password: YES)")

原因通常是密码错、用户没建、或者 host 不匹配。容器里MYSQL_USER建的用户默认只允许从%访问，但如果你手动建过'dev'@'localhost'，从容器外连就会失败。解决：

CREATE USER 'dev'@'%' IDENTIFIED BY 'dev123'; GRANT ALL PRIVILEGES ON test_db.* TO 'dev'@'%'; FLUSH PRIVILEGES;

如果报的是RuntimeError: 'cryptography' package is required，那是缺依赖，pip install cryptography即可。

5.2 Lost connection to MySQL server during query

报错：

pymysql.err.OperationalError: (2013, 'Lost connection to MySQL server during query')

这是连接被服务端断开了。原因：wait_timeout到期、网络抖动、或者连接池里的连接太久没用。解决三件套：pool_pre_ping=True、pool_recycle=3600、DBUtils 的ping=1。三个都配上基本不会再出现。

5.3 reading choices 相关报错

如果你在 SQLAlchemy 里用db.query(User).all()报类似AttributeError: 'NoneType' object has no attribute 'reading choices'，通常是会话被提前关闭了。检查是不是在with块外访问了 ORM 对象，或者get_db的finally里close()执行太早。ORM 对象在会话关闭后访问懒加载属性会报错，要么在会话内取完数据，要么用joinedload预加载。

5.4 OAuth 与模型调用报错

如果你的项目里同时调 TaoToken 的模型接口，报 401 通常是 API Key 没配或过期。检查请求头：

Authorization: Bearer <你的Key>

Base URL 用https://taotoken.net/api，别多加/v1或漏掉。Key 在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=api-keys 生成。如果报 OAuth 相关错误，确认你用的是 API Key 而不是网页登录态，两者不通用。

5.5 字符集乱码

报错：

pymysql.err.DataError: (1366, "Incorrect string value: '\\xF0\\x9F...' for column 'name'")

存 emoji 失败。检查三处：连接串charset=utf8mb4、建表DEFAULT CHARSET=utf8mb4、MySQL 服务端character-set-server=utf8mb4。三处都对了才能存。

5.6 连接池耗尽

报错：

TimeoutError: QueuePool limit of size 10 overflow 20 reached

连接池满了。原因通常是会话没关，或者有慢查询占着连接。检查get_db的finally是否执行，慢查询加索引。临时可以调大max_overflow，但根治要找到泄漏点。

6. 从脚本到服务：把数据访问层稳定下来的几个实操建议

数据访问层搭好之后，有几个习惯能省很多事。第一，所有查询走参数化，永远不拼字符串，ORM 的filter和 PyMySQL 的%s占位符都是安全的。第二，事务边界要清晰，一个业务操作一个事务，别在一个事务里做网络请求。第三，连接池参数按实际并发调，pool_size不是越大越好，MySQL 的max_connections默认 151，池开太大反而会打满服务端。

如果你在写 Agent 或者需要长期跑的编码任务，模型调用和数据库访问可以分开配置。TaoToken 的 Coding Plan 适合需要持续调用模型的场景：https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=coding-plan 接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content=doc 有完整的 Base URL、Key 和 Model ID 三件套说明。

最后留一个我常用的排查顺序：连不上先看ping和pool_pre_ping，乱码先看三处字符集，事务不生效先看autocommit和引擎，性能差先看慢查询日志。按这个顺序走，大部分问题十分钟内能定位。