将 Python Flask App 配置为使用“create_app”工厂并在模型类中使用数据库

Question

我在使用 create_app() 函数时无法启动我的应用程序。我不熟悉以这种方式构建应用程序，从我所有的研究来看，我的方法似乎有所不同，因为我使用的是我自己的数据库包装器而不是 SQLAlchemy——这很容易，因为可以使用db.init_app(app)。

我的问题是：我似乎无法访问 /models/user.py 中的数据库连接...我该如何解决这个问题，以便可以使用该文件中的数据库连接？

这是我的应用程序文件夹结构，后面是列出的文件：

/api
    /common
        database.py
    /models
        user.py
    /resources
        user.py
    app.py
run.py

这是我的文件

#
#   File: run.py
#

from api.app import create_app

app = create_app(debug=True)
app.run(
    host=app.config['APP_HOST'],
    port=app.config['APP_PORT'],
    debug=app.config['APP_DEBUG_FLASK'],
    ssl_context=app.config['APP_SSL_CONTEXT']
)

#
#   File: app.py
#

from logging.config import dictConfig

from flask import Flask
from flask_restful import Api
from api.config import LocalConfig, LiveConfig
from api.extensions import bcrypt, cors, jwt
from api.resources.user import *
from api.common.database import Database

def create_app(debug=True):
    config = LocalConfig if debug else LiveConfig

    # Create app
    app = Flask(__name__)

    # Set configuration variables
    app.config.from_object(config)
    app.secret_key = app.config['APP_SECRET_KEY']
    app.url_map.strict_slashes = False  

    # Create api
    api = Api(app, prefix='/api/v2')

    # Initializing the logger
    dictConfig(app.config['LOGGING'])

    # Connect to mysql
    db = Database(
        host=app.config['MYSQL_HOST'],
        db=app.config['MYSQL_DB'],
        user=app.config['MYSQL_USER'],
        passwd=app.config['MYSQL_PASS'],
    )

    register_decorators(app)
    register_extensions(app)
    register_endpoints(api)

    return app

def register_extensions(app):
    bcrypt.init_app(app)
    jwt.init_app(app)

def register_endpoints(api):
    api.add_resource(UserLogin, '/login')

#
#   File: /resources/user.py
#

from flask_restful import Resource, reqparse
from api.models.user import *

class UserLogin(Resource):

    def __init__(self):
        self.reqparse = reqparse.RequestParser()
        self.reqparse.add_argument('username', type=str, required=True, 
                                    help='Username is required.',
                                    location='json')
        self.reqparse.add_argument('password', type=str, default='', location='json')

    def post(self):     
        args = self.reqparse.parse_args()
        print(args['username'])
        user = UserModel.get_by_username(args['username'])
        return {'message': 'Wrong credentials'}

#
#   File: /models/user.py
#

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

class UserModel:

    @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

#
#   File: /common/database.py
#

import MySQLdb

class Database:
    conn = None
    cur = None
    conf = None

    def __init__(self, **kwargs):
        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()

    def connect(self):
        try:
            if not self.conf['ssl']:
                self.conn = MySQLdb.connect(db=self.conf['db'], 
                                        host=self.conf['host'],
                                        port=self.conf['port'], 
                                        user=self.conf['user'],
                                        passwd=self.conf['passwd'],
                                        charset=self.conf['charset'])
            else:
                self.conn = MySQLdb.connect(db=self.conf['db'], 
                                        host=self.conf['host'],
                                        port=self.conf['port'], 
                                        user=self.conf['user'],
                                        passwd=self.conf['passwd'],
                                        ssl=self.conf['ssl'],
                                        charset=self.conf['charset'])

            self.cur = self.conn.cursor(MySQLdb.cursors.DictCursor)
            self.conn.autocommit(self.conf['autocommit'])
        except:
            print ('MySQL connection failed')
            raise

    def getOne(self, table=None, fields='', where=None, order=None, limit=(1,)):
        ### code that handles querying database directly ###

Answer 1

我已经开始转向使用 Miguel Grinberg 在其Flask Mega-Tutorial 的第 XV 部分中介绍的 create_app 模式的形式。

在您的情况下，对 db 对象的引用锁定在 create_app 内部的局部变量中。诀窍是让它可见。考虑一下这个方案，我将它与 SQLAlchemy 一起使用，并且你会适应它来使用你的包装器：

main.py
config.py
tests.py
app/
    __init__.py

首先，看起来像这样的默认配置（为简洁起见）

# config.py
...
class Config:
    TESTING = False
    SQLALCHEMY_DATABASE_URI = ...
    ...

该配置将被工厂用作默认配置。

# app/__init__.py
...
db = SQLAlchemy()  # done here so that db is importable
migrate = Migrate()

def create_app(config_class=Config):
    app = Flask(__name__)
    app.config.from_object(config_class)
    db.init_app(app)
    migrate.init_app(app, db)
    ... register blueprints, configure logging etc.
    return app

请注意，from app import db 有效。

# main.py
from app import create_app
app = create_app()

然后FLASK_APP=main.py venv/bin/flask run。

为了测试，Config 的子类使用内存数据库（并进行其他调整以避免影响外部服务）。

# tests.py
...
class TestConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite://'

class ExampleTests(unittest.TestCase):
    def setUp(self):
        self.app = create_app(TestConfig)
        # See Grinberg's tutorial for the other essential bits