Instantly share code, notes, and snippets.

What would you like to do?
import os
import pytest
from alembic.command import upgrade
from alembic.config import Config
from project.factory import create_app
from project.database import db as _db
TESTDB = 'test_project.db'
TESTDB_PATH = "/opt/project/data/{}".format(TESTDB)
ALEMBIC_CONFIG = '/opt/project/alembic.ini'
def app(request):
"""Session-wide test `Flask` application."""
settings_override = {
'TESTING': True,
app = create_app(__name__, settings_override)
# Establish an application context before running the tests.
ctx = app.app_context()
def teardown():
return app
def apply_migrations():
"""Applies all alembic migrations."""
config = Config(ALEMBIC_CONFIG)
upgrade(config, 'head')
def db(app, request):
"""Session-wide test database."""
if os.path.exists(TESTDB_PATH):
def teardown():
os.unlink(TESTDB_PATH) = app
return _db
def session(db, request):
"""Creates a new database session for a test."""
connection = db.engine.connect()
transaction = connection.begin()
options = dict(bind=connection, binds={})
session = db.create_scoped_session(options=options)
db.session = session
def teardown():
return session
from flask.ext.sqlalchemy import SQLAlchemy, SignallingSession, SessionBase
class _SignallingSession(SignallingSession):
"""A subclass of `SignallingSession` that allows for `binds` to be specified
in the `options` keyword arguments.
def __init__(self, db, autocommit=False, autoflush=True, **options): = db.get_app()
self._model_changes = {}
self.emit_modification_signals = \['SQLALCHEMY_TRACK_MODIFICATIONS']
bind = options.pop('bind', None)
if bind is None:
bind = db.engine
binds = options.pop('binds', None)
if binds is None:
binds = db.get_binds(
class _SQLAlchemy(SQLAlchemy):
"""A subclass of `SQLAlchemy` that uses `_SignallingSession`."""
def create_session(self, options):
return _SignallingSession(self, **options)
db = _SQLAlchemy()

This comment has been minimized.

chekunkov commented Oct 23, 2015

Looks like signalling session from flask-sqlalchemy should work out of the box now mitsuhiko/flask-sqlalchemy@ac135c6


This comment has been minimized.

krmarien commented Nov 7, 2015

Thank you to share this example code! This is really helpful.
Would it be possible to share also an example of the project.factory module?


This comment has been minimized.

vlttnv commented Dec 25, 2015

I got the following error when I was trying this example (excluding the custom DB file):
TypeError: __init__() got multiple values for keyword argument 'binds'
What fixed it for me is removing binds on Line 68


This comment has been minimized.

flyte commented Mar 10, 2016

I'm having the same issue as @krmarien - the example won't work without that factory function!


This comment has been minimized.

asfaltboy commented Apr 5, 2016

Hi @krmarien and @flyte in case you're still searching, my solution was to substitute app = create_app(__name__, settings_override) with:

app = Flask(__name__)

Which is what, I assume, the function must do, given it's name and arguments.

And of course, the change @vlttnv mentioned is also required, since the gist were probably written with an older version of flask-sqlalchemy in mind, i.e pre- mitsuhiko/flask-sqlalchemy@ac135c6 as @chekunkov mentions. After these changes, this recipe works like a charm. Thanks @alexmic !

There I did it, I mentioned everyone in the thread 🎉 !



This comment has been minimized.

Tim-Erwin commented Apr 28, 2016

I have code that imports the session from a directly. Overwriting that (models.session = ourNewSession) does not work since it's already imported. However, since it's a ScopedSession, we can trick it:

def session(db, request):
    session = db.create_scoped_session(options={'bind': connection, 'binds': {}})
    def getSession():
        return session
    models.session.registry = getSession

This comment has been minimized.

aphillipo commented Jun 15, 2016

Is it possible to change the app to used the session with rollback too? It doesn't look like this will work with app.test_client() for example.


This comment has been minimized.

MarSoft commented Jul 23, 2016

@chekunkov, the commit you pointed to didn't actually help: empty dict {} is still recognized as False and is replaced with db.get_binds(app) anyway. The only actual change that commit introduced is that db.get_binds() function is not called if binds argument passed and is not-False.
I tried to pass "mock" value like {None: None} which would not be recognized as False, but it didn't help and instead caused an error.
So there is still a problem in flask-sqlalchemy SignallingSession contructor. Will try to make a pull request.


This comment has been minimized.

MarSoft commented Jul 23, 2016

Hm, they already fixed it in upstream, but current version 2.1 still didn't receive that fix.
Here is my approach for work-around:

from flask_sqlalchemy import SignallingSession

def wrap_signalling_session(cls):
    _original_init = cls.__init__
    def _fixed_init(self, *args, **kwargs):
        binds = kwargs.get('binds')
        if binds == {}:
            def empty(*args, **kwargs):
            self._add_bind = empty  # override class' method with a function
        _original_init(self, *args, **kwargs)
        if binds == {}:
            del self._add_bind
    cls.__init__ = _fixed_init

This comment has been minimized.

MarSoft commented Jul 23, 2016

Another catch: the approach described in the article is great, but it has a non-obvious caveat - took me several hours to debug.
Here is the problem: if your code always uses db.session, everything will work fine. But the famous Flask-Admin extension expects you to pass a session object when you instantiate a flask_admin.contrib.sqla.ModelView. As a result, it will use original session (the scoped_session object which was created during app initialization), not test-specific one.
I worked it around using LocalProxy to access current session:

from werkzeug.local import LocalProxy

db = ...

def init_app():
    session = LocalProxy(lambda: db.session)
    admin = ...
    admin.add_view(ModelView(User, session))

As a result, it always uses the correct session.


This comment has been minimized.

skortchmark9 commented Jul 28, 2016

I didn't want to have to change my application code to fix my tests, so I opted for the hack suggested here and faithfully reproduced for your convenience:

class _dict(dict):
    def __nonzero__(self):
        return True

connection = db.engine.connect()
transaction = connection.begin()
options = dict(bind=connection, binds=_dict())
session = db.create_scoped_session(options=options)

Thanks for the help y'all!


This comment has been minimized.

stefanprobst commented Oct 13, 2016

Could somebody please explain to me why the line = app is needed in the db fixture?


This comment has been minimized.

ben4ever commented Nov 1, 2016

@stefanprobst, it seems like he is doing that to replace the live app instance with the test app instance which he creates with the app fixture and which has specific test related config values set in settings_override (namely TESTING and SQLALCHEMY_DATABASE_URI).
So without that line the _SQLAlchemy instance wouldn't have the proper config values set for the test environment.


This comment has been minimized.

ev-agelos commented Jan 8, 2017

I was using Flask 0.11 and Line 76: session.remove() wasnt working, updated to Flask==0.12(latest as for now) and worked.
(Probably this Flask PR has to do with the fix.)
(spent some hours until i realize why it didn't work in case anybody else wastes time on that)


This comment has been minimized.

jacol12345 commented Jul 12, 2017

I have a similar use case. However, I don't need to use transaction for each test, I'm fine with new DB each time. Did anyone succeed at getting it work without subclassing SignallingSession? Sample fixtures which fail with in-memory SQLite (I did not test it with other DB yet).

def app():
    """An application for the tests."""
    _app = create_app(TestConfig)
    ctx = _app.test_request_context()

    yield _app


def db(request, app): = app
    with app.app_context():
        from flask_migrate import upgrade

    yield _db

    # Explicitly close DB connection

I can confirm that Alembic migration gets executed, yet tests still fail with "no such table" exceptions.


This comment has been minimized.

skftn commented Aug 28, 2017

Would be nice if someone could post a full working example so I/we can observe:

  • The flask directory structure
  • Where and how the models are defined (what Base do they use, etc)
  • Where and how is made
  • How to run the flask application without using pytest

Given the current snippets of code, it is hard to 'guess' how the rest of the application should look like. Instead of a gist with 2 files, a full working Flask application would be best.

One example, snippet has db = _SQLAlchemy(), however, SQLAlchemy should be called with an app parameter. How is this suppose to work?

@alexmic what's the point of writing a tutorial if you do not give the full code? :)


This comment has been minimized.

gaffney commented Aug 18, 2018

I second that, I came here expecting to see a fully working example and am instead leaving with a bunch of question marks... this snippet requires too much guessing and imagination.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment