如何在uWSGI+Nginx部署的Flask API中实现MySQL数据库连接？

一、连接方式选择：用连接池替代单次创建/长期存活 #

• 别每次请求新建连接：频繁创建销毁连接会消耗大量服务器资源，拖慢接口响应速度
• 也别长期保持单个连接：连接长时间闲置会被MySQL服务器主动断开，导致后续请求报错
• 最优方案：用连接池管理连接，Flask生态里可以直接用SQLAlchemy（自带连接池），或者基于现有脚本封装连接池逻辑

二、Flask中集成数据库连接的代码结构 #

方案1：用SQLAlchemy（推荐，简化开发） #

如果不想自己维护连接池，直接用Flask-SQLAlchemy，它会自动处理连接的复用、回收：

1. 安装依赖：pip install flask-sqlalchemy
2. 初始化配置（可放在单独配置文件或app初始化文件）：

# app.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
# 替换为你的数据库信息
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://用户名:密码@localhost/数据库名'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 可选连接池配置，默认参数已满足多数场景
app.config['SQLALCHEMY_POOL_SIZE'] = 10  # 连接池最大连接数
app.config['SQLALCHEMY_POOL_RECYCLE'] = 3600  # 闲置1小时后自动回收连接，避免被MySQL断开

db = SQLAlchemy(app)

3. 定义数据模型（用ORM更规范，也可直接执行原生SQL）：

# models.py
from app import db

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password = db.Column(db.String(120), nullable=False)

4. 在登录接口中调用：

# routes.py
from flask import request, jsonify
from app import db
from models import User

@app.route('/login', methods=['POST'])
def login():
    data = request.get_json()
    username = data.get('username')
    password = data.get('password')
    
    # SQLAlchemy自动从连接池取连接，用完自动放回
    user = User.query.filter_by(username=username, password=password).first()
    if user:
        return jsonify({'code': 0, 'msg': '登录成功'})
    else:
        return jsonify({'code': 1, 'msg': '账号或密码错误'}), 401

方案2：基于现有脚本封装连接池（不想用ORM时） #

如果要保留自己的MySQL连接脚本，用DBUtils实现连接池：

1. 安装依赖：pip install DBUtils pymysql
2. 创建独立的数据库连接管理类（比如db_manager.py）：

# db_manager.py
import pymysql
from DBUtils.PooledDB import PooledDB

class DBManager:
    __pool = None
    
    def __init__(self):
        self.conn = self.__get_conn()
        self.cursor = self.conn.cursor()
    
    @classmethod
    def __get_conn(cls):
        if cls.__pool is None:
            cls.__pool = PooledDB(
                creator=pymysql,
                maxconnections=10,
                mincached=2,
                maxcached=5,
                blocking=True,
                host='localhost',
                port=3306,
                user='你的用户名',
                password='你的密码',
                database='你的数据库名',
                charset='utf8mb4'
            )
        return cls.__pool.connection()
    
    def query(self, sql, args=None):
        try:
            self.cursor.execute(sql, args)
            return self.cursor.fetchall()
        except Exception as e:
            self.conn.rollback()
            raise e
    
    def execute(self, sql, args=None):
        try:
            result = self.cursor.execute(sql, args)
            self.conn.commit()
            return result
        except Exception as e:
            self.conn.rollback()
            raise e
    
    def close(self):
        self.cursor.close()
        self.conn.close()

3. 在接口中使用：

# routes.py
from flask import request, jsonify
from db_manager import DBManager

@app.route('/login', methods=['POST'])
def login():
    data = request.get_json()
    username = data.get('username')
    password = data.get('password')
    
    db = DBManager()
    try:
        sql = "SELECT * FROM user WHERE username = %s AND password = %s"
        result = db.query(sql, (username, password))
        if result:
            return jsonify({'code': 0, 'msg': '登录成功'})
        else:
            return jsonify({'code': 1, 'msg': '账号或密码错误'}), 401
    finally:
        # 关闭连接实际是放回连接池，必须执行避免泄漏
        db.close()

三、关键问题解答 #

• 连接是否要每次请求后关闭？
  用连接池的话，关闭操作是把连接放回池子里而非销毁，必须执行避免连接泄漏；用SQLAlchemy则无需手动关闭，它会自动回收连接。
• 代码结构怎么安排？
  推荐拆分文件，避免堆在单个文件：
  • app.py：Flask应用初始化
  • config.py：数据库、Flask相关配置
  • models.py：数据模型（用ORM时）
  • db_manager.py：数据库连接管理（自定义连接池时）
  • routes.py：接口路由定义
    这种结构清晰，便于后续维护和扩展。
• 元类实现连接为什么不够高效？
  元类是用来控制类创建逻辑的，用它管理数据库属于过度设计，很难实现连接池的复用、自动回收等核心功能，反而增加代码复杂度，不如直接用成熟的连接池方案。

内容的提问来源于stack exchange，提问作者Andrew