Using Flask SQLAlchemy

This guide will show the steps of migrating from flask-sqlalchemy to this package.

Background

Suppose that you are working with the codes written by Flask SQLAlchemy. You are planning to use Flask SQLAlchemy Lite to replace Flask SQLAlchemy. Indeed, the best way is to rewrite all codes using Flask SQLAlchemy Lite. However, some legacy APIs like Model.query or db.query_or_404 are not available in Flask SQLAlchemy Lite. In this case, you may want to make minimal changes to deliver a new version first, and gradually finalize the code migration plan in the future versions. Then, you can condsider to use this package.

The following codes show a complete project written by Flask SQLAlchemy, where the legacy APIs are used.

app.py

app.py

import sqlalchemy as sa
import flask
from flask import request
from werkzeug.exceptions import HTTPException

from models import db, Folder, Text


def create_app() -> flask.Flask:
    app = flask.Flask(__name__)

    @app.route("/")
    def index():
        """Index, quering all folder db items."""
        return {
            "message": "All folders found",
            "folders": [
                {"id": folder.id, "title": folder.title}
                for folder in db.session.scalars(sa.select(Folder).order_by(Folder.id))
            ],
        }

    @app.route("/folder")
    def folder():
        """Query a folder db item with all its texts."""
        folder_id = request.args.get("id", None, int)
        if folder_id is None:
            return flask.abort(404, "Folder ID is not specified.")
        folder = db.session.get(Folder, folder_id)
        if folder is None:
            return flask.abort(404, "Folder is not found.")
        return {
            "message": "Texts of the folder are found.",
            "folder": folder.id,
            "texts": [{"id": text.id, "title": text.title} for text in folder.texts],
        }

    @app.route("/text")
    def text():
        """Query a text db item."""
        text_id = request.args.get("id", None, int)
        if text_id is None:
            return flask.abort(404, "Text ID is not specified.")
        text = db.session.get(Text, text_id)
        if text is None:
            return flask.abort(404, "Text is not found.")
        return {
            "message": "Text is found.",
            "id": text.id,
            "title": text.title,
            "text": text.text,
        }

    @app.errorhandler(HTTPException)
    def handle_exception(exc: HTTPException):
        """Return JSON instead of HTML for HTTP errors."""
        response = flask.make_response(
            {
                "code": exc.code,
                "name": exc.name,
                "description": exc.description,
            },
            exc.code,
            exc.get_headers(),
        )
        response.content_type = "application/json"
        return response

    return app


def init_db_data() -> None:
    db.drop_all()
    db.create_all()

    folder_1 = Folder(title="Folder 1")
    folder_1.texts.append(Text(title="F1 Text 1", text="Lorem ipsum, ..."))
    folder_1.texts.append(Text(title="F1 Text 2", text="dolor sit amet, ..."))
    folder_1.texts.append(Text(title="F1 Text 3", text="consectetur adipiscing, ..."))
    folder_1.texts.append(Text(title="F1 Text 4", text="elit, sed do eiusmod, ..."))
    db.session.add(folder_1)

    folder_2 = Folder(title="Folder 2")
    folder_2.texts.append(Text(title="F2 Code", text="Code is 0887967868"))
    folder_2.texts.append(Text(title="F2 Report", text="The report of the ..."))
    db.session.add(folder_2)

    db.session.commit()


if __name__ == "__main__":
    app = create_app()
    app.config.update({"SQLALCHEMY_DATABASE_URI": "sqlite://"})
    db.init_app(app)

    with app.app_context():
        init_db_data()
    app.run(host="127.0.0.1", port=8080)

models.py

models.py

from typing import Optional

try:
    from typing import List
except ImportError:
    from builtins import list as List

import sqlalchemy as sa
import sqlalchemy.orm as sa_orm

from base import db


__all__ = ("db", "Text", "Folder")


class Text(db.Model):
    id: sa_orm.Mapped[int] = sa_orm.mapped_column(init=False, primary_key=True)
    title: sa_orm.Mapped[str] = sa_orm.mapped_column()
    text: sa_orm.Mapped[str] = sa_orm.mapped_column(sa.Text, deferred=True)

    folder_id: sa_orm.Mapped[Optional[int]] = sa_orm.mapped_column(
        sa.ForeignKey("folder.id"), default=None, nullable=True
    )
    folder: sa_orm.Mapped[Optional["Folder"]] = db.relationship(default=None)


class Folder(db.Model):
    id: sa_orm.Mapped[int] = sa_orm.mapped_column(init=False, primary_key=True)
    title: sa_orm.Mapped[str] = sa_orm.mapped_column()
    texts: sa_orm.Mapped[List[Text]] = db.relationship(
        default_factory=list, back_populates="folder"
    )

base.py

base.py

import sqlalchemy.orm as sa_orm
from sqlalchemy.pool import StaticPool

from flask_sqlalchemy import SQLAlchemy


__all__ = ("Base", "db")


class Base(sa_orm.MappedAsDataclass, sa_orm.DeclarativeBase):
    pass


engine_options = {
    "connect_args": {"check_same_thread": False},
    "poolclass": StaticPool,
}

db = SQLAlchemy(model_class=Base, engine_options=engine_options)

This application will hold a demo database in the memory, and provide several APIs for querying the example data. For example, when the program is running,

Address

Accessing the following address

http://127.0.0.1:8080/folder?id=1

will let users query the Folder item with id=1. All texts belonging to this folder will be listed, too.

Response

The response shown in the browser should be:

{
  "folder": 1,
  "message": "Texts of the folder are found.",
  "texts": [
    {
      "id": 1,
      "title": "F1 Text 1"
    },
    {
      "id": 2,
      "title": "F1 Text 2"
    },
    {
      "id": 3,
      "title": "F1 Text 3"
    },
    {
      "id": 4,
      "title": "F1 Text 4"
    }
  ]
}

The next step is to check the details of the text.

Address

Accessing the following address

http://127.0.0.1:8080/text?id=1

will let users query the Text item with id=1. This item has been shown in the previous response.

Response

The response shown in the browser should be:

{
  "id": 1,
  "message": "Text is found.",
  "text": "Lorem ipsum, ...",
  "title": "F1 Text 1"
}

Extend the compatibility with flask-sqlalchemy-lite

Migrating to this project will let codes working with the flask_sqlalchemy_lite backend. The change of the codes is quite simple. The only thing that needs to be altered is the base.py. See the following codes:

base.py

import sqlalchemy.orm as sa_orm
from sqlalchemy.pool import StaticPool

from flask_sqlalchemy import SQLAlchemy
import flask_sqlalchemy_compat as fsc


__all__ = ("Base", "db")


class Base(sa_orm.MappedAsDataclass, sa_orm.DeclarativeBase):
    pass


engine_options = {
    "connect_args": {"check_same_thread": False},
    "poolclass": StaticPool,
}

db = SQLAlchemy(model_class=Base, engine_options=engine_options)
db = fsc.get_flask_sqlalchemy(Base, engine_options=engine_options)

The main change is that we let db be given by the factory function fsc.get_flask_sqlalchemy, where the first argument of this function is the original base model.

This function will do nothing if flask_sqlalchemy is installed. In other words, db would be the same as the one created by flask_sqlalchemy.SQLAlchemy(), the first argument Base will be used as the model_class arugment when initializing the extension.

However, if flask_sqlalchemy is not installed, this function will attempt to find flask_sqlalchemy_lite and use flask_sqlalchemy_lite to mimic the behaviors of flask_sqlalchemy. In this case, db would be backended by flask_sqlalchemy_lite.SQLAlchemy(), while db.Model will be a modified Base class. The modification is inplace, so using db.Model is equivalent to using Base.

note

Note that the compatible model will take effect only when flask_sqlalchemy is not installed.

Explicitly use flask-sqlalchemy-lite to mimic flask-sqlalchemy

The above section shows the usage of fsc.get_flask_sqlalchemy(...). It will automatically switch to Flask SQLAlchemy Lite if the regular version is not installed. However, in some cases, users may want to deliberately use Flask SQLAlchemy Lite to mimic the behaviors of Flask SQLAlchemy. In that case, the script can be modified as follows:

base.py

import sqlalchemy.orm as sa_orm
from sqlalchemy.pool import StaticPool

from flask_sqlalchemy import SQLAlchemy
from flask_sqlalchemy_lite import SQLAlchemy
import flask_sqlalchemy_compat as fsc

__all__ = ("Base", "db")


class Base(sa_orm.MappedAsDataclass, sa_orm.DeclarativeBase):
    pass


engine_options = {
    "connect_args": {"check_same_thread": False},
    "poolclass": StaticPool,
}

db = SQLAlchemy(model_class=Base, engine_options=engine_options)
db = fsc.as_flask_sqlalchemy(
    db=SQLAlchemy(model_class=Base, engine_options=engine_options),
    model_class=Base
)

In this case, the database extension db will be notated by fsc.SQLAlchemyProxy which is a wrapper that provides the APIs of flask_sqlalchemy.SQLAlchemy() by using flask_sqlalchemy_lite.SQLAlchemy().

info

When using fsc.get_flask_sqlalchemy(...), it supposes that you are developing your codes with flask_sqlalchemy.SQLAlchemy(). Therefore, its returned value is always notated by flask_sqlalchemy.SQLAlchemy() even if the package flask_sqlalchemy is not installed. In that case, the notation of fsc.get_flask_sqlalchemy(...) will not work, but that does not matter because the environment without flask_sqlalchemy is not the developer's environment. In run time, the returned value of this function will fall back to the returned value of fsc.as_flask_sqlalchemy(...) automatically.

warning

In the compatible mode, the fall-back option fsc.SQLAlchemyProxy will not provide all functionalities of flask_sqlalchemy.SQLAlchemy(). Currently, the following features are not supported yet. If any of the following feature is urgent to you, please file an issue:

• Extensive methods of db.Model like db.Model.query.get_or_404(...).
• Paginate method db.paginate(...).

These unsupported features can be used by flask_sqlalchemy but will not work if the backend falls back to fsc.SQLAlchemyProxy driven by flask_sqlalchemy_lite.

Explicitly notate the extension by the compatible version

As mentioned above, the returned value of fsc.get_flask_sqlalchemy(...) is always notated by flask_sqlalchemy.SQLAlchemy() even if the backend package is not available. It may be not a good choice if users want to check the exact behavior of the compatible mode in the developer's environment. For example, users may want to know which API will cause issues in the compatible mode. In that case, a good choice is to temporarily replace the extension by _proxy_ver(...).

db = fsc.get_flask_sqlalchemy(Base, engine_options=engine_options)
db = fsc.get_flask_sqlalchemy_proxy_ver(Base, engine_options=engine_options)

This modification will not take any effect in run time. The only difference is that db will be always notated by fsc.SQLAlchemyProxy even if db is a flask_sqlalchemy.SQLAlchemy(). Supposes that users have used an unsupported functionality like db.paginate(...). Switching to this notation will let the static type checker aware of the unsupported method immediately.

tip

Another advantage of using fsc.get_flask_sqlalchemy_proxy_ver(...) is the better type notations. Flask SQLAlchemy still has several notation issues until the version 3.1.x. For example, the db.relationship is notated by a wrong type. If users choose the fsc.SQLAlchemyProxy notation, the notation will become correct.

Disable a backend

When users have installed both the flask_sqlalchemy and flask_sqlalchemy_lite, they may want to test the run-time behavior of fsc.get_flask_sqlalchemy(...) in the compatible mode even if the flask_sqlalchemy is available. In that case, users can turn off the support of flask_sqlalchemy globally by using

import flask_sqlalchemy_compat as fsc

# Make `flask_sqlalchemy` invisible to `fsc`.
fsc.backends.proxy.fsa = None

This configuration needs to be used before calling fsc.get_flask_sqlalchemy(...). It will explicitly make flask_sqlalchemy invisible to fsc, thus causing the method fsc.get_flask_sqlalchemy(...) forced to fall back to the compatible mode.

danger

You cannot use fsc.get_flask_sqlalchemy(...) if neither Flask SQLAlchemy nor Flask SQLAlchemy Lite is not installed. Therefore, if the Flask SQLAlchemy is the only installed package in your environment, making it invisible will cause fsc.get_flask_sqlalchemy(...) to raise a ModuleNotFoundError.