SQLAlchemy 2 实战：第八章——SQLAlchemy 与 Web 应用SQLAlchemy 2 In Practice - Chapter 8: SQLAlchemy and the Web

这是《SQLAlchemy 2 in Practice》一书的第八章，也是最后一章。如果你想支持我的工作，我建议你购买这本书，可以直接在我的商店购买，也可以在亚马逊上购买。谢谢！

无论你是构建传统 Web 应用，还是与前端网页或智能手机应用协同工作的 Web API，SQLAlchemy 都是为 Python Web 服务器添加数据库支持的绝佳选择之一。本章将演示与 Flask 和 FastAPI 的两个示例集成。它们是两个最受欢迎的 Python Web 框架，即使你使用其他框架，这些示例也值得参考。

供你参考，以下是本书内容的摘要：

通用集成方法

如果你正在寻找将 SQLAlchemy 集成到 Web 应用中最简单的方法，那么可以考虑完全不使用任何集成方式。

按照本书介绍的结构，你可以将 db.py 和 models.py 模块添加到应用中，然后使用 session 上下文管理器，在需要数据库功能的所有位置引入它。事实上，这正是前几章中所有导入脚本所使用的技术。

这种方法不仅适用于 Web 应用，也适用于其他类型的 Python 应用，其优势在于无需额外的依赖或扩展。

SQLAlchemy 集成技术

虽然上一节建议的无集成方案在许多项目中都能很好地工作，但你可能更倾向于使用一种能封装并简化数据库功能、便于你选择的 Web 框架使用的解决方案。以下章节将讨论一些应考虑的实现细节。

SQLAlchemy 导入的消歧义

SQLAlchemy 有时让人感觉繁琐的一个方面是它导出了许多类和函数。考虑前一章的 models.py 模块中使用的 SQLAlchemy 导入：

from sqlalchemy import String, Text, ForeignKey, Table, Column
from sqlalchemy.orm import Mapped, WriteOnlyMapped, mapped_column, relationship

如果你需要运行数据库查询，就必须导入 Session、select() 函数以及根据需求导入的其他几个符号。

在应用中拥有这些冗长的导入列表有两个缺点。首先，在每个需要使用数据库的模块顶部维护这些导入列表非常耗时；更重要的是，像 select 这样名称较为通用的导入可能会与其他依赖项或应用本身的符号发生冲突。

一种解决这两个问题的方法是通过仅导入父模块来命名空间所有导入。对于 SQLAlchemy ORM 应用，通常有两个父模块：sqlalchemy 和 sqlalchemy.orm，因此可以直接导入它们：

import sqlalchemy
import sqlalchemy.orm

这样做后，所有符号都可以通过其父模块访问，例如 sqlalchemy.select 或 sqlalchemy.orm.relationship。为了避免名称过长，可以将导入重命名为较短的别名：

import sqlalchemy as sa
import sqlalchemy.orm as so

现在，符号分别以 sa. 和 so. 作为前缀，分别对应 SQLAlchemy Core 和 ORM。以下是使用此风格编写的 Order 模型示例：

class Product(Model):
    __tablename__ = 'products'

    id: so.Mapped[int] = so.mapped_column(primary_key=True)
    name: so.Mapped[str] = so.mapped_column(
        sa.String(64), index=True, unique=True)
    manufacturer_id: so.Mapped[int] = so.mapped_column(
        sa.ForeignKey('manufacturers.id'), index=True)
    year: so.Mapped[int] = so.mapped_column(index=True)
    cpu: so.Mapped[Optional[str]] = so.mapped_column(sa.String(32))

    manufacturer: so.Mapped['Manufacturer'] = so.relationship(
        back_populates='products')
    countries: so.Mapped[list['Country']] = so.relationship(
        secondary=ProductCountry, back_populates='products')
    order_items: so.WriteOnlyMapped['OrderItem'] = so.relationship(
        back_populates='product')
    product_reviews: so.WriteOnlyMapped['ProductReview'] = so.relationship(
        back_populates='product')
    blog_articles: so.WriteOnlyMapped['BlogArticle'] = so.relationship(
        back_populates='product')

    def __repr__(self):
        return f'Product({self.id}, "{self.name}")'

模型序列化

对于 Web 应用程序和基于 Web 的 API 而言，一个特殊需求是将模型发送给请求它们的客户端。为了通过网络传输这些实体，必须对其进行序列化——即将 Python 模型实例从其内部二进制表示转换为可传输的字符串或字节序列，并在另一端重建。

最常用的序列化格式是 JavaScript 对象表示法（JavaScript Object Notation），简称 JSON（发音为“Jason”）。下面是一个来自 RetroFun 数据库的 Product 实体在序列化为 JSON 格式后的示例：

{
  "id": 41,
  "name": "Commodore 64",
  "manufacturer": {
    "id": 14,
    "name": "Commodore"
  },
  "countries": [
    {
      "id": 3,
      "name": "USA"
    }
  ],
  "year": 1982,
  "cpu": "6510"
}

JSON 格式的语法与 Python 中的字典、列表以及整数和字符串等原始类型相似。事实上，Python 标准库中的 json 模块可以将具有上述结构的 Python 字典渲染为对应的 JSON 序列化表示形式，并以字符串形式返回，该字符串可用于响应客户端的请求。

序列化的逆过程称为反序列化。对于 JSON 而言，这一过程由 JSON 解码器完成。所有语言和开发栈都提供了 JSON 解码器，因此当客户端从服务器接收到包含上述内容的响应字符串时，可以将其解码为便于使用的格式。例如，在浏览器中运行的 JavaScript 可以通过 JSON.parse() 函数将 JSON 数据转换为基于 JavaScript 对象、数组和原始类型的结构。

仔细查看上述示例的结构会发现，这种特定表示不仅包含一个产品，还嵌入了其关联关系，即制造商和原产地国家，它们各自也有嵌入父实体中的 JSON 表示。

构建一个能够递归地包含关联关系并轻松生成响应的序列化系统其实并不难。基本思路是为每个模型类添加一个 to_dict() 方法，该方法返回一个可被序列化为 JSON 格式的字典版本对象。

以下是一个 Product 模型的 to_dict() 方法实现示例：

class Product(Model):
    # ...

    def to_dict(self):
        return {
            'id': self.id,
            'name': self.name,
            'manufacturer': self.manufacturer.to_dict(),
            'year': self.year,
            'cpu': self.cpu,
            'countries': [country.to_dict() for country in self.countries],
        }

此示例展示了如何通过调用相关对象的 to_dict() 方法来将相关对象嵌入到父对象的表示中，从而确保序列化实体的逻辑集中在一个地方管理。

备注： 对于大型且复杂的模型，使用像 Marshmallow 这样的专用序列化库可能是 to_dict() 方法的替代方案，效果更佳。

Alchemical 包

在使用 SQLAlchemy 的项目中，通常需要创建以下几个对象：

在之前章节提供的代码示例中，所有这些对象都在 db.py 模块中初始化，任何集成了 SQLAlchemy 的项目都需要包含此模块或类似功能的模块。

Alchemical 包（由本书作者创建和维护）旨在通过将所有上述内容封装到一个 Alchemical 实例中来简化 SQLAlchemy 的使用。你可以使用 pip 安装 Alchemical：

(venv) $ pip install alchemical

考虑以下示例，它实现了与 db.py 模块相同的功能：

import os
from dotenv import load_dotenv
from alchemical import Alchemical

load_dotenv()

db = Alchemical(os.environ['DATABASE_URL'])

从 Alchemical 类创建的 db 对象包含了上面列出的所有 SQLAlchemy 项目。以下是访问它们的方式：

Alchemical 类提供了额外功能：

示例 Web 应用

本书的 GitHub 仓库包含一个完整的小型 Web 应用示例，展示 RetroFun 订单表格，并支持高效的分页、排序、搜索和信息提示功能。Flask 和 FastAPI 的实现分别作为传统和异步集成的示例。

表格前端

示例应用中使用的表格基于 grid.js 库构建。该表格配置为“服务器端”模式，即通过向服务器发起请求来获取数据。这是最高效的配置方式，因为仅请求需要显示的数据。每当用户点击分页链接、列排序标题或在搜索框中输入内容时，都会向服务器发送新请求以刷新表格信息。

本书不深入讨论该表格的前端实现细节，但可在两个示例应用的 index.html 文件中查看客户端源代码。

需要了解的是，当表格需要显示新数据时，会向服务器的 /api/orders 端点发起请求，并附带以下查询字符串参数：

以下是前端可能发出的请求 URL 示例：获取第三页结果（每页 10 条），按订单总额升序排序，搜索字符串为 Dylan。

http://domain.com/api/orders?start=21&length=10&sort=%2Btotal&search=Dylan

注意 sort 参数的值是 %2Btotal。URL 中的某些字符必须进行转义，+ 就是其中之一。编码方式是使用字符的十六进制 ASCII 码，并在前面加上 % 前缀。像 Flask 或 FastAPI 这样的 Web 框架会自动处理字符转义，因此开发人员通常无需关心这一任务。

/api.orders 端点在后台的作用是接收这四个查询字符串参数，根据它们执行数据库查询，并以以下 JSON 结构返回请求的项目：

{
  "data": [
    { ... order ... },
    { ... order ... },
    ...
  ],
  "total": <n>
}

响应的数据部分必须包含一个订单数组，每个订单都按照其 to_dict() 序列化方法格式化。响应中最多包含 length 个订单。total 字段应包含满足当前搜索条件的条目总数，或者在没有定义搜索时显示订单总数。这样表格可以显示类似“显示第 31 至 40 项，共 4798 条结果”的图例。

数据库查询

后端最重要的部分之一是生成解决客户端请求的查询逻辑。这里用的是复数形式，因为所需的 JSON 负载需要一条查询用于响应的数据部分，另一条用于 total。

实际上，total 的查询更简单。该查询需要计算与搜索字符串匹配的订单数量，或者在没有搜索字符串时计算订单总数。此查询仅使用搜索查询字符串参数。start、length 和 sort 参数对此计算没有影响。

下面显示的 total_orders() 函数将作为示例应用程序中 queries.py 模块的一部分，创建此查询。

def total_orders(search):
    if not search:
        return sa.select(sa.func.count(Order.id))

    return (
        sa.select(sa.func.count(sa.distinct(Order.id)))
            .join(Order.customer)
            .join(Order.order_items)
            .join(OrderItem.product)
            .where(
                sa.or_(
                    Customer.name.ilike(f'%{search}%'),
                    Product.name.ilike(f'%{search}%'),
                )
            )
    )

根据是否存在搜索字符串，该查询有两种不同的实现方式。当未提供搜索字符串时，使用一个简单的查询来返回订单总数。

当存在搜索字符串时，查询会更复杂。select() 部分仍指定一个计数，但这次必须统计唯一订单，因为与 customers 和 products 的连接可能会产生重复结果，正如你在许多示例查询中看到的那样。

将 Order 与 Customer 连接使客户名称在查询中可搜索。为了也能搜索产品名称，Order 与 OrderItem 连接，而 OrderItem 又与 Product 连接。请记住，OrderItem 是 orders 和 products 之间多对多关系的连接表。

在所有连接完成后，搜索通过 where() 子句执行，该子句包含两个条件，并使用“或”逻辑运算符组合。ilike() 函数用于对 Customer.name 和 Product.name 列执行不区分大小写的模式搜索。

返回一页订单的查询实现在 paginated_orders() 函数中，如下所示。

def paginated_orders(start, length, sort, search):
    # base query to retrieve orders with their total amount
    total = sa.func.sum(OrderItem.quantity * OrderItem.unit_price).label(None)
    q = (
        sa.select(Order, total, Customer)
            .options(so.selectinload(Order.customer))  # only needed for async
            .join(Order.customer)
            .join(Order.order_items)
            .join(OrderItem.product)
            .group_by(Order, Customer)
            .distinct()
    )

    # add search filters
    if search:
        q = q.where(
            sa.or_(
                Customer.name.ilike(f'%{search}%'),
                Product.name.ilike(f'%{search}%'),
            )
        )

    # add sorting
    if sort:
        order = []
        for s in sort.split(','):
            direction = s[0]  # first character is either + or -
            name = s[1:]  # rest of the string is the column name
            if name == 'customer':
                column = Customer.name
            elif name == 'total':
                column = total
            else:
                column = getattr(Order, name)
            if direction == '-':
                column = column.desc()
            order.append(column)
        if not order:
            order = [Order.timestamp.desc()]
        q = q.order_by(*order)

    # add pagination
    q = q.offset(start).limit(length)

    return q

这个查询的构建工作量更大。该函数中的一个有趣实现选择是将查询分为四个独立的部分，然后依次拼接，而不是作为 select() 函数调用后的一条连续子句链。

基础查询获取订单及其总额，其方式与前面章节中的示例非常相似。Order 与 OrderItem 的连接是必要的，以便计算总额；而与 Customer 和 Product 的连接则是预防性的，以支持搜索和排序功能。由于应用程序允许用户按客户名称对数据进行排序，因此 select() 和 group_by() 子句中包含了 Customer。当使用异步查询时，.options() 子句是必需的，因为 Order.customer 关系在 models.py 中配置为 lazy='joined'，而对于使用分组连接的情况，SQLAlchemy 会隐式添加这些连接，若数据库无法确定如何对连接检索的数据进行分组，则可能导致错误。

查询的第二部分是条件性的，仅在存在搜索字符串时生效。这部分会附加一个 where() 子句，该子句与用于计算订单总数的查询所使用的子句完全相同。

第三部分也是条件性的，仅在存在排序请求时使用。排序字符串以逗号分隔的形式传入，因此此部分将 sort 的值拆分为各个部分，并分别获取要排序的列。支持的列包括 Customer.name、总额标签或 Order 模型的主键列（本例中仅使用了 timestamp）。如果列名以 - 开头，则在排序属性上调用 desc() 方法以实现降序排列。解析排序字符串时收集到的列列表随后被包含在一个 order_by() 子句中，并附加到查询上。

第四部分也是最后一部分，添加了分页的 offset() 和 limit() 子句，以确保检索到正确范围的结果。

需要注意的是，这两个函数仅创建查询。将查询的创建与执行分离，使得这些查询可以以一种完全通用的方式编写，无需修改即可适用于 Flask 和 FastAPI 示例。

端点

该应用需要两个端点。根 URL 将返回一个包含前端 JavaScript 代码的 HTML 页面。当前端需要根据用户操作（如点击分页链接）更新显示的项目时，它将向 /api/orders 端点发送请求。

端点的编码必须遵循所选 Web 框架设定的约定。Flask 和 FastAPI 都将端点定义为带有路由装饰器的函数。两种实现的语法并不完全相同，但非常相似。

根 URL 的处理程序不需要访问数据库，因为它只需返回包含前端代码的 HTML 文件。而 /api/orders 端点的处理程序则定义了驱动表格内容的核心逻辑。此端点必须完成以下任务：

Flask 路由

存储在 routes.py 模块中的两个端点的 Flask 版本如下所示。

from flask import Blueprint, render_template, request
from .models import db
from . import queries

bp = Blueprint('routes', __name__)


@bp.route('/')
def index():
    return render_template('index.html')


@bp.route('/api/orders')
def get_orders():
    start = request.args.get('start')
    length = request.args.get('length')
    sort = request.args.get('sort')
    search = request.args.get('search')

    data_query = queries.paginated_orders(start, length, sort, search)
    total_query = queries.total_orders(search)

    orders = db.session.execute(data_query)
    data = [{**o[0].to_dict(), 'total': o[1]} for o in orders]
    return {
        'data': data,
        'total': db.session.scalar(total_query),
    }

在 Flask 中，通常通过蓝图（blueprints）来定义应用程序的路由。每个蓝图随后会注册到应用实例上，以便将其路由包含进来。在此应用中，bp 是唯一一个蓝图，上述两个端点分别在 index() 和 get_orders() 函数中实现。

index() 函数使用 Flask 的 render_template() 函数返回 HTML 页面。该页面的 JavaScript 逻辑将开始向 /api/orders 发送请求，以填充订单表格。

get_orders() 函数负责生成表格内容。首先从查询字符串中提取四个参数，Flask 通过 request.args 字典暴露这些参数。data_query 和 total_query 数据库查询是通过调用前面描述过的函数生成的。

借助 Alchemical 提供的 Flask 集成，db.session 是一个相当“神奇”的属性，它会在首次使用时自动启动会话。这是 Flask 及其许多扩展中常见的一种模式，因此 Alchemical 也采用这种方式集成到 Flask 中。因此无需使用 Session 上下文管理器来启动会话。Alchemical 会在请求结束时关闭 db.session 对象。

get_orders() 函数的其余部分分别通过 db.session.execute() 和 db.session.scalar() 执行这两个查询，并返回一个符合客户端要求的字典。Flask 会自动将返回的字典渲染为 JSON 格式的响应。

以下片段需要仔细研究才能完全理解：

    orders = db.session.execute(data_query)
    data = [{**o[0].to_dict(), 'total': o[1]} for o in orders]

data_query 的执行结果存储在 orders 变量中。这是一个 SQLAlchemy 结果对象，具有可迭代性。在第二行中，列表推导式遍历结果并为 JSON 响应创建 data 部分。每个结果列表中的元素必须是序列化的 Order 模型，可通过表达式 o[0].to_dict() 获得。但这还不够，因为客户端期望包含一个 total 属性，而该属性不属于 Order 模型，也需要一并包含在订单数据中。total 作为每行结果的第二个值返回，因此每个订单返回的字典由 Order.to_dict() 方法的所有数据加上 total 结果组装而成。

FastAPI 路由

对于 FastAPI，端点存储在应用程序的 router.py 模块中，接下来你可以看到相关内容。

from fastapi import APIRouter
from fastapi.responses import FileResponse
from .models import db
from . import queries

router = APIRouter()


@router.get('/')
async def index():
    return FileResponse('retrofun/html/index.html')


@router.get('/api/orders')
async def get_orders(start: int, length: int, sort: str = '',
                     search: str = ''):
    data_query = queries.paginated_orders(start, length, sort, search)
    total_query = queries.total_orders(search)

    async with db.Session() as session:
        orders = await session.stream(data_query)
        data = [{**o[0].to_dict(), 'total': o[1]} async for o in orders]
        return {
            'data': data,
            'total': await session.scalar(total_query),
        }

FastAPI 中相当于 Flask 蓝图的概念称为 API 路由器（API router），除了名称不同外，两者功能相同。由于 FastAPI 是异步框架，处理函数定义为 async def 函数。

index() 函数被注册为应用程序根 URL 的处理程序，仅返回包含前端代码的 HTML 文件。

get_orders() 函数处理 /api/orders 端点。在 FastAPI 中，查询字符串参数作为类型化参数定义在函数中，因此四个预期的查询参数都包含在函数声明中。start 和 length 参数始终由客户端提供，因此无需为其设置默认值；但对于 sort 和 search，则使用空字符串作为默认值，以防客户端未发送这些参数。

查询通过调用前面讨论的两个函数生成，与 Flask 版本中的实现方式完全相同。

执行查询时，会通过上下文管理器启动一个会话。在使用 Alchemical 包时，会话的基类是 db.Session，其中 db 为 Alchemical 实例。

此版本中，由于数据库运行在异步模式下，因此查询使用 await 发出。会话的 stream() 方法被用于替代 execute()，以便结果以异步迭代器的形式返回。订单列表通过异步列表推导式转换为 JSON 格式的订单列表，这与 Flask 应用中的处理方式相同。

Flask 后端

所有有趣的实现细节已在前面讨论过，剩余部分仅为连接各组件的样板代码和胶水代码。该应用的完整代码可在 GitHub 仓库中找到。

项目具有以下结构：

- main.py           # The entry point of the application
- config.py         # Flask configuration variables
- retrofun          # Python package with the application logic
  - __init__.py     # Package initialization (load environment variables)
  - app.py          # The application factory function
  - models.py       # The Alchemical database instance and models
  - queries.py      # The database queries that support the orders table pagination
  - routes.py       # A Flask blueprint with the two routes of the application
  - templates       # Flask templates directory
    - index.html    # The HTML page of the application
- migrations        # The Alembic migrations directory
- alembic.ini       # The Alembic configuration file
- .flaskenv         # Flask-specific environment variables
- .env.template     # Template for the environment variables needed
- requirements.txt  # the dependencies used by this application

retrofun 目录是一个包含应用程序逻辑的 Python 包。该包中的 retrofun/queries.py 和 retrofun/routes.py 已在本章前面的章节中描述。

retrofun/models.py 模块定义了 Alchemical 数据库实例 db 以及所有模型，这些模型与第6章中的定义一致，并扩展了每个模型的 to_dict() 方法。

该应用没有 db.py 模块来存放数据库初始化代码，因为使用 Alchemical 时，数据库只需一行代码即可初始化。因此，这部分初始化代码已移至 models.py。以下是 models.py 中的数据库初始化方式：

from alchemical.flask import Alchemical

db = Alchemical()

从 alchemical.flask 包导入 Alchemical 类，以启用 Flask 集成。数据库 URL 从 Flask 配置中的 ALCHEMICAL_DATABASE_URL 获取，该配置定义在 config.py 模块中：

import os

ALCHEMICAL_DATABASE_URL = os.environ.get('DATABASE_URL')

retrofun/app.py 负责使用工厂模式初始化 Flask 应用实例。models.py 中创建的 db 数据库实例作为标准的 Flask 扩展，因此在工厂函数中进行初始化。具体实现如下：

from flask import Flask
from .models import db


def create_app():
    app = Flask(__name__)
    app.config.from_object('config')

    db.init_app(app)

    from .routes import bp
    app.register_blueprint(bp)

    return app

项目顶级目录下的 main.py 模块调用 create_app() 函数：

from retrofun.app import create_app

app = create_app()

使用 flask run 命令启动应用时，.flaskenv 文件将 main.py 配置为应用定义的位置，并设置调试模式：

FLASK_APP=main.py
FLASK_DEBUG=true

当使用 Gunicorn 等 WSGI 生产级 Web 服务器时，应用实例的位置通过 main:app 表示。以下是使用 Gunicorn 启动 Web 服务器的方式：

(venv) $ gunicorn -b :5000 main:app

flask 命令会自动导入 .env 文件中定义的变量，但其他 Web 服务器不会。retrofun/__init__.py 模块会调用 load_dotenv() 函数，以防 Web 服务器未自动加载环境变量：

from dotenv import load_dotenv

load_dotenv()

Alembic 数据库迁移仓库由 Alchemical 包创建，因此与你使用 alembic init 命令创建的版本略有不同。要使用 Alchemical 创建迁移仓库，需执行以下命令：

(venv) $ python -m alchemical.alembic.cli init migrations

这样可确保 alembic.ini 和 migrations/env.py 文件与 Alchemical 兼容。采用这种方式创建仓库后，唯一需要配置的项是修改 alembic.ini 文件中的 alchemical_db 值：

alchemical_db = retrofun.models:db

requirements.txt 文件列出了运行应用所需的所有依赖项。可通过以下命令安装它们：

(venv) $ pip install -r requirements.txt

该项目不包含 .env 文件，因为该文件的内容取决于您希望与应用程序一起使用的数据库。包含了一个 .env.template 文件作为真实 .env 文件的模板。您应该将此文件复制为 .env 并重设 DATABASE_URL 变量的值为您自己的数据库地址。此应用程序与您在此书早期章节中创建的数据库兼容，因此您可以使用相同的数据库 URL 进行测试。

FastAPI 后端

FastAPI 示例也已在 GitHub 仓库中提供。

这是本项目的结构：

- retrofun          # Python package with the application logic
  - __init__.py     # Package initialization (load environment variables)
  - app.py          # The FastAPI application instance
  - models.py       # The Alchemical database instance and models
  - queries.py      # The database queries that support the orders table pagination
  - router.py       # An API router with the two routes of the application
  - html            # HTML pages directory
    - index.html    # The HTML page for the application
- migrations        # The Alembic migrations directory
- alembic.ini       # The Alembic configuration file
- .env.template     # Template for the environment variables needed
- requirements.txt  # the dependencies used by this application

与 Flask 版本类似，retrofun 目录是一个 Python 包，包含应用程序逻辑。同样，类似于 Flask 版本，retrofunc/__init__.py 文件会从 .env 文件中导入环境变量。

retrofun/app.py 模块初始化 FastAPI 应用程序并注册其路由。

from fastapi import FastAPI
from .router import router

app = FastAPI()
app.include_router(router)

retrofun/models.py 模块定义了 db 数据库实例，并包含了所有模型，这些模型与第 7 章中为异步应用程序定义的模型匹配，每个模型类中都添加了 to_dict() 方法。以下是为此应用程序定义的 Alchemical 数据库实例：

import os
from alchemical.aio import Alchemical

db = Alchemical(os.environ['DATABASE_URL'])

从 alchemical.aio 包导入了 Alchemical 的异步版本。此版本进行了所有必要的调整，以防止隐式数据库查询。

retrofun/queries.py 模块与框架无关，因此它正是本章前面所描述的内容，也是 Flask 应用程序所使用的内容。retrofun/router.py 模块定义了应用程序的两个路由，并且之前已经介绍过。

与 Flask 版本一样，Alembic 数据库迁移仓库已创建为与 Alchemical 包兼容。它不是使用 alembic init 命令创建的，而是 Alchemical 提供了自己的方法来创建仓库，以便将其配置插入其中。

(venv) $ python -m alchemical.alembic.cli init migrations

对于使用此命令创建的迁移仓库，唯一需要进行的配置是编辑 alembic.ini 中的 alchemical_db 变量，使其指向 db 数据库实例。

alchemical_db = retrofun.models:db

与大多数 Python 项目一样，包含了一个 requirements.txt 文件，列出了应用程序所需的所有依赖项。这些依赖项可通过以下方式使用 pip 安装：

(venv) $ pip install -r requirements.txt

.env 文件未包含在代码仓库中，因为其内容取决于您打算与应用程序一起使用的数据库。包含了一个 .env.template 文件，用作创建 .env 文件时的参考示例。要配置项目，请将 .env.template 文件复制为 .env，然后适当地设置 DATABASE_URL 变量的值。本项目使用您按照本书创建的 retrofun 数据库，因此您可以将其指向同一数据库，但请确保使用异步数据库驱动程序。

最后的话

恭喜您完成了本书的阅读！与其他技术主题一样，学习 SQLAlchemy 并未就此结束，对您和对我而言，这段旅程仍将继续。

尽管我尽力涵盖广泛的使用场景和解决方案，SQLAlchemy 是一个非常庞大的框架，无法完全通过本书的教程形式来全面覆盖。好消息是，这个库的所有细节都在官方文档中有详细说明。

我想特别提及三个我尚未涵盖的领域，如果您有兴趣可以自行深入研究：

子查询和 CTEs 是标准的 SQL 特性，提供了两种不同的方法来实现可以从其他查询递归调用的查询。SQLAlchemy ORM 同时支持这两种方式。

SQLAlchemy 拥有一个相当复杂的事件子系统，允许应用程序通过回调函数在特定事件发生时收到通知。本书的异步版本数据库中添加了一个事件处理程序，但还有很多其他方式可以利用这一功能。

ORM 模块包含多个可选的扩展，但只有实现异步支持的那个在本书中有所介绍。还有其他一些非常有用的扩展，例如 Automap（用于根据数据库模式生成模型类）、Association Proxy（用于简化多关系间的导航）以及 Hybrid Attributes（用于定义基于其他属性计算值的模型属性），都值得深入探索。

祝你在 SQLAlchemy 项目中一切顺利！

感谢访问我的博客！如果你喜欢这篇文章，请考虑通过“Buy me a coffee”支持我的工作，并通过小额一次性捐赠让我保持精力充沛。谢谢！