如何用SQLAlchemy设置Flask应用程序进行测试？

Question

似乎在瓶通常的做法是这样开始：

from flask import Flask 
from flaskext.sqlalchemy import SQLAlchemy 
app = Flask(__name__) 
SQLALCHEMY_DATABASE_URI = 'something' 
app.config.from_object(__name__) 
db = SQLAlchemy(app) 

，然后导入和使用app和db无处不在。但是当你像这样创建db时，它会从应用中获取配置，而且看起来这种配置一旦发生就不能被覆盖。在Flask的网站上有一些关于制作应用程序工厂的网页，但是如果我这样做的话，我还不清楚如何在任何地方仍然可以使用app和db。

如何编写脚本来测试具有不同数据库的Flask应用程序？我应该如何构建我的应用程序才能实现这一点？我必须使用module s吗？

Answer 1

你本能地使用环境变量是正确的。但是，使用错误的db运行单元测试存在一些危险。此外，您可能不希望connect_db与每个请求和您想要使用的任何地方db。您可以使用明确设置的配置目录和环境变量。这是迄今为止我所见过的最好的。

run.py 
shell.py 

config/__init__.py 
config/test.py 
config/postgres.py 
... 

main/__init__.py 
main/someapp/__init__.py 
main/someapp/models.py 

... 
main/tests/__init__.py 
main/tests/testutils.py 

因此，该配置文件可能是：

# config/test.py 
SQLALCHEMY_DATABASE_URI = "sqlite://" 

和

# config/postgres.py 
SQLALCHEMY_DATABASE_URI = 'postgresql://user:pw@localhost/somedb' 

所以，我可以明确地设置数据库在我的基地测试用例：

import os 
from flask.ext.testing import TestCase 

os.environ["DIAG_CONFIG_MODULE"] = "config.test" 
from main import app, db 


class SQLAlchemyTest(TestCase): 

    def create_app(self): 
     return app 

    def setUp(self): 
     db.create_all() 

    def tearDown(self): 
     db.session.remove() 
     db.drop_all() 

然后，main/__init__.py，对我来说：

import os 

from flask import Flask, render_template, g 
from flask.ext.sqlalchemy import SQLAlchemy 

# by default, let's use a DB we don't care about 
# but, we can override if we want 
config_obj = os.environ.get("DIAG_CONFIG_MODULE", "config.test") 
app = Flask(__name__) 
app.config.from_object(config_obj) 
db = SQLAlchemy(app) 

@app.before_request 
def before_request(): 
    g.db = db 
    g.app = app 

# ... 
@app.route('/', methods=['GET']) 
def get(): 
    return render_template('home.html') 
# ...  
from main.someapp.api import mod as someappmod 
app.register_blueprint(someappmod) 

然后，在其他文件，在那里我知道我要运行的配置，有可能：

# run.py 
import os 
os.environ["DIAG_CONFIG_MODULE"] = "config.postgres" 
from main import app 
app.run(debug=True) 

和

# shell.py 
import os 
os.environ["DIAG_CONFIG_MODULE"] = "config.postgres" 

from main import app, db 
from main.symdiag.models import * 
from main.auth.models import * 
print sorted(k for k in locals().keys() if not k.startswith("_")) 
import IPython 
IPython.embed() 

也许..迄今为止最好的：P。

Answer 2

您不想在导入时连接到数据库。继续在导入时配置您的应用程序，因为在尝试测试或运行您的应用程序之前，您始终可以在测试中调整配置。在下面的示例中，您将在使用应用程序配置的某些函数背后连接数据库连接，因此在unittest中，您可以实际更改数据库连接以指向其他文件，然后继续并在您的设置中明确连接。

假设你有一个包含myapp.py它看起来像一个MYAPP包：

# myapp/myapp.py 
from __future__ import with_statement 
from sqlite3 import dbapi2 as sqlite3 
from contextlib import closing 
from flask import Flask, request, session, g, redirect, url_for, abort, \ 
    render_template, flash 

# configuration 
DATABASE = '/tmp/flaskr.db' 
DEBUG = True 
SECRET_KEY = 'development key' 
USERNAME = 'admin' 
PASSWORD = 'default' 

# create our little application :) 
app = Flask(__name__) 
app.config.from_object(__name__) 
app.config.from_envvar('MYAPP_SETTINGS', silent=True) 

def connect_db(): 
    """Returns a new connection to the database.""" 
    return sqlite3.connect(app.config['DATABASE']) 


def init_db(): 
    """Creates the database tables.""" 
    with closing(connect_db()) as db: 
     with app.open_resource('schema.sql') as f: 
      db.cursor().executescript(f.read()) 
     db.commit() 


@app.before_request 
def before_request(): 
    """Make sure we are connected to the database each request.""" 
    g.db = connect_db() 


@app.after_request 
def after_request(response): 
    """Closes the database again at the end of the request.""" 
    g.db.close() 
    return response 

@app.route('/') 
def show_entries(): 
    cur = g.db.execute('select title, text from entries order by id desc') 
    entries = [dict(title=row[0], text=row[1]) for row in cur.fetchall()] 
    return render_template('show_entries.html', entries=entries) 

if __name__=="__main__": 
    app.run() 

您的测试文件的myapp/test_myapp.py看起来就像这样：

import os 
import myapp 
import unittest 
import tempfile 

class MyappTestCase(unittest.TestCase): 

    def setUp(self): 
     self.db_fd, myapp.app.config['DATABASE'] = tempfile.mkstemp() 
     self.app = myapp.app.test_client() 
     myapp.init_db() 

    def tearDown(self): 
     os.close(self.db_fd) 
     os.unlink(myapp.app.config['DATABASE']) 

    def test_empty_db(self): 
     rv = self.app.get('/') 
     assert 'No entries here so far' in rv.data 

当然，如果你”喜欢使用SQLAlchemy，你必须适当地更新connect_db和init_db函数，但希望你能明白。

Answer 3

首先，不是直接在脚本中实例化Flask应用程序，而是使用application factory。这意味着你创建一个函数，将你的配置文件作为参数，并返回实例化的应用程序对象。然后，创建不带参数的全局SQLAlchemy对象，并在创建应用程序时对其进行配置，即as explained here。

db = SQLAlchemy() 

def create_app(configfile): 
    app = Flask(__name__) 

    app.config.from_pyfile(config, silent=True) 
    db.init_app(app) 

    # create routes, etc. 

    return app 

运行应用程序，你只需做这样的事情：

app = create_app('config.py') 
app.run() 

要运行单元测试，你可以这样做：

class Test(TestCase): 
    def setUp(self): 
     # init test database, etc. 
     app = create_app('test_config.py') 
     self.app = app.test_client() 
    def tearDown(self): 
     # delete test database, etc. 

就我而言，我使用SQLAlchemy直接使用scoped_session而不是Flask-SQLAlchemy。 我做了同样的事情，但用Lazy SQLAlchemy setup。