Flask create_app工厂模式下，模型类无法访问自定义数据库连接的解决

我明白你现在遇到的困扰了——用Flask的应用工厂模式（create_app）构建项目时，自定义的数据库封装没办法像SQLAlchemy那样轻松在model文件里调用，核心问题在于：你在create_app里创建的db实例是局部变量，models/user.py里的UserModel根本拿不到它；如果直接在model里实例化Database，又没有正确的配置参数，自然会连接失败。

下面给你几种实用的解决方案，按照Flask的设计习惯推荐第一种：

方案一：将数据库实例绑定到Flask App对象（最推荐） #

这种方式贴合Flask的上下文设计，把数据库实例挂载到app对象上，再通过current_app代理在model中访问。

步骤1：修改app.py，绑定db到app #

在create_app函数中，创建完db实例后，把它绑定到app对象上：

def create_app(debug=True):
    config = LocalConfig if debug else LiveConfig
    app = Flask(__name__)
    app.config.from_object(config)
    # ... 其他配置代码 ...
    
    # 创建数据库实例并绑定到app
    db = Database(
        host=app.config['MYSQL_HOST'],
        db=app.config['MYSQL_DB'],
        user=app.config['MYSQL_USER'],
        passwd=app.config['MYSQL_PASS'],
    )
    app.db = db  # 把db挂载到app属性上
    
    register_decorators(app)
    register_extensions(app)
    register_endpoints(api)
    return app

步骤2：修改models/user.py，通过current_app获取db #

导入Flask的current_app代理，在model方法中调用绑定的数据库实例：

import datetime
import json
import logging
from flask import current_app  # 导入current_app

class UserModel:
    @classmethod
    def get_by_username(cls, username=None):
        # 从当前app上下文获取数据库实例
        user = current_app.db.getOne(
            table='users',
            fields=['user_id','data'],
            where=('username = %s', [username])
        )
        if user:
            user['data'] = json.loads(user['data'])
        return user

说明：

current_app是Flask提供的上下文代理，只要在请求上下文或应用上下文存在的环境中（比如接口请求、脚本中手动推送上下文），它就会指向当前的Flask app实例，完美适配应用工厂模式。如果需要在非请求环境（比如脚本）中调用model，可以手动推送上下文：

from api.app import create_app
app = create_app()
with app.app_context():
    # 在这里调用UserModel的方法
    UserModel.get_by_username("test")

方案二：让Database类实现单例模式 #

通过单例模式确保整个应用中只有一个Database实例，create_app中初始化一次后，model里获取的就是同一个配置好的实例。

修改common/database.py实现单例： #

import MySQLdb

class Database:
    _instance = None  # 存储单例实例
    conn = None
    cur = None
    conf = None

    def __new__(cls, **kwargs):
        # 只创建一次实例
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, **kwargs):
        # 仅第一次初始化时配置连接
        if self.conf is None:
            self.conf = kwargs
            self.conf['keep_alive'] = kwargs.get('keep_alive', False)
            self.conf['charset'] = kwargs.get('charset', 'utf8')
            self.conf['host'] = kwargs.get('host', 'localhost')
            self.conf['port'] = kwargs.get('port', 3306)
            self.conf['autocommit'] = kwargs.get('autocommit', False)
            self.conf['ssl'] = kwargs.get('ssl', False)
            self.connect()

    # ... 原有的connect、getOne等方法保持不变 ...

修改models/user.py使用单例： #

import datetime
import json
import logging
from api.common.database import Database

class UserModel:
    # 直接获取单例实例（create_app中已经初始化过配置）
    db = Database()

    @classmethod
    def get_by_username(cls, username=None):
        user = cls.db.getOne(
            table='users',
            fields=['user_id','data'],
            where=('username = %s', [username])
        )
        if user:
            user['data'] = json.loads(user['data'])
        return user

说明：

这种方式适合简单场景，但要注意：如果你的应用是多线程/多进程部署，当前的conn是实例属性，可能会出现连接共享的问题，需要额外处理连接的线程隔离（比如每次请求前检查连接状态，或者为每个线程创建独立连接）。

方案三：依赖注入（耦合度较高，不推荐） #

在资源层把数据库实例传递给model方法，虽然能解决问题，但会增加代码耦合度：

# resources/user.py 中修改post方法
def post(self):
    args = self.reqparse.parse_args()
    # 从current_app获取db并传给model
    user = UserModel.get_by_username(args['username'], db=current_app.db)
    return {'message': 'Wrong credentials'}

# models/user.py 中修改方法
@classmethod
def get_by_username(cls, username=None, db=None):
    user = db.getOne(
        table='users',
        fields=['user_id','data'],
        where=('username = %s', [username])
    )
    if user:
        user['data'] = json.loads(user['data'])
    return user

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