后端代码：from flask import Flask, jsonify, request from flask_cors import CORS import mysql.connector from mysql.connector import Error app = Flask(__name__) CORS(app, resources={r"/recommend": {"origins": ["http://localhost:8900"]}}) # 数据库配置 DB_CONFIG = { 'host': 'localhost', 'user': 'root', 'password': '123456', 'database': 'mbti_recommendation', 'charset': 'utf8mb4', 'pool_name': 'mypool', 'pool_size': 5, 'pool_reset_session': True } app.config['JSON_AS_ASCII'] = False # 全局参数校验 VALID_MBTI_TYPES = {'ESTJ', 'INFP', 'ENTP', 'ESFJ', 'ISTJ', 'ISFJ', 'ESTP', 'ESFP', 'ENTJ', 'INTJ', 'ENFP', 'INTP', 'ISTP', 'ISFP', 'INFJ', 'ENFJ'} VALID_TABS = {'major', 'career', 'test'} def get_db_connection(): """带连接池的数据库连接""" try: conn = mysql.connector.connect(**DB_CONFIG) if conn.is_connected(): print("✅ 数据库连接成功") return conn except Error as e: app.logger.error(f"[数据库连接失败]

<think>好的，用户现在需要了解如何在Flask后端中使用mysql.connector实现数据库连接池，以及如何实现一个/recommend的API接口。我需要先整理一下相关知识，确保回答准确。 首先，关于数据库连接池。Flask本身不内置连接池，但可以通过mysql.connector的pooling模块来实现。需要创建连接池，并在应用启动时初始化，然后在请求处理中获取连接，使用后归还。记得配置池的大小、名称等参数，还要处理可能的异常，比如连接超时或者获取失败的情况。 然后是API接口/recommend的实现。用户可能希望这个接口处理GET或POST请求，从数据库获取推荐数据，比如根据用户ID查询。这里需要用到Flask的路由装饰器，定义对应的函数。在函数内部，从连接池获取连接，执行查询，处理结果，最后返回JSON格式的数据。要注意SQL注入的问题，使用参数化查询，比如cursor.execute的第二个参数传递元组。 另外，还需要考虑跨域问题，可以使用flask-cors扩展来处理，这样前端调用时不会遇到CORS错误。在返回数据时，使用jsonify将结果转换为JSON响应。 用户可能还会关心错误处理，比如数据库连接失败或者查询出错的情况，应该捕获异常并返回相应的错误信息，比如500状态码。同时，连接使用后一定要确保关闭，放回连接池，避免资源泄露。 现在需要检查mysql.connector的版本是否支持连接池。mysql-connector-python库确实提供了连接池的功能，所以需要确认用户是否已经正确安装了这个库。如果用户使用的是旧版本，可能需要升级。 示例代码部分，应该包括连接池的初始化，在应用工厂函数中配置，比如在create_app函数里创建连接池。然后在路由中使用，比如在/recommend接口中获取连接，执行查询，处理结果，最后释放连接。 可能的相关问题包括如何配置连接池参数，如何处理数据库事务，如何优化查询性能，如何测试API接口，以及如何部署Flask应用。这些可以作为相关问题推荐给用户，帮助他们进一步学习和解决问题。 最后，确保代码示例中的占位符（如数据库名、表名、字段）被正确替换，用户需要根据自己的实际数据库结构进行调整。同时提醒用户注意敏感信息的保护，比如数据库密码不要明文写在代码中，可以使用环境变量或配置文件。</think># Flask连接MySQL数据库连接池及推荐接口实现 ## 一、数据库连接池配置 使用mysql.connector.pooling模块创建MySQL连接池： ```python import mysql.connector from mysql.connector import pooling from flask import Flask, jsonify def create_app(): app = Flask(__name__) # 创建连接池（在应用工厂函数中初始化） app.config['mysql_pool'] = pooling.MySQLConnectionPool( pool_name="flask_pool", pool_size=5, pool_reset_session=True, host='localhost', user='your_username', password='your_password', database='recommend_db' ) return app ``` ## 二、推荐接口实现 ### 1. 基础路由配置 ```python from flask import request @app.route('/recommend', methods=['GET']) def get_recommendations(): # 获取请求参数 user_id = request.args.get('user_id', type=int) # 参数验证 if not user_id: return jsonify({'error': 'Missing user_id parameter'}), 400 connection = None try: # 从连接池获取连接 connection = app.config['mysql_pool'].get_connection() cursor = connection.cursor(dictionary=True) # 执行查询（示例查询） query = """SELECT item_id, score FROM user_recommendations WHERE user_id = %s ORDER BY score DESC LIMIT 10""" cursor.execute(query, (user_id,)) # 处理结果集 results = cursor.fetchall() return jsonify({'recommendations': results}), 200 except mysql.connector.Error as err: # 错误处理 return jsonify({'error': f'Database error: {err}'}), 500 finally: # 释放连接回连接池 if connection: connection.close() ``` ### 2. 连接池特性说明 1. `pool_size=5` 设置初始连接数 2. `get_connection()` 自动管理连接复用 3. 使用with语句实现自动归还连接： ```python with app.config['mysql_pool'].get_connection() as conn: # 自动管理连接的获取和释放 ``` ## 三、性能优化建议 1. 使用参数化查询防止SQL注入 2. 添加缓存层（Redis）减少数据库访问 3. 异步任务处理推荐计算： ```python from concurrent.futures import ThreadPoolExecutor executor = ThreadPoolExecutor(2) @app.route('/recommend') def async_recommend(): user_id = request.args.get('user_id') executor.submit(background_task, user_id) return jsonify({'status': 'processing'}), 202 ``` ## 四、安全增强措施 1. 添加CORS支持： ```python from flask_cors import CORS CORS(app, resources={r"/recommend": {"origins": "*"}}) ``` 2. 请求频率限制： ```python from flask_limiter import Limiter limiter = Limiter(app, key_func=get_remote_address) @app.route('/recommend') @limiter.limit("100/day") def limited_recommend(): # ... ```