Transactional Unit Tests with Pytest and Async SQLAlchemy
This is a quick how-to post if you’re working on migrating your database-backed applications to async-sqlalchemy and are looking to get a test framework in place.
TL;DR
If you just want the code, this is it.
Understanding Session, Engine and the Connection object
Connection
This is the lowest abstraction that lets us execute queries and create transactions and for a lot of tasks could be just about enough.
Engine
Engine is a layer of abstraction that let’s create connection pools. This layer lets you choose your pooling implementations from QueuePool, StaticPool among many others. However, it primarily offers a means to access the connection object.
The engine offers a convenience method execute which is the same execute method that we call on the connection object.
This could also be done directly on the engine
And we have the same sequence of operations.
Session
Session lets us work with ORMs. It’s the highest layer of abstraction and offers a similar interface with the execute method.
Transactional Tests and Fixtures with SQLAlchemy
The code above creates a top-level transaction followed by a savepoint. The code then yields the session to the application code, which is our tests in this case.
Now,
Our application can either COMMIT or open another nested transaction within this savepoint
We ensure that there is always an active savepoint by using the event listener to check for a transaction ending and open a new one.
Finally, when all the transaction work is done, we check for an open nested transaction and roll it back.
We also rollback the top-level transaction restoring the database state back to what it was at the beginning of the test function.
It’s not necessary that we rollback the nested transaction identified in point 3 as the top-level rollback will rollback all sub transactions.
Async SQLAlchemy
For an asynchronous application most of our structure remains the same except that we use pytest-asyncio to create async fixtures.
The catch here is that there are no event handlers for async after_transaction_end events. So, we use the synchronous session handler to ensure the closed save_point is re-opened.
Note, that we generate all the transactions on the connection object. The async session object joins in on the open transactions.
Event Loop Closed Error
Now, depending on the version of the libraries you've used you may encounter the event loop closer error.
I've also encountered the Task Pending error.
Essentially, this could be summed up into 2 scenarios, where in one, the event loop is closed before the last operation is invoked and in the second, a new event loop is created for a specific operation.
To solve this, we create another fixture which loads a single event loop for a session. The event loop is yielded at the start of the session and closed when the suite is completed. We add this to conftests.py.
