wake-up-neo.com

Konfigurieren Sie die Python Flask-App für die Verwendung der Factory "create_app" und verwenden Sie die Datenbank in der Modellklasse

Ich habe Probleme, meine App zu starten, wenn ich eine Funktion create_app() verwende. Ich bin neu darin, Apps auf diese Weise zu erstellen, und nach all meinen Recherchen scheint mein Ansatz anders zu sein, da ich meinen eigenen Datenbank-Wrapper und nicht SQLAlchemy verwende - was es einfach macht, weil db.init_app(app) verwendet werden kann.

Meine Frage ist: Ich kann scheinbar nicht auf meine Datenbankverbindung in /models/user.py zugreifen. Wie behebe ich das, damit ich die Datenbankverbindung in dieser Datei verwenden kann?

Dies ist meine Ordnerstruktur für die App, gefolgt von den folgenden Dateien:

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

Hier sind meine Dateien

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

Ich habe angefangen, die Form des create_app-Musters zu verwenden, das Miguel Grinberg in Teil XI seines Flask Mega-Tutorials illustriert.

In Ihrem Fall ist der Verweis auf Ihre db in einer lokalen Variablen innerhalb von create_app gesperrt. Der Trick ist, es sichtbar zu machen. Betrachten Sie dieses Schema, das ich mit SQLAlchemy verwende und das Sie anpassen würden, um Ihren Wrapper zu verwenden:

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

Erstens die Standardkonfiguration, die ungefähr so ​​aussieht (der Kürze halber getrimmt)

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

Die Konfiguration wird ab Werk als Standardeinstellung verwendet.

# 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

Beachten Sie, dass from app import db funktioniert.

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

Und dannFLASK_APP=main.py venv/bin/flask run.

Zum Testen verwendet eine Unterklasse von Config eine speicherinterne Datenbank (und nimmt andere Optimierungen vor, um zu vermeiden, dass externe Dienste beeinträchtigt werden).

# 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
1
Dave W. Smith