Flask 中 SQLAlchemy 的四种集成方式：从 Declarative 到 SQL 抽象层与 teardown_appcontext 源码解析

本文基于 Flask 官方文档 docs/patterns/sqlalchemy.rst 展开，系统介绍在 Flask 应用中接入 SQLAlchemy 的四种主流方式——Flask-SQLAlchemy 扩展、Declarative 声明式映射、手动 ORM 映射与纯 SQL 抽象层。读完后你将掌握每种方式的完整 database.py / models.py 代码骨架、init_db() 建库流程与查询插入用法，并能从 teardown_appcontext 注册机制 与 应用上下文弹出流程 的源码层面，理解"为什么数据库会话能在请求结束时被自动清理"这一核心机制。

一、前提：为什么建议把应用组织成 package

原文档开篇即指出：如果你选择 SQLAlchemy 作为数据库访问层，建议将 Flask 应用从单个模块升级为 Python package，并把模型（models）放到独立模块中。这样做虽然"并非必须"，但能显著提升代码组织性。package 的具体搭建步骤（__init__.py、views.py、pyproject.toml 与 flask --app 运行方式）见 Large Applications as Packages。

后文所有示例代码都假设应用结构如下：

/yourapplication
    pyproject.toml
    /yourapplication
        __init__.py
        database.py
        models.py
        /static
        /templates

另一个贯穿全文的关键前提是：Flask 不会自动管理 SQLAlchemy 的生命周期，我们需要用 @app.teardown_appcontext 在每次应用上下文结束时移除会话。这一点在 SQLite3 模式文档 中同样成立，官方教程示例 flaskr 的 db.py 就通过 app.teardown_appcontext(close_db)（见 第 55 行）注册清理函数，与本文的 shutdown_session 属于同一模式。

二、方式一：Flask-SQLAlchemy 扩展（快速起步首选）

原文档给出的第一条建议是：由于 SQLAlchemy 是常见的数据库抽象层和对象关系映射器，且需要一定的配置工作，官方提供了 Flask 扩展来替你处理这些配置。如果你希望快速起步，这是官方推荐的方式。

安装方式：

pip install Flask-SQLAlchemy

相比后三种"裸用 SQLAlchemy"的方式，Flask-SQLAlchemy 会替你完成引擎创建、会话绑定、db.Model 基类、与 Flask 配置项（SQLALCHEMY_DATABASE_URI）的对接，以及与应用上下文的生命周期集成。它的具体用法（如 应用工厂中初始化 db 的方式）在扩展自己的文档中有详细说明，本文档的定位是给出选择建议，并以下面三种方式帮助你在不使用扩展时也能直接上手。

三、方式二：Declarative 声明式映射

SQLAlchemy 的 Declarative 扩展是使用 SQLAlchemy 的最新方式：像 Django 一样，把表和模型一次定义完成。原文档推荐配合 SQLAlchemy 官方 declarative 文档一起阅读。

3.1 database.py：引擎、会话与 Base

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker, declarative_base

engine = create_engine('sqlite:////tmp/test.db')
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))
Base = declarative_base()
Base.query = db_session.query_property()

def init_db():
    # import all modules here that might define models so that
    # they will be registered properly on the metadata.  Otherwise
    # you will have to import them first before calling init_db()
    import yourapplication.models
    Base.metadata.create_all(bind=engine)

几个值得注意的实现细节：

• create_engine('sqlite:////tmp/test.db')：四个斜杠是 SQLAlchemy 连接串语法，表示绝对路径 /tmp/test.db；
• scoped_session(sessionmaker(...))：这里 autocommit=False、autoflush=False 是显式关闭自动提交与自动刷新，把事务控制交给业务代码；
• Base.query = db_session.query_property()：给所有模型类挂上 query 属性，从而支持 User.query.filter(...) 这类简洁写法；
• 文档特别解释了线程问题：在 SQLite3 示例 中需要用 flask.g 手动管理"每个请求一条连接"，而这里不必操心——scoped_session 已经按线程/上下文作用域做了会话隔离。

3.2 应用模块中注册会话清理

把以下代码放进应用模块，Flask 会在请求结束或应用关闭时自动移除数据库会话：

from yourapplication.database import db_session

@app.teardown_appcontext
def shutdown_session(exception=None):
    db_session.remove()

这一行注册代码在 Flask 侧的行为可以从源码确认：

1. teardown_appcontext 装饰器在 src/flask/sansio/app.py 第 827-855 行 实现，它把回调函数追加到 teardown_appcontext_funcs 列表（第 360 行）。文档注释明确说明：上下文在"请求结束、CLI 命令结束或手动 with 块退出"时弹出，teardown 函数在应用上下文被标记为 inactive 之前调用；
2. 实际的调用发生在 AppContext.pop()：先执行请求级 teardown，再调用 self.app.do_teardown_appcontext(self, exc)；
3. do_teardown_appcontext 定义在 src/flask/app.py 第 1453-1479 行，它会以注册的反序执行所有 teardown 回调，然后发出 appcontext_tearing_down 信号。从源码结构看，即使某个回调抛异常，也会收集错误后继续执行其余回调（Flask 3.2 起的行为），最后统一抛出。

这也解释了为什么 teardown 回调签名要带 exception=None 参数——当上下文因未捕获异常弹出时，该异常对象会作为 exc 传入。

3.3 models.py：定义模型

from sqlalchemy import Column, Integer, String
from yourapplication.database import Base

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)
    email = Column(String(120), unique=True)

    def __init__(self, name=None, email=None):
        self.name = name
        self.email = email

    def __repr__(self):
        return f'<User {self.name!r}>'

3.4 建库、插入与查询

用 init_db() 创建数据库表（注意函数内部会先导入 models 模块，确保模型注册到 metadata 上）：

>>> from yourapplication.database import init_db
>>> init_db()

插入数据：

>>> from yourapplication.database import db_session
>>> from yourapplication.models import User
>>> u = User('admin', 'admin@localhost')
>>> db_session.add(u)
>>> db_session.commit()

查询则得益于 3.1 节挂在 Base 上的 query 属性：

>>> User.query.all()
[<User 'admin'>]
>>> User.query.filter(User.name == 'admin').first()
<User 'admin'>

四、方式三：Manual Object Relational Mapping（手动 ORM）

手动 ORM 与 Declarative 的对比各有优劣：表和类分开定义，再映射到一起。优点是更灵活，缺点是代码量稍多。整体流程和 Declarative 方式一致，因此同样建议把应用拆分为 package 中的多个模块。

4.1 database.py

from sqlalchemy import create_engine, MetaData
from sqlalchemy.orm import scoped_session, sessionmaker

engine = create_engine('sqlite:////tmp/test.db')
metadata = MetaData()
db_session = scoped_session(sessionmaker(autocommit=False,
                                         autoflush=False,
                                         bind=engine))

def init_db():
    metadata.create_all(bind=engine)

与 Declarative 的区别：这里没有 Base，取而代之的是显式创建的 MetaData 对象，init_db() 直接对 metadata 调 create_all。

会话清理代码与 3.2 节完全相同（@app.teardown_appcontext + db_session.remove()），因为两者都使用 scoped_session。

4.2 models.py：表与类分开定义

from sqlalchemy import Table, Column, Integer, String
from sqlalchemy.orm import mapper
from yourapplication.database import metadata, db_session

class User(object):
    query = db_session.query_property()

    def __init__(self, name=None, email=None):
        self.name = name
        self.email = email

    def __repr__(self):
        return f'<User {self.name!r}>'

users = Table('users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(50), unique=True),
    Column('email', String(120), unique=True)
)
mapper(User, users)

关键点：

• users = Table(...)：直接声明表结构并挂到 metadata 上；
• mapper(User, users)：把普通类与表映射为 ORM 类；
• query = db_session.query_property()：因为不再继承 Base，需要在类上手动挂 query 属性，从而保持与 Declarative 方式一致的 User.query 查询体验。

原文档指出：查询与插入的用法与 Declarative 示例完全相同（db_session.add / db_session.commit / User.query.filter(...)）。

五、方式四：SQL Abstraction Layer（纯 SQL 抽象层）

如果你只想要数据库系统（和 SQL）的抽象层，而不需要 ORM，那么基本上只需要一个 engine：

from sqlalchemy import create_engine, MetaData, Table

engine = create_engine('sqlite:////tmp/test.db')
metadata = MetaData(bind=engine)

建表有两种途径：像上面示例那样在代码中声明 Table，或者从已有数据库自动加载：

from sqlalchemy import Table

users = Table('users', metadata, autoload=True)

插入数据时，需要先拿到连接以便使用事务：

>>> con = engine.connect()
>>> con.execute(users.insert(), name='admin', email='admin@localhost')

SQLAlchemy 会替我们自动提交。

查询可以直接用 engine 或连接：

>>> users.select(users.c.id == 1).execute().first()
(1, 'admin', 'admin@localhost')

结果行是"字典风格的元组"（dict-like tuples），既可按位置也可按列名取值：

>>> r = users.select(users.c.id == 1).execute().first()
>>> r['name']
'admin'

此外还可以向 execute 方法直接传 SQL 字符串（使用命名/位置绑定参数，避免字符串拼接带来的 SQL 注入风险，这一点与 SQLite3 文档中"绝不要用字符串格式化拼接 SQL"的告诫一致）：

>>> engine.execute('select * from users where id = :1', [1]).first()
(1, 'admin', 'admin@localhost')

六、四种方式的选择对照与小结

方式	适用场景	核心组件	会话生命周期管理
Flask-SQLAlchemy 扩展	快速起步、常规 CRUD 应用	扩展提供的 db 对象	由扩展与 Flask 上下文自动集成
Declarative	想要 ORM 且希望代码最简洁	declarative_base + scoped_session	@app.teardown_appcontext + db_session.remove()
Manual ORM	表与类需要分开、映射更灵活	MetaData + Table + mapper	同 Declarative
SQL 抽象层	不需要 ORM，只要 SQL 抽象	create_engine + MetaData	无需会话，连接随用随开

需要提醒的适用前提：本文示例沿用官方文档中的 SQLAlchemy 1.x 时代写法（如 MetaData(bind=engine)、autoload=True、con.execute(insert, **kwargs) 等 API），这些 API 在新版 SQLAlchemy 中已有演进（例如 autoload 被 autoload_with 取代、Core 插入需先 connection.begin() 等）。在按本文搭建新项目前，建议对照当前安装的 SQLAlchemy 版本 API 做相应调整，但本文讲解的架构分层（database.py / models.py 分离、init_db() 建库、teardown 清理会话）与 Flask 侧的 teardown_appcontext 机制（注册 → 执行 → 触发点）不随 SQLAlchemy 版本变化，是长期有效的参考。

更多 SQLAlchemy 的细节请查阅其官方网站文档；Flask 侧其他数据库访问模式（如 SQLite3 直连）以及 应用工厂模式 也可与本文配合使用。