How to bite Flask, SQLAlchemy and pytest all at once
As a person who started with Django, I had some hard time figuring out how application and request contexts work in Flask. I think I finally got to understand them. I also learnt how to use SQLAlchemy’s scoped sessions properly and how to test my REST applications with pytest efficiently.
Here’s what I got to know, split into topics.
- How does SQLAlchemy handle sessions
- Transactions in SQLAlchemy
- What are Flask application and request contexts
- Zipping Flask request contexts and SQLAlchemy scoped sessions together
- Testing everything
- Sewing it all together: Flask, SQLAlchemy and pytest
How does SQLAlchemy handle sessions
Session is the main place where you usually talk to
This code is perfectly fine if used non-concurrently. Non-concurrently, that means only one user can use session at once; that means, only one user connects to your website at once.
In reality that’s not always a case. I bet you want your site to be safely accessible by many users all at once.
If so, then you can’t use
sessionmaker alone. It’s not
to create safe
Session objects for multiple threads.
So what’s the solution? It’s actually pretty clever. It’s called scoped sessions: sessions, that are bound to the specific “scope” of your application, for example: the scope of one user’s connection.
It’s quite easy to understand how session for application scope is being
constructed. For every incoming request, a different
Session object is
So user Mark gets session A, user Ellen gets session B and user Sam gets session C. The key is that all these sessions are accessible in your code via the very same line:
(Look at the previous snippet; they’re almost the same!)
All the required setup is this one little object, scoped_session.
You feed it with some session-factory-maker, like
There’s only one part missing… how does scoped session know when to “spawn” a different session? It somehow has to recognize that a new user is requesting your views.
I’ll come back to that in later after explaining what are Flask application and request contexts and how to work with them.
Transactions in SQLAlchemy
SQLAlchemy supports at least two different kinds of transactions. The most popular type is Session based transaction:
The second type is more superior. It can roll back even committed session changes! It’s really powerful for testing purposes.
What are Flask application and request contexts
I like to think about Flask application context as being bound to one thread of your actual application (website). That context might be a set of global objects, like database connection and app settings. These objects should only exist once per your application, right? (I don’t see a point in duplicating app settings or database connections all over the place).
In Flask, current_app is aware of the active application context. If you have your web application running on two threads, and one user accesses the first thread, they’ll use different Flask application than the other user accessing second thread.
Request context is very similar to the application context. Every time anyone goes to some page on your site (ie. sends request), a new context is created.
This new context holds information that should only be available within that particular second when the user is being served. I’m assuming you can serve your user within one second :)
For example, imagine you have a view that adds a new blog post to your site:
Flask internals ensure that you do not access a different request’s data. Two requests may be simultaneous, yet you will access exactly the correct request in your code.
Zipping Flask request contexts and SQLAlchemy scoped sessions together
So now you know what powers Flask contexts and that you should choose
scoped SQLAlchemy sessions over “normal” ones. But how to make a
scoped_session that works with Flask contexts?
scopefunc– optional function which defines the current scope. If not passed, the
scoped_sessionobject assumes “thread-local” scope, and will use a Python
threading.local()in order to maintain the current
Session. If passed, the function should return a hashable token; this token will be used as the key in a dictionary in order to store and retrieve the current
scopefunc has to unambiguously represent each individual
context. I was looking for a good way of handling that, and found one in
internal context stack to build hashable context tokens. The code looks
Because of the aforementioned flexibility that Flask and SQLAlchemy have, I had really hard time figuring how to test the whole thing. Testing is very important, and with the help of wonderful Python libraries like pytest it actually becomes a pleasure.
The biggest change is in the ideology: now you don’t have to write classes (test cases) to test your code. You can write a lot simpler functions instead.
A fixture is a function that, for example, returns a database session object, which can be leveraged by your tests.
Or it can return a file descriptor to the file in
your application object. Or Redis connection object.
Look at fixtures docs for more examples.
Every fixture can be set for a
module scope, or
function scope. This means, that the fixture is only run once per
testing session, or once per whole module (containing tests), or once
for every test function.
Take for example this
It’s dumb and won’t work, but I hope you get the gist. Even if you have a thousand tests that use this fixture, it will be invoked only once, then memorized (cached).
I suggest to (at least) create a session-scoped fixture that builds your Flask application object (using application factory), and a session-scoped fixture that builds your SQLAlchemy session and manages transactions.
Transactions in tests
Shortly: it’s way faster to roll back all the changes from database than to recreate whole database from scratch on every new test.
I really like the fixtures
that leverage Python’s
yield, the above fixture example looks a lot clearer now:
Sewing it all together: Flask, SQLAlchemy and pytest
- Use application factory to easily create Flask application object. This will be used in different parts of your codebase, like tests or dev server.
- Create global scoped_session object that spawns actual SQLAlchemy
sessions when accessed. Use
scopefunckeyword argument to provide hashable function that’s used to recognize context switches.
- Don’t bind that
scoped_sessionobject to any engine yet. Bind it in your application factory using scoped_session.configure().
app.teardown_appcontextremove database sessions.
Tests with pytest
This one’s more complicated, so I’ll paste some boilerplate below.
- In your
conftest.pyprepare one session-scoped fixture that creates your app (using factory), creates all the tables, explicitely pops a connection, binds global
scoped_sessionto that connection and yields that app
- Prepare second fixture, that creates a new transaction, new application context and yields database session.
Here’s that promised boilerplate. First application factory
And test configuration from