将 SQLAlchemy 2.0 与 FastAPI 结合使用的模式和实践

虽然 Django 和 Flask 仍然是许多 Python 工程师的首选， 已经被认为是无可否认的可靠选择。它是一个高度灵活、优化良好的结构化框架，为开发人员构建后端应用程序提供了无限的可能性。 但 FastAPI 使用数据库是大多数后端应用程序的一个重要方面。因此，ORM 在后端代码中起着至关重要的作用。然而，与 Django 不同的是，FastAPI 没有内置 ORM。选择合适的库并将其集成到代码库中完全是开发人员的责任。 Python 工程师普遍认为 是最流行的 ORM。这是一个传奇性的库，自 2006 年以来一直在使用，并已被数千个项目采用。 2023年，它获得了2.0版本的重大更新。与 FastAPI 类似，SQLAlchemy 为开发人员提供了强大的功能和实用程序，而不强迫他们以特定方式使用它们。从本质上讲，它是一个多功能工具包，使开发人员能够以他们认为合适的方式使用它。 SQLAlchemy FastAPI 和 SQLAlchemy 是天作之合。它们都是可靠、高性能的现代技术，可以创建强大且独特的应用程序。本文探讨如何创建一个使用 SQLAlchemy 2.0 作为 ORM 的 FastAPI 后端应用程序。内容涵盖： 使用 和 构建模型 Mapped mapped_column 定义抽象模型 处理数据库会话 使用 ORM 为所有模型创建公共存储库类 准备测试设置并添加测试 之后，您将能够轻松地将 FastAPI 应用程序与 SQLAlchemy ORM 结合起来。此外，您还将熟悉创建结构良好、健壮且高性能的应用程序的最佳实践和模式。 先决条件 本文中包含的代码示例来自 项目，该项目是用于创建和读取成分和药水对象的基本 API。文章的主要重点是探索 FastAPI 和 SQLAlchemy 的结合。它不涵盖其他主题，例如： alchemist 配置 Docker 设置 启动 uvicorn 服务器 设置 linting 如果您对这些主题感兴趣，您可以通过检查代码库自行探索它们。要访问 Alchemist 项目的代码存储库，请点击 的链接。此外，您还可以在下面找到该项目的文件结构： 此处 alchemist ├─ alchemist │ ├─ api │ │ ├─ v1 │ │ │ ├─ __init__.py │ │ │ └─ routes.py │ │ ├─ v2 │ │ │ ├─ __init__.py │ │ │ ├─ dependencies.py │ │ │ └─ routes.py │ │ ├─ __init__.py │ │ └─ models.py │ ├─ database │ │ ├─ __init__.py │ │ ├─ models.py │ │ ├─ repository.py │ │ └─ session.py │ ├─ __init__.py │ ├─ app.py │ └─ config.py ├─ requirements │ ├─ base.txt │ └─ dev.txt ├─ scripts │ ├─ create_test_db.sh │ ├─ migrate.py │ └─ run.sh ├─ tests │ ├─ conftest.py │ └─ test_api.py ├─ .env ├─ .gitignore ├─ .pre-commit-config.yaml ├─ Dockerfile ├─ Makefile ├─ README.md ├─ docker-compose.yaml ├─ example.env └─ pyproject.toml 虽然树看起来很大，但有些内容与本文的要点无关。此外，代码可能看起来比某些领域所需的代码更简单。例如，该项目缺乏： Dockerfile 中的生产阶段 迁移的 Alembic 设置 测试子目录 这样做是有意为之，以降低复杂性并避免不必要的开销。但是，如果处理更适合生产的项目，请务必牢记这些因素。 API 要求 开始开发应用程序时，考虑您的应用程序将使用的模型至关重要。这些模型将代表您的应用程序将使用的 和 ，并将在 API 中公开。就炼金术士应用程序而言，有两个实体：成分和药水。 API 应允许创建和检索这些实体。 文件包含将在 API 中使用的模型： 对象 实体 alchemist/api/models.py import uuid from pydantic import BaseModel, Field class Ingredient(BaseModel): """Ingredient model.""" pk: uuid.UUID name: str class Config: orm_mode = True class IngredientPayload(BaseModel): """Ingredient payload model.""" name: str = Field(min_length=1, max_length=127) class Potion(BaseModel): """Potion model.""" pk: uuid.UUID name: str ingredients: list[Ingredient] class Config: orm_mode = True class PotionPayload(BaseModel): """Potion payload model.""" name: str = Field(min_length=1, max_length=127) ingredients: list[uuid.UUID] = Field(min_items=1) API 将返回 和 模型。在配置中将 设置为 将使将来更容易使用 SQLAlchemy 对象。 模型将用于创建新对象。 Ingredient Potion orm_mode True Payload pydantic 的使用使得类的角色和功能更加详细和清晰。现在，是时候创建数据库模型了。 声明模型 模型本质上是某种事物的 。在 API 上下文中，模型表示后端在请求正文中期望的内容以及将在响应数据中返回的内容。另一方面，数据库模型更加复杂，表示数据库中存储的 以及它们之间的 类型。 表示 数据结构 关系 文件包含成分和药水对象的模型： alchemist/database/models.py import uuid from sqlalchemy import Column, ForeignKey, Table, orm from sqlalchemy.dialects.postgresql import UUID class Base(orm.DeclarativeBase): """Base database model.""" pk: orm.Mapped[uuid.UUID] = orm.mapped_column( primary_key=True, default=uuid.uuid4, ) potion_ingredient_association = Table( "potion_ingredient", Base.metadata, Column("potion_id", UUID(as_uuid=True), ForeignKey("potion.pk")), Column("ingredient_id", UUID(as_uuid=True), ForeignKey("ingredient.pk")), ) class Ingredient(Base): """Ingredient database model.""" __tablename__ = "ingredient" name: orm.Mapped[str] class Potion(Base): """Potion database model.""" __tablename__ = "potion" name: orm.Mapped[str] ingredients: orm.Mapped[list["Ingredient"]] = orm.relationship( secondary=potion_ingredient_association, backref="potions", lazy="selectin", ) SQLAlchemy 中的每个模型都以 类开始。继承它允许构建与 Python 类型检查器兼容的数据库模型。 DeclarativeBase 创建一个 模型（在本例中 类）也是一个很好的做法，其中包含所有模型所需的字段。这些字段包括主键，它是每个对象的唯一标识符。抽象模型通常还存储对象的创建和更新日期，这些日期是在创建或更新对象时自动设置的。然而， 模型将保持简单。 抽象 Base Base 继续到 模型， 属性指定数据库表的名称，而 字段使用新的 SQLAlchemy 语法，允许使用类型注释来声明模型字段。这种简洁而现代的方法对于类型检查器和 IDE 来说既强大又有利，因为它将 字段识别为字符串。 Ingredient __tablename__ name name 模型中的事情变得更加复杂。它还包括 和 属性，但最重要的是，它存储与成分的关系。 的使用表明药水可以包含多种成分，在这种情况下，关系是多对多（M2M）。这意味着一种成分可以分配给多种药剂。 Potion __tablename__ name Mapped[list["Ingredient"]] M2M 需要额外的配置，通常涉及创建存储两个实体之间的连接的关联表。在这种情况下， 对象仅存储成分和药水的标识符，但它也可以包含额外的属性，例如药水所需的特定成分的数量。 potion_ingredient_association 函数配置药水及其成分之间的关系。 参数指定应如何加载相关项。换句话说：当你获取药水时，SQLAlchemy 应该如何处理相关成分。将其设置为 意味着成分将与药水一起加载，从而无需在代码中进行其他查询。 relationship lazy selectin 使用 ORM 时，构建设计良好的模型至关重要。完成此操作后，下一步就是建立与数据库的连接。 会话处理程序 使用数据库时，特别是使用 SQLAlchemy 时，必须了解以下概念： 方言 引擎 联系 连接池 会议 在所有这些术语中，最重要的一个是 。根据SQLAlchemy文档，引擎对象负责连接 和 以促进数据库连接和行为。简单来说，引擎对象是数据库连接的源，而 提供高级功能，例如执行 SQL 语句、管理事务以及从数据库检索结果。 发动机 Pool Dialect 连接 是一个工作单元，它将相关操作分组到单个事务中。它是对底层数据库连接的抽象，可以有效地管理连接和事务行为。 会话 是一个为特定数据库后端提供支持的组件。它充当 SQLAlchemy 和数据库之间的中介，处理通信的细节。 alchemist 项目使用 Postgres 作为数据库，因此方言必须与此特定数据库类型兼容。 Dialect 最后一个问号是 。在 SQLAlchemy 的上下文中，连接池是一种管理数据库连接集合的机制。它旨在通过重用现有连接而不是为每个请求创建新连接来提高数据库操作的性能和效率。通过重用连接，连接池减少了建立新连接和拆除连接的开销，从而提高了性能。 连接池 掌握了这些知识后，您现在可以查看 文件，其中包含一个将用作连接数据库的依赖项的函数： alchemist/database/session.py from collections.abc import AsyncGenerator from sqlalchemy import exc from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.config import settings async def get_db_session() -> AsyncGenerator[AsyncSession, None]: engine = create_async_engine(settings.DATABASE_URL) factory = async_sessionmaker(engine) async with factory() as session: try: yield session await session.commit() except exc.SQLAlchemyError as error: await session.rollback() raise 要注意的第一个重要细节是函数 是一个生成器函数。这是因为 FastAPI 依赖系统支持生成器。因此，该函数可以处理成功和失败的情况。 get_db_session 函数的前两行创建一个数据库引擎和一个会话。然而，会话对象也可以用作上下文管理器。这使您可以更好地控制潜在的异常情况和成功的结果。 get_db_session 尽管 SQLAlchemy 处理连接的关闭，但最好在连接完成后显式声明如何处理连接。在 函数中，如果一切顺利，则提交会话；如果引发异常，则回滚会话。 get_db_session 请务必注意，此代码是围绕 asyncio 扩展构建的。 SQLAlchemy 的这一功能允许应用程序与数据库异步交互。这意味着对数据库的请求不会阻止其他 API 请求，从而使应用程序更加高效。 设置模型和连接后，下一步是确保将模型添加到数据库中。 快速迁移 SQLAlchemy 模型表示数据库的结构。但是，简单地创建它们不会导致数据库立即发生更改。要进行更改，您必须首先 它们。这通常是使用迁移库（例如 alembic）来完成的，它跟踪每个模型并相应地更新数据库。 应用 由于在此场景中没有计划对模型进行进一步的更改，因此基本的迁移脚本就足够了。下面是 文件中的示例代码。 scripts/migrate.py import asyncio import logging from sqlalchemy.ext.asyncio import create_async_engine from alchemist.config import settings from alchemist.database.models import Base logger = logging.getLogger() async def migrate_tables() -> None: logger.info("Starting to migrate") engine = create_async_engine(settings.DATABASE_URL) async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) logger.info("Done migrating") if __name__ == "__main__": asyncio.run(migrate_tables()) 简而言之， 函数读取模型的结构并使用 SQLAlchemy 引擎在数据库中重新创建它。要运行此脚本，请使用 命令。 migrate_tables python scripts/migrate.py 这些模型现在同时存在于代码和数据库中，并且 可以促进与数据库的交互。您现在可以开始处理 API 逻辑。 get_db_session API 与 ORM 如前所述，成分和药水的 API 旨在支持三种操作： 创建对象 列出对象 通过ID检索对象 由于之前的准备工作，所有这些功能已经可以使用 SQLAlchemy 作为 ORM 和 FastAPI 作为 Web 框架来实现。首先，查看位于 文件中的成分 API： alchemist/api/v1/routes.py import uuid from fastapi import APIRouter, Depends, HTTPException, status from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.api import models from alchemist.database import models as db_models from alchemist.database.session import get_db_session router = APIRouter(prefix="/v1", tags=["v1"]) @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = db_models.Ingredient(**data.dict()) session.add(ingredient) await session.commit() await session.refresh(ingredient) return models.Ingredient.from_orm(ingredient) @router.get("/ingredients", status_code=status.HTTP_200_OK) async def get_ingredients( session: AsyncSession = Depends(get_db_session), ) -> list[models.Ingredient]: ingredients = await session.scalars(select(db_models.Ingredient)) return [models.Ingredient.from_orm(ingredient) for ingredient in ingredients] @router.get("/ingredients/{pk}", status_code=status.HTTP_200_OK) async def get_ingredient( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Ingredient: ingredient = await session.get(db_models.Ingredient, pk) if ingredient is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Ingredient does not exist", ) return models.Ingredient.from_orm(ingredient) 在 API 下，有 3 种路线可用。 POST 端点将成分有效负载作为来自先前创建的模型和数据库会话的对象。 生成器函数初始化会话并启用数据库交互。 /ingredients get_db_session 在实际的函数体中，发生了五个步骤： 成分对象是根据传入的有效负载创建的。 会话对象的 方法将成分对象添加到会话跟踪系统并将其标记为待插入数据库。 add 会议已提交。 刷新成分对象以确保其属性与数据库状态匹配。 使用 方法将数据库成分实例转换为 API 模型实例。 from_orm 为了进行快速测试，可以针对正在运行的应用程序执行简单的curl： curl -X 'POST' \ 'http://localhost:8000/api/v1/ingredients' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty water"}' 在响应中，应该有一个成分对象，其 ID 来自数据库： { "pk":"2eb255e9-2172-4c75-9b29-615090e3250d", "name":"Salty water" } 尽管 SQLAlchemy 的多层抽象对于简单的 API 来说似乎是不必要的，但它们使 ORM 细节保持分离，并有助于 SQLAlchemy 的效率和可扩展性。当与 asyncio 结合使用时，ORM 功能在 API 中表现得非常好。 其余两个端点不太复杂并且有相似之处。值得更深入解释的一部分是 函数内 方法的使用。使用 SQLAlchemy 查询数据库时， 方法通常与查询作为参数一起使用。虽然 方法返回类似行的元组， 直接返回ORM实体，使端点更清晰。 get_ingredients scalars execute execute scalars 现在，考虑同一个文件中的 potions API： @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: data_dict = data.dict() ingredients = await session.scalars( select(db_models.Ingredient).where( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) ) potion = db_models.Potion(**data_dict, ingredients=list(ingredients)) session.add(potion) await session.commit() await session.refresh(potion) return models.Potion.from_orm(potion) @router.get("/potions", status_code=status.HTTP_200_OK) async def get_potions( session: AsyncSession = Depends(get_db_session), ) -> list[models.Potion]: potions = await session.scalars(select(db_models.Potion)) return [models.Potion.from_orm(potion) for potion in potions] @router.get("/potions/{pk}", status_code=status.HTTP_200_OK) async def get_potion( pk: uuid.UUID, session: AsyncSession = Depends(get_db_session), ) -> models.Potion: potion = await session.get(db_models.Potion, pk) if potion is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Potion does not exist", ) return models.Potion.from_orm(potion) 药水的 GET 端点与成分的 GET 端点相同。但是，POST 函数需要额外的代码。这是因为创建药水涉及至少包含一个成分 ID，这意味着必须获取这些成分并将其链接到新创建的药水。为了实现这一点，再次使用 方法，但这次使用指定所获取成分 ID 的查询。魔药制作过程的其余部分与原料的制作过程相同。 scalars 要测试端点，可以再次执行curl命令。 curl -X 'POST' \ 'http://localhost:8000/api/v1/potions' \ -H 'accept: application/json' \ -H 'Content-Type: application/json' \ -d '{"name": "Salty soup", "ingredients": ["0b4f1de5-e780-418d-a74d-927afe8ac954"}' 其结果如下： { "pk": "d4929197-3998-4234-a5f7-917dc4bba421", "name": "Salty soup", "ingredients": [ { "pk": "0b4f1de5-e780-418d-a74d-927afe8ac954", "name": "Salty water" } ] } 重要的是要注意，由于关系中指定的 参数，每种成分都表示为药剂中的完整对象。 lazy="selectin" API 可以正常运行，但代码存在一个重大问题。虽然 SQLAlchemy 让您可以自由地与数据库进行交互，但它不提供任何类似于 Django 的 的高级“管理器”实用程序。因此，您需要自己创建它，这本质上是成分和药水 API 中使用的逻辑。但是，如果您直接将此逻辑保留在端点中而不将其提取到单独的空间中，那么您最终会得到大量重复的代码。此外，对查询或模型进行更改将变得越来越难以管理。 Model.objects 下一章将介绍存储库模式：一种用于提取 ORM 代码的优雅解决方案。 存储库 模式允许抽象出使用数据库的细节。在使用 SQLAlchemy 的情况下，例如炼金术士的示例，存储库类将负责管理多个模型并与数据库会话交互。 存储库 看一下 文件中的以下代码： alchemist/database/repository.py import uuid from typing import Generic, TypeVar from sqlalchemy import BinaryExpression, select from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models Model = TypeVar("Model", bound=models.Base) class DatabaseRepository(Generic[Model]): """Repository for performing database queries.""" def __init__(self, model: type[Model], session: AsyncSession) -> None: self.model = model self.session = session async def create(self, data: dict) -> Model: instance = self.model(**data) self.session.add(instance) await self.session.commit() await self.session.refresh(instance) return instance async def get(self, pk: uuid.UUID) -> Model | None: return await self.session.get(self.model, pk) async def filter( self, *expressions: BinaryExpression, ) -> list[Model]: query = select(self.model) if expressions: query = query.where(*expressions) return list(await self.session.scalars(query)) 类保存先前包含在端点中的所有逻辑。不同之处在于，它允许在 方法中传递特定的模型类，从而可以重用所有模型的代码，而不是在每个端点中重复它。 DatabaseRepository __init__ 此外， 使用 Python 泛型，其中 泛型类型绑定到抽象数据库模型。这允许存储库类从静态类型检查中获得更多好处。当与特定模型一起使用时，存储库方法的返回类型将反映该特定模型。 DatabaseRepository Model 因为存储库需要使用数据库会话，所以它必须与 依赖项一起初始化。考虑 文件中的新依赖项： get_db_session alchemist/api/v2/dependencies.py from collections.abc import Callable from fastapi import Depends from sqlalchemy.ext.asyncio import AsyncSession from alchemist.database import models, repository, session def get_repository( model: type[models.Base], ) -> Callable[[AsyncSession], repository.DatabaseRepository]: def func(session: AsyncSession = Depends(session.get_db_session)): return repository.DatabaseRepository(model, session) return func 简单来说， 函数就是一个依赖工厂。它首先采用您将使用存储库的数据库模型。然后，它返回依赖项，该依赖项将用于接收数据库会话并初始化存储库对象。为了更好地理解，请从 文件中查看新的 API。它仅显示 POST 端点，但它应该足以让您更清楚地了解代码如何改进： get_repository alchemist/api/v2/routes.py from typing import Annotated from fastapi import APIRouter, Depends, status from alchemist.api import models from alchemist.api.v2.dependencies import get_repository from alchemist.database import models as db_models from alchemist.database.repository import DatabaseRepository router = APIRouter(prefix="/v2", tags=["v2"]) IngredientRepository = Annotated[ DatabaseRepository[db_models.Ingredient], Depends(get_repository(db_models.Ingredient)), ] PotionRepository = Annotated[ DatabaseRepository[db_models.Potion], Depends(get_repository(db_models.Potion)), ] @router.post("/ingredients", status_code=status.HTTP_201_CREATED) async def create_ingredient( data: models.IngredientPayload, repository: IngredientRepository, ) -> models.Ingredient: ingredient = await repository.create(data.dict()) return models.Ingredient.from_orm(ingredient) @router.post("/potions", status_code=status.HTTP_201_CREATED) async def create_potion( data: models.PotionPayload, ingredient_repository: IngredientRepository, potion_repository: PotionRepository, ) -> models.Potion: data_dict = data.dict() ingredients = await ingredient_repository.filter( db_models.Ingredient.pk.in_(data_dict.pop("ingredients")) ) potion = await potion_repository.create({**data_dict, "ingredients": ingredients}) return models.Potion.from_orm(potion) 需要注意的第一个重要功能是 的使用，这是一种处理 FastAPI 依赖项的新方法。通过将依赖项的返回类型指定为 并使用 声明其用法，您最终可以在端点中使用简单的类型注释： 。 Annotated DatabaseRepository[db_models.Ingredient] Depends(get_repository(db_models.Ingredient)) repository: IngredientRepository 感谢存储库，端点不必存储所有与 ORM 相关的负担。即使在更复杂的药剂情况下，您所需要做的就是同时使用两个存储库。 您可能想知道初始化两个存储库是否会初始化会话两次。答案是不。 FastAPI 依赖系统在单个请求中缓存相同的依赖调用。这意味着会话初始化被缓存，并且两个存储库使用完全相同的会话对象。 SQLAlchemy 和 FastAPI 组合的另一个伟大功能。 该 API 功能齐全，并具有可重用的高性能数据访问层。下一步是通过编写一些端到端测试来确保满足要求。 测试 测试在软件开发中起着至关重要的作用。项目可以包含单元、集成和端到端 (E2E) 测试。虽然通常最好进行大量有意义的单元测试，但至少编写一些 E2E 测试以确保整个工作流程正常运行也很好。 要为 Alchemist 应用程序创建一些 E2E 测试，需要两个额外的库： pytest 实际创建并运行测试 httpx 在测试中发出异步请求 安装这些之后，下一步就是建立一个单独的测试数据库。您不希望您的默认数据库被污染或删除。由于 Alchemist 包含 Docker 设置，因此只需要一个简单的脚本来创建第二个数据库。看一下 文件中的代码： scripts/create_test_db.sh #!/bin/bash psql -U postgres psql -c "CREATE DATABASE test" 为了执行脚本，必须将其作为卷添加到 Postgres 容器中。这可以通过将其包含在 文件的 部分来实现。 docker-compose.yaml volumes 准备工作的最后一步是在 文件中创建pytest装置： tests/conftest.py from collections.abc import AsyncGenerator import pytest import pytest_asyncio from fastapi import FastAPI from httpx import AsyncClient from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from alchemist.app import app from alchemist.config import settings from alchemist.database.models import Base from alchemist.database.session import get_db_session @pytest_asyncio.fixture() async def db_session() -> AsyncGenerator[AsyncSession, None]: """Start a test database session.""" db_name = settings.DATABASE_URL.split("/")[-1] db_url = settings.DATABASE_URL.replace(f"/{db_name}", "/test") engine = create_async_engine(db_url) async with engine.begin() as conn: await conn.run_sync(Base.metadata.drop_all) await conn.run_sync(Base.metadata.create_all) session = async_sessionmaker(engine)() yield session await session.close() @pytest.fixture() def test_app(db_session: AsyncSession) -> FastAPI: """Create a test app with overridden dependencies.""" app.dependency_overrides[get_db_session] = lambda: db_session return app @pytest_asyncio.fixture() async def client(test_app: FastAPI) -> AsyncGenerator[AsyncClient, None]: """Create an http client.""" async with AsyncClient(app=test_app, base_url="http://test") as client: yield client 在测试中需要改变的一件事是应用程序与数据库的交互方式。这不仅包括更改数据库 URL，还包括通过从空数据库开始来确保每个测试都是隔离的。 装置实现了这两个目标。其主体采取以下步骤： db_session 使用修改后的数据库 URL 创建引擎。 删除所有现有表以确保测试拥有干净的数据库。 创建数据库内的所有表（与迁移脚本中的代码相同）。 创建并生成一个会话对象。 测试完成后手动关闭会话。 尽管最后一步也可以作为上下文管理器来实现，但在这种情况下手动关闭效果很好。 剩下的两个装置应该是不言自明的： 是 文件中的 FastAPI 实例，其中 依赖项替换为 固定装置 test_app alchemist/app.py get_db_session db_session 是 httpx ，它将针对 发出 API 请求 client AsyncClient test_app 所有这些都设置完毕后，最后就可以编写实际的测试了。为了简洁起见， 文件中的以下示例仅显示了创建成分的测试： tests/test_api.py from fastapi import status class TestIngredientsAPI: """Test cases for the ingredients API.""" async def test_create_ingredient(self, client): response = await client.post("/api/v2/ingredients", json={"name": "Carrot"}) assert response.status_code == status.HTTP_201_CREATED pk = response.json().get("pk") assert pk is not None response = await client.get("/api/v2/ingredients") assert response.status_code == status.HTTP_200_OK assert len(response.json()) == 1 assert response.json()[0]["pk"] == pk 该测试使用在固定装置中创建的客户端对象，该对象向具有覆盖依赖项的 FastAPI 实例发出请求。因此，测试能够与一个单独的数据库交互，该数据库将在测试完成后被清除。两个 API 的其余测试套件的结构几乎相同。 概括 FastAPI 和 SQLAlchemy 是用于创建现代且强大的后端应用程序的出色技术。它们提供的自由、简单和灵活性使它们成为基于 Python 的项目的最佳选择之一。如果开发人员遵循最佳实践和模式，他们就可以创建高性能、健壮且结构良好的应用程序，轻松处理数据库操作和 API 逻辑。本文旨在让您更好地了解如何设置和维护这个令人惊叹的组合。 来源 炼金术士项目的源代码可以在这里找到： 。 链接 也发布 。 在这里