11

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

1 Answer 1

15

I've started migrating toward using the form of the create_app pattern that Miguel Grinberg introduces in part XV 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
2
  • Hey Dave, thanks for suggesting an answer! So there are lots of tutorials out there exactly like what you've supplied, however, the problem is that I'm not using SQLAlchemy... I would have to "adapt" the wrapper, but the question is how because it would seemingly require building a flask extension and utilizing the init_app method. So far my solution has been to attach the db connection to the app like this: app.db = Database(...) and then to utilize Flask's current_app: current_app.db. Not sure this is a good way to do it however...
    – stwhite
    Commented May 16, 2018 at 5:56
  • If you make your db wrapper importable, which is what the pattern above does, there's no reason to attach it to the app object. And depending on your wrapper, db.init_app() might be optional, but it's worth takeing a look at how that's implemented in the SQLAlchemy wrapper to make sure you wouldn't be missing someting important. Commented May 16, 2018 at 16:32

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.