使用Flask SQLAlchemy Lite

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

背景

假设你正在使用Flask SQLAlchemy Lite撰写代码。然而，现在、在一些旧的设备上，你必须支持Python 3.7或Python 3.8。以下代码可以作为一个完整的Flask项目的范例：

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, Base, 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:
    Base.metadata.drop_all(db.engine)
    Base.metadata.create_all(db.engine)

    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_ENGINES": {"default": "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 Base, db


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


class Text(Base):
    __tablename__ = "text"
    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"]] = sa_orm.relationship(default=None)


class Folder(Base):
    __tablename__ = "folder"
    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]] = sa_orm.relationship(
        default_factory=list, back_populates="folder"
    )

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

from flask_sqlalchemy_lite 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(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"
}

扩展兼容性到`Python<3.9`

迁移到本项目，将使得以上代码可以在Python 3.7和3.8使用。所要做的修改非常简单。唯一需要改动之处是base.py，参见以下代码：

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

from flask_sqlalchemy_lite import SQLAlchemy
import flask_sqlalchemy_compat as fsc


__all__ = ("Base", "db")


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


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

db = SQLAlchemy(engine_options=engine_options)
db, Base = fsc.get_flask_sqlalchemy_lite(_Base, engine_options=engine_options)

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

若flask_sqlalchemy_lite已经安装，则该函数不会产生任何实际效果。换言之，db将与调用flask_sqlalchemy_lite.SQLAlchemy()构建的对象相同，且返回值Base和输入_Base是同一对象。

不过，若flask_sqlalchemy_lite未安装，则该函数会试图寻找flask_sqlalchemy，并使用flask_sqlalchemy模拟flask_sqlalchemy_lite的行为。在此情况下，db将实际由flask_sqlalchemy.SQLAlchemy()驱动，且Base将与_Base不同，因为该值实际上由SQLAlchemy().Model提供。

备注

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

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

上一节展示了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_lite import SQLAlchemy
from flask_sqlalchemy import SQLAlchemy
import flask_sqlalchemy_compat as fsc

__all__ = ("Base", "db")


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


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

db = SQLAlchemy(engine_options=engine_options)
db = fsc.as_flask_sqlalchemy_lite(
    db=SQLAlchemy(model_class=_Base, engine_options=engine_options),
)
Base = db.Model

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

信息

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

注意

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

异步engine和异步session。

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

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

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

db, Base = fsc.get_flask_sqlalchemy_lite(Base, engine_options=engine_options)
db, Base = fsc.get_flask_sqlalchemy_lite_proxy_ver(Base, engine_options=engine_options)

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

禁用一种后端

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

import flask_sqlalchemy_compat as fsc

# 令`flask_sqlalchemy_lite`在`fsc`中不可见。
fsc.backends.proxy.fsa_lite = None

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

危险

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

背景​

扩展兼容性到Python<3.9​

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

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

禁用一种后端​

背景

扩展兼容性到`Python<3.9`

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

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

禁用一种后端