将Python Flask应用配置为使用"create_app"工厂并在模型类中使用数据库 [英] Configure Python Flask App to use "create_app" factory and use database in model class

问题描述

使用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 ###

解决方案

我已开始迁移使用Miguel Grinberg在其create_app模式的形式. blog.miguelgrinberg.com/post/the-flask-mega-tutorial-part-xi-facelift"rel =" noreferrer> Flask Mega-Tutorial .

在您的情况下，对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

I'm having trouble getting my app to start when using a create_app() function. I'm new to building apps to this way and from all of my research it seems that my approach is different because I'm using my own database wrapper and not SQLAlchemy—which makes it easy because db.init_app(app) can be used.

My question is: I can't seem to access my database connection in /models/user.py... how do I fix this so I can use the db connection in that file?

This is my folder structure for the app, followed by those files listed:

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

Here are my files

#
#   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 ###

解决方案

I've started migrating toward using the form of the create_app pattern that Miguel Grinberg illustrations in part XI of his Flask Mega-Tutorial.

In your case, the reference to your db object locked in a local variable inside of create_app. The trick is to get it visible. Consider this scheme, which I use with SQLAlchemy, and which you would adapt to use your wrapper:

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

First, the default configuration which looks something like this (trimmed for brevity)

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

The config will be used as a default by the factory.

# 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

Note that from app import db works.

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

And thenFLASK_APP=main.py venv/bin/flask run.

And for testing, a subclass of Config uses an in-memory database (and makes other tweaks to avoid hitting external services).

# 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

这篇关于将Python Flask应用配置为使用"create_app"工厂并在模型类中使用数据库的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！