### Flask 和 PyMySQL 的完整示例代码（含连接池） 以下是基于 Flask 和 PyMySQL 构建的一个完整的 Web 应用程序示例，其中包括数据库连接池的实现。此示例展示了如何通过配置和使用 `pymysql` 来管理数据库连接并优化性能。 #### 配置与初始化 首先，在项目中引入必要的模块，并完成基本的应用配置： ```python from flask import Flask, request, jsonify import pymysql.cursors from DBUtils.PooledDB import PooledDB app = Flask(__name__) # 数据库连接池配置 POOL = PooledDB( creator=pymysql, maxconnections=6, # 连接池允许的最大连接数，默认为0表示不限制 mincached=2, # 初始化时至少创建的空闲连接数量 maxcached=5, # 连接池中最多闲置的连接数量 blocking=True, # 当连接池耗尽时是否阻塞等待(True)，还是抛出异常(False) ping=0, # 检查MySQL服务状态的时间间隔(秒)，0表示不检查 host='127.0.0.1', port=3306, user='root', password='your_password', # 替换为实际密码 database='test_db', # 替换为实际数据库名 charset='utf8mb4' ) @app.route('/get_user/<int:user_id>', methods=['GET']) def get_user(user_id): """ 查询指定用户的详细信息。 """ connection = POOL.connection() try: with connection.cursor(pymysql.cursors.DictCursor) as cursor: sql = "SELECT * FROM users WHERE id=%s" cursor.execute(sql, (user_id,)) result = cursor.fetchone() if not result: return jsonify({"error": "User not found"}), 404 return jsonify(result), 200 finally: connection.close() if __name__ == '__main__': app.run(debug=True) ``` --- #### 关键点解析 1. **连接池的作用** 使用 `PooledDB` 创建了一个持久化的数据库连接池[^3]。这有助于减少频繁建立和销毁连接带来的开销，从而提升应用程序的整体性能。 2. **动态查询处理** 路由 `/get_user/<int:user_id>` 接收用户 ID 并返回对应的用户数据。如果未找到匹配记录，则会返回 HTTP 状态码 404 及错误提示。 3. **资源释放机制** 在每次操作完成后，确保调用 `connection.close()` 方法将连接归还给连接池而不是完全断开连接[^4]。 --- #### 扩展功能建议 为了进一步增强系统的健壮性和可维护性，可以考虑以下改进措施： - 添加全局异常捕获逻辑以优雅地处理运行时错误。 - 将敏感信息（如数据库凭证）存储于环境变量或专用配置文件中而非硬编码。 - 引入 ORM 工具（例如 SQLAlchemy），以便更方便地管理和操作复杂的数据结构[^1]。 ---