使用Flask SQLAlchemy

本指南展示了从flask-sqlalchemy迁移到本项目的步骤。

背景

假设你正在使用Flask SQLAlchemy撰写代码。但现在，你正计划着要迁移到Flask SQLAlchemy Lite。诚然，最佳做法还是将所有的代码都用Flask SQLAlchemy Lite重写一次。然而，Flask SQLAlchemy Lite并不支持某些旧的API、例如Model.query或db.query_or_404。在此情形下，可以考虑透过最小的修改，先完成一个初版的迁移代码，再在此基础上，慢慢修改、完成基于重写的迁移计划。那么，可以考虑使用此项目。

以下代码展示了一个完整的Flask SQLAlchemy写就的项目，其中使用了旧的API。

app.py
models.py
base.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():
        """主页，遍历所有文件夹数据项。"""
        return {
            "message": "找到了所有文件夹。",
            "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():
        """检索一个文件夹数据项，并返回其内含的所有文本项。"""
        folder_id = request.args.get("id", None, int)
        if folder_id is None:
            return flask.abort(404, "未指定文件夹ID。")
        folder = db.session.get(Folder, folder_id)
        if folder is None:
            return flask.abort(404, "未找到文件夹。")
        return {
            "message": "找到了文件夹内的文本项。",
            "folder": folder.id,
            "texts": [{"id": text.id, "title": text.title} for text in folder.texts],
        }

    @app.route("/text")
    def text():
        """检索一个文本数据项。"""
        text_id = request.args.get("id", None, int)
        if text_id is None:
            return flask.abort(404, "未指定文本ID。")
        text = db.session.get(Text, text_id)
        if text is None:
            return flask.abort(404, "未找到文本。")
        return {
            "message": "找到了文本。",
            "id": text.id,
            "title": text.title,
            "text": text.text,
        }

    @app.errorhandler(HTTPException)
    def handle_exception(exc: HTTPException):
        """针对HTTP错误，返回JSON而非HTML。"""
        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
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
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)

该应用会在内存里维护一个范例数据库，并提供用来检索范例数据的一些API。例如，当程序正在运行之时，

地址
响应

访问以下地址：

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

将检索Folder类型、id=1的数据项。所有属于该文件夹下的文字数据项亦都会展示。

浏览器返回的响应应当如下：

{
  "folder": 1,
  "message": "找到了文件夹内的文本项。",
  "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"
    }
  ]
}

下一步是检查文本项的详情。

地址
响应

访问以下地址：

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

将检索Text类型、id=1的数据项。该数据项已经列出在上一个响应的内容里。

浏览器返回的响应应当如下：

{
  "id": 1,
  "message": "找到了文本。",
  "text": "Lorem ipsum, ...",
  "title": "F1 Text 1"
}

扩展兼容性到`flask-sqlalchemy-lite`

迁移到本项目，将使得以上代码可以在后端包改为flask_sqlalchemy_lite后、仍然能正常工作。所要做的修改非常简单。唯一需要改动之处是base.py，参见以下代码：

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)

此处作出的主要改动，是令db由工厂函数fsc.get_flask_sqlalchemy给定。该函数的第一个输入值，是原始的基ORM模型。

若flask_sqlalchemy已经安装，则该函数不会产生任何实际效果。换言之，db将与调用flask_sqlalchemy.SQLAlchemy()构建的对象相同，且输入值Base将会在扩展初始化时、传递给model_class参数。

不过，若flask_sqlalchemy未安装，则该函数会试图寻找flask_sqlalchemy_lite，并使用flask_sqlalchemy模拟flask_sqlalchemy的行为。在此情况下，db将实际由flask_sqlalchemy_lite.SQLAlchemy()驱动，且所得的属性db.Model会是一个修改过的Base类。该修改是原处的，因此使用db.Model和Base是等价的。

备注

注意，兼容模式只会在flask_sqlalchemy未安装时生效。

显式使用`flask-sqlalchemy-lite`模拟`flask-sqlalchemy`

上一节展示了fsc.get_flask_sqlalchemy_lite(...)的用法。在Lite版未安装时，该方法会自动切换到Flask SQLAlchemy。然而，在某些情况下，用户可能会需要刻意地使用Flask SQLAlchemy来模拟Flask SQLAlchemy Lite的行为。在此情形下，上述的脚本可以修改如下：

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
)

在此情形下，数据库扩展db将会注解为fsc.SQLAlchemyProxy。这是用来提供flask_sqlalchemy.SQLAlchemy()的API的、一个针对flask_sqlalchemy_lite.SQLAlchemy()的封装。

信息

使用fsc.get_flask_sqlalchemy(...)隐含了“用户正在使用flask_sqlalchemy.SQLAlchemy()开发代码”的背景信息。因此，即使flask_sqlalchemy并未安装，该函数的返回值也总是注解为flask_sqlalchemy.SQLAlchemy()。未安装flask_sqlalchemy时，fsc.get_flask_sqlalchemy(...)的注解当然不会工作，但那并不产生什么实际影响，因为未安装flask_sqlalchemy的环境、并非是用来开发代码的环境。在运行时，该函数的返回值能自动退到fsc.as_flask_sqlalchemy(...)的返回值。

注意

兼容模式下，回退的封装fsc.SQLAlchemyProxy不能提供flask_sqlalchemy.SQLAlchemy()的全部特性。目前，以下的功能还未得到支持。若其中某些功能是你目前急需的，请报告问题。

db.Model的扩展方法，例如db.Model.query.get_or_404(...)。
分页方法db.paginate(...)。

这些未支持的功能可以在flask_sqlalchemy中使用，但若是退化到flask_sqlalchemy_lite驱动的fsc.SQLAlchemyProxy，则不会工作。

显式地将扩展标记成兼容版本

如上文所述，即使后端的包不可用，fsc.get_flask_sqlalchemy(...)也总是注解为flask_sqlalchemy.SQLAlchemy()。如果用户想要在开发者环境里、了解兼容模式下确切的行为，这就有可能不是一个较好的选择。例如，用户可能需要知道，哪些API在兼容模式下可能会引发异常。在此情形下，推荐将用来获取扩展的函数暂时修改为_proxy_ver(...)。

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

该修改在运行时下不会产生任何效果。改动唯一的效果是、即使db已经是flask_sqlalchemy.SQLAlchemy()对象，其也会总是注解为fsc.SQLAlchemyProxy。假设用户使用了不支持的功能，例如db.paginate(...)，切换到该注解将有助于静态类型检查器立刻发现不支持的方法。

提示

另一个使用fsc.get_flask_sqlalchemy_proxy_ver(...)的优势，是更好的类型注解。直到3.1.x版，Flask SQLAlchemy仍然有不少类型注解错误。例如，db.relationship就被注解为了错误的类型。若用户选择使用fsc.SQLAlchemyProxy作为注解，则这些注解错误都会被修正。

禁用一种后端

设若用户同时安装了flask_sqlalchemy和flask_sqlalchemy_lite，那就有可能需要在flask_sqlalchemy存在的情况下，测试fsc.get_flask_sqlalchemy(...)的兼容模式行为。在此情形下，用户可以按照下例、全局地关闭flask_sqlalchemy的支持：

import flask_sqlalchemy_compat as fsc

# 令`flask_sqlalchemy`在`fsc`中不可见。
fsc.backends.proxy.fsa = None

该设置需要在调用fsc.get_flask_sqlalchemy(...)之前完成。这会使得flask_sqlalchemy在fsc中不可见，从而使得fsc.get_flask_sqlalchemy(...)强行退到兼容模式下。

危险

倘若Flask SQLAlchemy和Flask SQLAlchemy Lite都没有安装，则不可以使用fsc.get_flask_sqlalchemy(...)。因此，假使环境中只安装了Flask SQLAlchemy，将其标记为不可见、将会导致fsc.get_flask_sqlalchemy(...)抛出ModuleNotFoundError。

背景​

扩展兼容性到flask-sqlalchemy-lite​

显式使用flask-sqlalchemy-lite模拟flask-sqlalchemy​

显式地将扩展标记成兼容版本​

禁用一种后端​

背景

扩展兼容性到`flask-sqlalchemy-lite`

显式使用`flask-sqlalchemy-lite`模拟`flask-sqlalchemy`

显式地将扩展标记成兼容版本

禁用一种后端