If you are using the new Flask configuration together with the application factory pattern, one thing you will want to do if using SQLAlchemy is initialize a SQLAlchemy session for different requirements. For example, for unit tests you don't want to use the production database.
Furthermore you may need to use a SQLAlchemy session outside the request scope, for example in the shell.
The scoped_session function, which provides a thread-safe session, expects a factory function. Normally we would use sessionmaker:
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
engine = create_engine("sqlite:///myapp.db")
db_session = scoped_session(sessionmaker(bind=engine))
In order to initialize SQLAlchemy dynamically however we need to pass in a factory function that does not require a ready engine instance. For this we can use create_session:
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, create_session
engine = None
db_session = scoped_session(lambda: create_session(bind=engine))
We then need a function to create the engine when needed:
def init_engine(uri, **kwargs):
global engine
engine = create_engine(uri, **kwargs)
return engine
Provided you call init_engine first you can then use db_session thereafter, as the scoped_session will bind the session to the current value of engine.
You can call init_engine in your application factory function:
def create_app(config):
app = Flask(__name__)
app.config.from_pyfile(config)
init_engine(app.config['DATABASE_URI'])
return app
Now you can use the db_session anywhere in your application, as long as you first call create_app.
This snippet by Dan Jacob can be used freely for anything you like. Consider it public domain.
Comments
Using local context by Dan Jacob on 2010-05-28 @ 08:22
Another pattern is to attach the database engine to the application. Instead of the engine global, you can access the engine using current_app:
The pattern for accessing local context is described here:
http://flask.pocoo.org/docs/shell/#firing-before-after-request
In your unit tests (and elsewhere where you need to access the database outside the request context) you can do this:
import unittest from myapp import create_app from myapp.database import db_session, create_all, drop_all class TestCase(unittest.TestCase): def setUp(self): self.app = create_app("test.cfg") self.client = self.app.test_client() self.ctx = self.app.test_request_context() self.ctx.push() create_all() def tearDown(self): db_session.remove() drop_all() self.ctx.pop()Fix your settings by Amit Matani on 2012-01-22 @ 02:22
sessionmaker and create_session have opposite default arguments. If you take this route, make sure to set autoflush, autocommit, and expire_on_commit to the way you want them.
no go for session listener by sebastianmarkow on 2012-09-09 @ 12:14
beware if you want to use a session event listener like 'after_flush'. you will need to register it on the the Session object, thus you can't encapsulate create_session in a lambda function.
This is how I did it... by Donald Curtis on 2013-01-27 @ 17:57
engine = None sessionmaker = sa.orm.sessionmaker() session = sa.orm.scoped_session(sessionmaker)
def configure_engine(url): global sessionmaker, engine, session
engine = sa.create_engine(url) session.remove() sessionmaker.configure(bind=engine)
ACK! Don't know how to delete the previous comment. by Donald Curtis on 2013-01-27 @ 18:27
engine = None sessionmaker = sa.orm.sessionmaker() session = sa.orm.scoped_session(sessionmaker) def configure_engine(url): global sessionmaker, engine, session engine = sa.create_engine(url) session.remove() sessionmaker.configure(bind=engine)