When it comes to connecting to database in Flask project, we tend to use the Flask-SQLAlchemy extension that handles the lifecycle of database connection, add a certain of utilities for defining models and executing queries, and integrate well with the Flask framework. However, if you are developing a rather simple project with Flask and SQLAlchemy, and do not want to depend on another third-party library, or you prefer using SQLAlchemy directly, making the model layer agnostic of web frameworks, you can write your own extension. Besides, you will gain better type hints for SQLAlchemy model, and possibly easier migration to SQLAlchemy 2.x. This article will show you how to integrate SQLAlchemy 1.4 with Flask 2.1.
In the official document Flask Extension Development, it shows us writing a sqlite3 extension that plays well with Flask application context. So our first try is to replace sqlite3 with SQLAlchemy:
from typing import Optional
Several notes on our alpha version. First, it plays well with the Application Factories, that means the extension can be used to initialize multiple application instances, with different configurations for web server, testing, etc. The key point is to provide an
init_app method for different apps, and use the
current_app proxy during work. To initialize the app:
app = Flask(__name__)
Second, it plays well with The Application Context, by storing data on current app context’s stack, instead of on the extension instance, i.e.
self.some_attr. When the
engine attribute is first accessed, the extension creates a SQLAlchemy engine with the current app’s config, and stores it on the current app context. When this context is popped, the engine object is also disposed, releasing the connection pool. Flask will automatically push and pop application context during request and command line interface. Here is an example of CLI:
The major problem of the alpha version is constantly creating and disposing SQLAlchemy engine objects. And we known Engine is rather a heavy object to construct, and should be kept around throughout the lifespan of the application. There is an extension point of the Flask app instance, where we can store data for the entire app.
def init_app(self, app: Flask):
For working with SQLAlchemy, we often prefer using sessions, so in the extension we need to create a session for each request, and properly close it after the context is popped.
def connect(self) -> Session:
You may wonder why we don’t use
scoped_session, which is the recommended way to use sessions in a multi-thread environment. The answer is simple: an application context will not be shared by different workers, so it is safe to use the same session throughout the request. And, since session is a light-weight object, it is OK to create it on every request. Check Werkzeug Context Locals for more information.
Now that we have a simple but fully functional flask-sqlalchemy extension, we can start writing models in a native way.
from sqlalchemy import Column, Integer, String
There is no
db.Integer. The model base class need to be declared explicitly, as well as the table name of each model. Add a CLI that creates the tables:
And execute query in a view:
To enable type hints for SQLAlchemy models, install
sqlalchemy2-stubs and enable the plugin in
user.id will have the type
Column[Integer]. This will continue to work in SQLAlchemy 2.x, except no extra dependency is needed. You may want to read the document Mypy Support for ORM Mappings.
Source code in this article can be found on GitHub.
Flask’s built-in JSON serializer does not recoganize SQLAlchemy models, neither the frequently used
datetime objects. But we can easily enhance it with a custom encoder:
from decimal import Decimal
Row is for core engine use case, and
DeclarativeMeta for ORM. Add a line when creating the app:
app.json_encoder = CustomEncoder
If you are interested in supporting multiple binds like Flask-SQLAlchemy does, here is a proof of concept. But for such complex scenario, I suggest use the opensource extension instead, for it is more mature, feature-complete, and fully tested.
This time we do not create engine on app startup. We create scoped sessions on demand.
def connect(self, name: str) -> scoped_session:
The configuration style mimics Flask-SQLAlchemy. This version of
connect will return a properly configured
scoped_session object, and it will be shared among different workers, so we store it in the app’s
Note the creation of the
scoped_session object is not thread-safe, so we guard it with a lock. Again, this lock should not be stored as extension instance’s attribute, we create a
Holder class to hold both the lock and scoped sessions.
Do not forget to do the cleanup work:
def teardown(self, exception) -> None:
scoped_session.remove will invoke
close on the session and remove it from its registry. Next request will get a brand new session object.
We can verify if it uses the desired connection: