Настройте приложение Python Flask для использования фабрики "create_app" и использования базы данных в классе модели

Question

У меня проблемы с запуском приложения при использовании функции create_app(). Я новичок в создании приложений таким образом, и из всех моих исследований кажется, что мой подход отличается от других, потому что я использую свою собственную оболочку базы данных, а не SQLAlchemy, что облегчает задачу, поскольку можно использовать db.init_app(app).

У меня такой вопрос: я не могу получить доступ к своему соединению с базой данных в /models/user.py ... как я могу это исправить, чтобы я мог использовать соединение db в этом файле?

Это моя структура папок для приложения, за которой следуют перечисленные файлы:

/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

Я начал переходить на использование формы create_app, которую Мигель Гринберг иллюстрирует в части XI своего Мега-учебника по настойке .

В вашем случае ссылка на ваш 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