Flask 构建稳健 APIMySQL 参数化查询与 DBUtils 连接池的完整落地经验先说我自己的结论如果你还在 Flask 里每个请求临 new 一个数据库连接或者还在用字符串拼接 SQL那么接口迟早会在并发上来之后出问题。这不是理论问题是我在一个实际项目中吃过的亏。上一篇文章讲完了 Flask 项目的基础结构这篇就来解决数据库访问层的工程化问题主题很直白用 DBUtils 连接池管理 MySQL 连接用参数化查询干掉 SQL 注入隐患让 API 层既扛得住压力也经得起代码审查。下面所有内容都来自实际代码和踩坑记录新手可以直接照着抄老手可以重点看第四节的事务封装和第六节的排查思路。1. 为什么要把数据库访问改成“连接池 参数化查询”很多 Flask 入门教程教你的写法是在路由函数里pymysql.connect()查出数据之后db.close()。这个写法在 demo 和本地开发里完全没问题但一旦放到真实环境尤其是多 worker 部署之后问题会非常明显。1.1 直接连库的三个坑反复握手、并发连接打满、SQL 注入第一个坑是连接建立成本。MySQL 连接不是免费资源客户端与服务器之间需要完成 TCP 三次握手、MySQL 协议版本协商、认证、权限校验等步骤。在高频接口里哪怕每次连接只消耗 1~2 毫秒放大到每天几百万次请求浪费的时间和数据库端的资源开销都非常可观。更重要的是如果接口里还埋着慢查询连接持有时间会从毫秒级拉长到秒级这个时候并发稍微一高MySQL 的 max_connections 很容易被打满。第二个坑是并发场景下连接耗尽。Flask 自带的开发服务器是单进程单线程你在开发环境不会遇到并发问题。但生产环境用 gunicorn 起 4 个 worker每个 worker 内的每个请求如果都独立创建连接同一时刻数据库可能要同时处理几十甚至上百个连接。很多云数据库实例默认连接数上限只是 100 左右一旦冲上去服务直接报Too many connections而且这个错误还会引发连锁故障因为监控告警、日志上报这些系统也可能依赖同一个数据库。第三个坑是安全这个坑最致命。用字符串拼接 SQL 的典型写法是sql SELECT * FROM user WHERE name %s % name只要name来自前端请求参数攻击者传入 OR 11整张表就跪了传个; DROP TABLE user; --数据直接没了。参数化查询的核心作用不是让代码变好看而是把 SQL 结构和传入的数据彻底分离让数据库引擎把参数当作数据来处理而不是可执行的 SQL 片段。1.2 连接池解决什么问题复用连接与控制并发连接池的思路很朴素预先创建一批连接放在池子里每次请求需要数据库访问时从池子里“借”一个连接用完再“还”回去而不是销毁重建。这样做有两个直接好处一是连接可以反复复用正常情况下请求结束后连接回到池中下一个请求继续使用省掉了反复握手的开销二是可以硬性限制同时使用的连接数量超过上限的请求排队等待而不是无限打向数据库。打个比方不用连接池就像每次打电话都要重新拨号、重新认证挂了电话线路就释放用连接池就像公司总机再接分机电话打完挂掉线路仍然保留下一个内部通话直接接上。在 API 接口这种“短事务、高频次”的场景下连接池带来的收益非常明显。我自己在测试环境跑过一个简单压测无连接池时接口 QPS 大概在 500 左右加了连接池之后直接翻了一倍多而且数据库连接数从几十个降到个位数这个差距在中高并发下会继续拉大。1.3 参数化查询解决什么问题注入与隐式类型转换参数化查询在防注入之外还有一个容易被忽略的好处它强制你以“字段 值”的方式组织查询等于给 SQL 语句加了一层统一的格式化入口。很多团队代码规范里要求“禁止拼接 SQL”落实下来最实际的手段就是统一用参数化查询。MySQL 的驱动程序比如 PyMySQL、mysqlclient都在底层支持预编译或客户端参数转义参数会以正确的类型和转义后的形式传给数据库。而且参数化查询还能减少隐式转换带来的性能问题。举个例子如果 user_id 在数据库里是 bigint但你拼接 SQL 时传进来的是字符串MySQL 可能选择放弃索引做全表扫描。参数化查询传参时带上明确的 Python 类型驱动会尽量保持类型虽然不能百分之百杜绝隐式转换但至少可以避免常见的人为失误。2. 技术选型DBUtils 连接池与 MySQL 驱动怎么选DBUtils 是 Python 生态里老牌且极简的数据库连接池库它本身不实现 MySQL 协议只做连接管理底层依赖一个符合 DB-API 2.0 规范的驱动。这里先解决两个选型问题不然代码写了也白写。2.1 选 PyMySQL 还是 mysqlclient差别在编译与性能PyMySQL 是纯 Python 实现的 MySQL 客户端安装无痛pip install pymysql直接可用跨平台兼容性最好。mysqlclient 是 MySQLdb 的分支底层调用了 C 扩展性能略好但在 Windows 上需要预编译的 wheel 包在部分 Linux 环境下需要安装 libmysqlclient-dev 才能编译。我的建议是追求安装省事、环境不可控选 PyMySQL追求极限性能且具备编译条件选 mysqlclient。两者对 DBUtils 来说没有本质区别因为 DBUtils 接受任何一个 DB-API 2 驱动作为 creator。有一点必须提醒如果项目里有多个依赖间接引用了 MySQLdb比如某些 ORM 或老库而你的环境只装 PyMySQL可以在入口处加一句pymysql.install_as_MySQLdb()但这只是权宜之计长期还是建议固定一种驱动并统一封装。驱动版本建议锁大版本我在项目中就遇到过 PyMySQL 小版本升级后默认事务行为出现差异的情况。2.2 PooledDB 还是 PersistentDB区别在连接归谁持有DBUtils 1.x 提供两个核心类PooledDB 和 PersistentDB。PooledDB 是真正意义上的连接池连接在多个线程间轮转复用PersistentDB 则为每个线程单独建立一个专用连接线程结束后才关闭。PersistentDB 的好处是线程内始终拿着同一个连接不需要竞争锁坏处是一旦线程数多了连接数也跟着涨而且连接长期空闲容易触发 MySQL 的 wait_timeout。对于 Flask API 这种“请求处理短、线程池相对固定”的场景我更推荐 PooledDB因为它真正把连接数压到最低配合请求级借还模式最合适。2.3 PooledDB 关键参数参数调不好等于白配from dbutils.pooled_db import PooledDB import pymysql pool PooledDB( creatorpymysql, maxconnections20, mincached2, maxcached10, maxshared0, blockingTrue, maxusage1000, setsession[], ping1, host127.0.0.1, port3306, userroot, passwordpassword, databaseshop, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor )逐个参数说。maxconnections是连接池允许创建的最大连接数我一般结合 MySQL 实例的 max_connections 来定保留一部分给后台任务和管理员连接一般设为实例上限的三分之一左右。mincached是启动时预先创建的连接数设为 2~4 就够了太小起不到预热效果太大浪费资源。maxcached是池中最多保留的空闲连接数超过部分会被真的关闭。maxshared是共享连接数上限如果不用共享连接直接设 0 即可。blockingTrue很关键。当连接池里的连接全被借走时如果设 False新的连接请求会直接抛异常设 True则进入等待队列。这个参数要结合接口超时时间一起评估我通常保持 True并且把 API 层的读操作超时时间设置的比连接等待超时大避免前端直接 5xx。maxusage表示单个连接最多被复用多少次达到上限后这个连接会被关闭重建我设 1000 是为了规避 MySQL 服务端连接状态累积导致的内存膨胀。ping1表示每次从连接池取出连接时如果发现连接已失效比如被 MySQL 服务端关闭会自动重连这个参数很省心配合后面的 wait_timeout 问题一起看。3. 准备项目骨架与数据库配置这一步不是核心代码但直接决定后面好不好维护。很多教程把配置写在app.py里的一堆全局变量中连密码都是明文这个习惯在真实项目中非常危险。3.1 目录结构建议我习惯把数据库访问相关代码独立成包避免在视图层到处 import pymysql。一个典型的 Flask 项目布局如下config.py # 环境配置类 models/ __init__.py db.py # 数据库连接池封装 user_model.py # 用户表相关操作 app/ __init__.py # app 工厂 api/ users.py # 用户相关接口 utils/ response.py # 统一响应格式 requirements.txt run.pymodels/db.py里只放连接池初始化、获取连接、释放连接的工具函数业务 SQL 全部下沉到对应 model 文件中视图层不直接接触 SQL。这样分工的好处是要改数据库连接配置时只改一处要审 SQL 时只看 model 层的几个文件新增一张表时也只需要新增一个 model不会把接口文件搞成上千行的“屎山”。3.2 数据库配置怎么放配置不要写死在代码里也不要提交到仓库。最省事的做法是放在环境变量或者外置的.env文件里。Flask 项目的配置类可以这样写import os from dotenv import load_dotenv load_dotenv() class Config: DB_HOST os.getenv(DB_HOST, 127.0.0.1) DB_PORT int(os.getenv(DB_PORT, 3306)) DB_USER os.getenv(DB_USER, root) DB_PASSWORD os.getenv(DB_PASSWORD, ) DB_NAME os.getenv(DB_NAME, shop) DB_CHARSET os.getenv(DB_CHARSET, utf8mb4) DB_POOL_MAX int(os.getenv(DB_POOL_MAX, 20))这里我默认使用了python-dotenv如果你不熟悉它就是从.env文件里读取键值对并注入环境变量的工具。密码、端口这类敏感信息用环境变量管理至少能避免把生产密码推到代码仓库里的低级错误。3.3 为什么统一用 utf8mb4MySQL 的 utf8 字符集实际上最多只支持 3 个字节的字符emoji 和一些生僻汉字都是 4 字节在 utf8 字符集下插入会直接报错或者变成问号。utf8mb4 才是 MySQL 对完整 Unicode 的正确实现。所以连接参数里的 charset 选项我现在一律写utf8mb4同时确保建表语句里也是 utf8mb4否则两边不一致照样乱码。4. 连接池封装与 Flask 请求生命周期集成连接池不能让每个请求都去PooledDB(...)新建一遍否则连接池就失去了意义——每次请求实际上新建了一个全新的池旧池里的连接没人管最终全变成幽灵连接。正确做法是全局只初始化一次然后在每次请求时从池中借连接请求结束时归还。4.1 全局单例 g 对象 teardown 的接线方式Flask 提供了一个 app context 机制g对象就是绑定在 app context 上的存储空间。同一个请求上下文内g 对象只存活一次天然适合挂载请求级别的数据库连接。配合teardown_appcontext钩子无论视图函数正常返回还是抛出异常这个钩子都会被调用所以连接归还代码放在这里是绝对安全的。具体实现# models/db.py from flask import g from dbutils.pooled_db import PooledDB import pymysql from config import Config _conn_pool None def get_pool(): global _conn_pool if _conn_pool is None: _conn_pool PooledDB( creatorpymysql, maxconnectionsConfig.DB_POOL_MAX, mincached2, maxcached10, maxshared0, blockingTrue, maxusage1000, ping1, hostConfig.DB_HOST, portConfig.DB_PORT, userConfig.DB_USER, passwordConfig.DB_PASSWORD, databaseConfig.DB_NAME, charsetConfig.DB_CHARSET, cursorclasspymysql.cursors.DictCursor ) return _conn_pool def get_conn(): if not hasattr(g, _db_conn): g._db_conn get_pool().connection() return g._db_conn def close_conn(exceptionNone): conn g.pop(_db_conn, None) if conn is not None: conn.close() def init_db(app): app.teardown_appcontext def _close(exception): close_conn(exception)这里有个细节必须说明conn.close()在 DBUtils 的连接池语义下不是真正关闭连接而是把连接归还到池中。如果不调用 close连接会一直挂在请求上最终导致连接池被借空后续请求全部卡在等待队列里。这一点新手很容易搞混以为是没写完 SQL 就关数据库其实恰恰相反这里的 close 就是“还钥匙”。连接在 PyMySQL 里的默认 autocommit 是 False连接归还前如果有未提交事务会残留状态这个问题我在后面的排查章节专门讲。4.2 cursor 的管理是另一个隐藏雷区很多人写完查询就忘记关 cursor。在连接池复用机制下cursor 不关闭连接归还池中后下一个请求再用这个连接时会带着旧 cursor 的残留。更严重的场景是开着 cursor 的同时启动另一个查询MySQL 会认为连接状态不干净直接报Commands out of sync。所以我封装数据操作方法时严格遵循“with 打开 cursor用完立即关闭”的模式def query_all(sql, paramsNone): conn get_conn() with conn.cursor() as cursor: cursor.execute(sql, params) return cursor.fetchall()with cursor块结束时会自动执行cursor.close()不用手动操心。这个写法的好处是即使 SQL 执行过程中抛出异常cursor 也会被关闭连接收回后不会残留状态。4.3 为什么不推荐使用 Flask-SQLAlchemyFlask-SQLAlchemy 是很流行的方案但有一些人并不需要 ORM。我现在的项目里业务 SQL 量大、逻辑复杂单独给每个查询套 ORM 反而费劲。DBUtils 方案保留了原生 SQL 的控制力又解决了连接管理的痛点属于介于“裸连”和 ORM 之间的中间态。如果你团队里对 SQL 能力有信心DBUtils PyMySQL 足够简单可靠如果团队新人多担心手写 SQL 出问题那再考虑 ORM 也不迟。两者没有绝对优劣但不要为了“不上 ORM”就放弃连接池和参数化那是把婴儿和洗澡水一起倒掉了。5. 参数化查询 API 实战从列表查询到事务型接口这一节写几个高频 API 场景每种场景我都会给出带参数化查询的完整示例并说明容易踩的细节。5.1 分页列表查询的正确写法做列表接口最常用的 SQL 是 count 加 limit。参数化写法如下def list_users(page1, page_size20): offset (page - 1) * page_size sql SELECT id, username, email, created_at FROM user WHERE status %s ORDER BY id DESC LIMIT %s OFFSET %s return query_all(sql, (1, page_size, offset))这里有两个注意点。第一LIMIT %s OFFSET %s的参数在 PyMySQL 里是整型没问题。第二page 和 page_size 从前端传进来时都是字符串我这里先转成 int 再传参避免数据库端做隐式类型转换。转 int 这一步必须在函数入口处完成并且要放在 try except 里否则前端传个非法参数进来接口直接 500。生产实现可以用一个通用的分页参数解析函数。5.2 LIKE 模糊搜索占位符里不能写引号模糊搜索最常见的报错是写法上习惯性给参数加引号# 错误写法 sql SELECT * FROM user WHERE username LIKE % %s % cursor.execute(sql, (keyword,))这里的问题在于 Python 的字符串拼接把 SQL 结构搞乱了。正确做法是把“拼接好的模式字符串”作为参数传进去def search_users(keyword): like_pattern f%{keyword}% sql SELECT * FROM user WHERE username LIKE %s return query_all(sql, (like_pattern,))LIKE 的转义问题也要注意如果关键字本身包含%或_这两个字符在 LIKE 中会被当作通配符。对你的业务而言如果允许用户搜索这些字符但不希望它们作为通配符生效需要在拼接前做转义from pymysql.converters import escape_string keyword keyword.replace(\\, \\\\).replace(%, \\%).replace(_, \\_)这个细节很冷门但一旦业务中确实出现了搜索%的场景你会感谢自己写过这段。5.3 IN 列表查询与批量插入参数化查询不能直接传列表需要先动态生成占位符def get_users_by_ids(user_ids): if not user_ids: return [] placeholders ,.join([%s] * len(user_ids)) sql fSELECT id, username FROM user WHERE id IN ({placeholders}) return query_all(sql, tuple(user_ids))这里用f-string生成了占位符字符串但传入的参数仍是元组。id 列表的长度需要加个上限我一般限制 200防止前端传几千个 id 导致 SQL 过长。批量插入使用executemany写法是def batch_insert_users(users): sql INSERT INTO user (username, email) VALUES (%s, %s) conn get_conn() with conn.cursor() as cursor: cursor.executemany(sql, users) conn.commit()users 是一个元组列表每个元组对应一条记录的参数。这个地方还必须强调executemany并不是一次网络往返就完成了底层还是拆成多条 execute但驱动会做优化起码比在 Python 里 for 循环逐条 execute 快很多。如果插入量巨大比如十万条以上建议先分片每 1000 条一个事务然后调用conn.commit()。5.4 事务型 API订单创建与库存扣减跨表写操作必须放在事务里这是数据库开发的基本常识。下面给出一个“创建订单 扣减库存”的模拟实现def create_order(user_id, product_id, quantity): conn get_conn() try: with conn.cursor() as cursor: # 查询库存并锁定该行 cursor.execute( SELECT stock FROM product WHERE id %s AND status 1 FOR UPDATE, (product_id,) ) row cursor.fetchone() if not row: raise ValueError(product not exists) if row[stock] quantity: raise ValueError(insufficient stock) # 扣减库存 cursor.execute( UPDATE product SET stock stock - %s WHERE id %s, (quantity, product_id) ) # 创建订单 cursor.execute( INSERT INTO order (user_id, product_id, quantity, amount, status) VALUES (%s, %s, %s, %s, %s), (user_id, product_id, quantity, quantity * price, 1) ) conn.commit() return True except Exception: conn.rollback() raise finally: conn.close()这段代码有几个关键点。第一FOR UPDATE是行级悲观锁防止并发下多个请求同时读到同一个库存余量把库存扣成负数。真实生产中库存量大的话可以用乐观锁替代但这里强调事务正确性用悲观锁最直观。第二conn.commit()只调用一次但放在所有写操作之后异常时rollback()回滚全部操作。第三接口层如果还需要返回订单号可以在这个事务里查出插入后的自增 id并把它作为返回值带出。这里要专门提醒一个 DBUtils 的坑如果事务函数里忘了 call rollback连接还回池子里时事务依然是开着的下一个请求拿到这个连接继续执行新语句时会直接踩在未提交的快照上轻则查不到刚提交的数据重则出现死锁或脏读。所以事务函数里必须把 rollback 放在异常处理中这个习惯要养成。6. 常见问题与排查实录连接池上线之后你会遇到一些裸连时代没见过的怪问题。这里列几个我实际踩过的坑每个问题给出定位思路和解决方案。6.1 连接池耗尽看 processlist而不是瞎猜现象是接口响应时间突然拉长日志里出现大量Cant connect to MySQL server (Too many connections)或 DBUtils 排队超时。首要排查步骤是登录数据库执行SHOW PROCESSLIST;这个命令能列出当前所有连接、来源 IP、账号、状态和正在执行的 SQL。如果看到大量连接状态为Sleep说明连接没被正确归还检查代码里有没有在请求里手动 new 连接但忘记 close 的分支如果看到连接状态全部是Query说明是慢查询把连接占满此时去SHOW FULL PROCESSLIST看每条 SQL 的执行时间分析慢在哪。通过 mysql 的information_schema.processlist还能按来源 IP 聚合统计快速判断是哪个服务实例耗完了连接。DBUtils 参数上可以做兜底maxconnections不要设成无限大blockingTrue配合连接获取超时时间DBUtils 不直接提供可以在外面包一层threading.Condition或直接靠应用层超时防止进程无限卡死。更重要的还是源头治理确保所有分支都调用 close特别是异常分支和 return 分支。6.2 MySQL 8 小时空闲断开wait_timeout、ping、maxusage 三件套MySQL 服务端默认的wait_timeout是 28800 秒也就是 8 小时。连接池里的连接如果超过这个时间没有被使用服务端会主动关闭。这时连接池还以为连接是好的取出来一执行 SQL直接报MySQL server has gone away。DBUtils 提供了ping参数来应对。它接受一个整数表示检查策略ping0从不检查ping1每次取出连接时检查ping2每次取出和归还时都检查更大的值表示超过该秒数未使用的连接才 ping 检查。我推荐ping1检查成本不高效果最直接。同时设置maxusage1000让单条连接最多复用 1000 次后自动重建也能绕开连接状态越用越脏的问题。另外提一句如果你的数据库做了 VIP 漂移或者网络断连连接池中的连接在 TCP 层就已经断了此时靠 ping 可以保住命但底层驱动抛错之后要记得把旧连接从池里废弃掉DBUtils 在这块处理得比较成熟基本不用自己操心。6.3 事务残留与死锁事务结束前必须归还干净死锁的一个典型场景是A 请求持有连接 1 的事务B 请求持有连接 2 的事务两边都等待对方释放行锁MySQL 会在一方超时后自动回滚并报Deadlock found。排查死锁时执行SHOW ENGINE INNODB STATUS查看最下方的LATEST DETECTED DEADLOCK部分里面能看到两条事务各自持有的锁和等待的锁定位是哪几张表、哪几条 SQL 参与。还有一种不太容易发现的事务残留一个视图函数里既查了数据又调用了另一个写函数第二个函数事务 commit 前第一个函数的查询已经执行完了。在 autocommitFalse 的情况下前面的查询会被默认纳入后续事务导致事务持有时间被人为拉长容易出现锁等待超时。解决办法是每条写操作尽可能短、快事务内只做必要操作不能在事务里调用外部 HTTP 请求不能开文件不能等待队列任务。奉行的原则很简单事务越短锁竞争越小。6.4 中文乱码与 datetime 序列化问题中文乱码一般是 charset 设置不一致。连接侧设置了charsetutf8mb4但表结构或者数据库本身是 latin1 或 utf8写入后读取就会乱码。修法是把整库、表、字段的字符集统一改成 utf8mb4注意修改时要检查外键和索引长度。如果资金允许直接建一个新库把数据迁过去比就地改表安全得多。datetime 序列化问题发生在你把查询结果直接丢给jsonify的时候。MySQL 的 datetime 字段在 PyMySQL 中会被解析为 Python 的datetime.datetime而 Flask 原生 jsonify 不认识 datetime会报TypeError: Object of type datetime is not JSON serializable。两个解决思路一个是在 SQL 里用DATE_FORMAT格式化一个是自定义 JSON 序列化器。import json from datetime import datetime, date class CustomJSONEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, (datetime, date)): return obj.strftime(%Y-%m-%d %H:%M:%S) return super().default(obj) app.json_encoder CustomJSONEncoder这个方案的好处是全局生效不用每个字段单独格式化。第二个思路如果你有专门的响应工具函数来处理输出格式也可以在工具函数里把 datetime 递归转成字符串效果一样。6.5 连接池被多进程重复初始化使用 gunicorn 多 worker 部署时每个 worker 是一个独立进程全局变量_conn_pool在每个进程里都会被初始化一次这是正常的。但如果你在多个模块里分别引用了get_pool()而_conn_pool初始化时没有加线程锁或没有做单例保护在极端情况下可能被并发创建两次导致每个进程里有两个池子连接数翻倍。DBUtils 的官方文档其实也强调连接池应该全局单例。为了防止容器多线程环境下初始化竞态可以给初始化函数加个锁import threading _pool_lock threading.Lock() def get_pool(): global _conn_pool if _conn_pool is None: with _pool_lock: if _conn_pool is None: _conn_pool PooledDB(...) return _conn_pool双重检查锁的模式在这里是合理的代价几乎为零但能避免进程启动阶段的偶发问题。我在生产环境确实遇到过类似情况排查起来费了不少时间。7. 最后的经验与建议连接池和参数化查询做进项目后数据库访问层基本就稳了。但我想多分享一点技术选型之外团队的习惯更重要。代码评审时如果发现新的 SQL 是拼接字符串直接打回新增接口时必须复用get_conn()和统一的数据操作方法连接池参数不能轻易改尤其maxconnections不是越大越好数字越大对数据库的冲击越大。我自己在项目初期犯过一个低级错误为了“稳妥”把maxconnections设成了 100结果数据库实例最大连接数只有 100监控系统自己先连不上了业务反而被打爆。所以连接池参数的设定要以数据库规格为前提而不是拍脑袋。如果你正在从裸连改造过来我的建议是分三步走先把所有 SQL 改成参数化查询这是纯代码层面的改动测试覆盖够的情况下很快再接入 DBUtils 连接池并加上 teardown 回收最后才是压测调参根据 QPS、TP99 和数据库连接数来微调maxconnections、maxusage和ping这几个核心参数。这中间任何一步出问题前面讲的排查方法基本都能帮你定位。等这套架构稳定跑上几个月你会明显感觉到接口不再因为数据库连接而 5xx压测数据也好看很多代码审查时关于 SQL 安全的问题也几乎绝迹。
