使用Sqlalchemy 2.0与FastApi一起使用的模式和实践-DEV365 开发者社区

虽然Django和Blask仍然是许多Python工程师的第一批选择，但 fastapi 已经被认为是无可否认的可靠选择。这是一个高度灵活，优化，结构化的框架，它为开发人员提供了构建后端应用程序的无尽可能性。

使用数据库是大多数后端应用程序的重要方面。结果，ORM在后端代码中起着至关重要的作用。但是，与Django不同，Fastapi没有内置的ORM。选择合适的库并将其集成到代码库中完全是开发人员的责任。

Python工程师广泛认为 sqlalchemy 是最受欢迎的ORM。这是一个自2006年以来一直在使用的传奇图书馆，并已被成千上万的项目采用。在2023年，它收到了2.0版的重大更新。与FastApi类似，Sqlalchemy为开发人员提供了强大的功能和实用程序，而无需以特定的方式使用它们。本质上，这是一个多功能工具包，使开发人员使用它，但他们认为合适。

fastapi和sqlalchemy是天堂做的匹配。它们都是可靠，表现和现代技术，可以创建强大而独特的应用程序。本文探讨了创建使用SQLalchemy 2.0作为ORM的FastAPI后端应用程序。内容涵盖：

使用Mapped和mapped_column构建模型
定义抽象模型
处理数据库会话
使用ORM
为所有模型创建一个通用存储库类
准备测试设置并添加测试

之后，您将可以轻松地将FastApi应用程序与SQLalchemy Orm相结合。此外，您将熟悉最佳实践和模式，以创建结构良好，健壮和性能的应用程序。

先决条件

本文中包含的代码示例来自 alchemist '项目，这是用于创建和读取成分和药水对象的基本API。本文的主要重点是探索Fastapi和Sqlalchemy的组合。它不涵盖其他主题，例如：

配置Docker设置
启动Uvicorn Server
setting up linting

如果您对这些主题感兴趣，则可以通过检查代码库独自探索它们。要访问Alchemist项目的代码存储库，请点击此链接here。此外，您可以找到以下项目的文件结构：

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应允许创建和检索这些实体。 alchemist/api/models.py文件包含将在API中使用的模型：

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将返回Ingredient和Potion型号。将orm_mode设置为配置中的True将使将来与Sqlalchemy对象一起使用。 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中的每个模型都始于DeclarativeBase类。从中继承允许构建与Python类型检查器兼容的数据库模型。

在这种情况下，创建Anâ摘要 appract “模型Base”类，其中包括所有模型中所需的字段。这些字段包括主键，这是每个对象的唯一标识符。抽象模型通常还存储对象的创建和更新日期，该对象是在创建或更新对象时自动设置的。但是，Base模型将保持简单。

转到Ingredient模型，__tablename__属性指定数据库表的名称，而name字段使用新的SQLalchemy语法，允许使用类型注释来声明模型字段。这种简洁而现代的方法对于类型的调查器和IDE既有功能又有优势，因为它识别为name字段是字符串。

在Potion模型中，事情变得更加复杂。它还包括__tablename__和name属性，但最重要的是，它存储了与成分的关系。 Mapped[list["Ingredient"]]的使用表明该药水可以包含多种成分，在这种情况下，这种关系是多对多的（M2M）。这意味着可以将单个成分分配给多种药水。

m2m需要附加配置，通常涉及创建协会表，该联想表存储两个实体之间的连接。在这种情况下，potion_ingredient_association对象仅存储成分和药水的标识符，但它也可能包括额外的属性，例如药水所需的特定成分的数量。

relationship功能配置了药水及其成分之间的关系。 lazy参数指定应如何加载相关项目。换句话说：当您取药时，Sqlalchemy应该如何处理相关成分。将其设置为selectin意味着将加载水药，从而消除了代码中其他查询的需求。

与ORM一起工作时，建立精心设计的模型至关重要。完成此操作后，下一步就是建立与数据库的连接。

会话处理程序

使用数据库时，尤其是使用SQLalchemy时，必须了解以下概念：

dialect
英语
连接
连接池
会话

在所有这些术语中，最重要的是引擎。根据SQLalchemy文档，发动机对象负责连接Pool和Dialect，以促进数据库的连接和行为。简单地说，引擎对象是数据库连接的来源，而 Connection 提供了高级功能，例如执行SQL语句，管理交易和从数据库。

a session 是一个工作单位，将单个交易中的操作分组。这是基础数据库连接的抽象，并有效地管理连接和交易行为。

方言是为特定数据库后端提供支持的组件。它充当Sqlalchemy和数据库之间的中介，处理通信的细节。炼金术士项目使用Postgres作为数据库，因此方言必须与此特定数据库类型兼容。

最终的问号是 连接池 。在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

要注意的第一个重要细节是函数get_db_session是生成器函数。这是因为FastAPI依赖性系统支持生成器。结果，此功能可以处理成功和失败的方案。

get_db_session功能的前两行创建数据库引擎和一个会话。但是，会话对象也可以用作上下文管理器。这使您可以更多地控制潜在的例外和成功的结果。

尽管Sqlalchemy处理了连接的关闭，但最好在完成连接后明确声明如何处理连接。在get_db_session函数中，如果一切顺利，则会进行会话，如果提出异常。

重要的是要注意，此代码是围绕异步扩展而构建的。 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())

简单地说，migrate_tables功能读取模型的结构，并使用SQLalchemy Engine在数据库中重新创建它。要运行此脚本，请使用python scripts/migrate.py命令。

现在，模型都存在于代码和数据库中，而get_db_session可以促进与数据库的交互。您现在可以开始使用API逻辑。

与ORM的API

如前所述，成分和药水的API旨在支持三个操作：

创建对象
列表对象
通过ID检索对象

由于先前的准备工作，所有这些功能已经可以用Sqlalchemy作为ORM和FastApi作为Web框架实现。首先，查看位于alchemist/api/v1/routes.py文件中的成分API。

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)

在/ingredients API下，有三个可用的路线。帖子端点将成分有效载荷从先前创建的模型和数据库会话中作为对象。 get_db_session生成器函数初始化了会话并启用数据库交互。

在实际功能主体中，发生了五个步骤：

从传入有效载荷创建成分对象。
会话对象的add方法将成分对象添加到会话跟踪系统中，并将其标记为未决的插入数据库。
会话是进行的。
刷新成分对象以确保其属性与数据库状态匹配。
使用from_orm方法将数据库成分实例转换为API模型实例。

对于快速测试，可以针对运行应用程序执行简单的卷发：

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中表现出色。

其余两个端点不那么复杂，并且共享相似之处。值得深入解释的一部分是在get_ingredients函数中使用scalars方法。在使用SQLalchemy查询数据库时，execute方法通常与查询一起用作参数。当execute方法返回类似行的元组时，scalars直接返回ORM实体，使端点清洁器。

现在，在同一文件中考虑药水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)

药水的获取端点与成分的终点相同。但是，后功能需要其他代码。这是因为创建药水涉及至少一个成分ID，这意味着必须将成分获取并链接到新创建的药水。为了实现这一目标，再次使用了scalars方法，但是这次使用查询指定了获取成分的ID。药水创造过程的其余部分与成分相同。

要测试端点，可以再次执行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的Model.objects的任何高级“经理”实用程序。结果，您需要自己创建它，这实际上是成分和药水API中使用的逻辑。但是，如果您将此逻辑直接保存在端点，而不将其提取到单独的空间中，则最终将获得许多重复的代码。此外，对查询或模型进行更改将变得越来越难以管理。

即将到来的章节介绍了存储库模式：提取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__方法中传递特定的模型类，从而可以重复使用所有模型的代码，而不是在每个端点中复制它。

此外，DatabaseRepository使用python generics，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

简而言之，get_repository功能是一个依赖工厂。它首先采用您将使用存储库的数据库模型。然后，它返回依赖项，该依赖项将用于接收数据库会话并初始化存储库对象。要获得更好的理解，请查看alchemist/api/v2/routes.py文件的新API。它仅显示端口端点，但应该足以让您更清楚地了解代码的改进：

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)

要注意的第一个重要功能是使用Annotated，这是一种使用FastApi依赖关系的新方法。通过将依赖项的返回类型指定为DatabaseRepository[db_models.Ingredient]并用Depends(get_repository(db_models.Ingredient))声明其用法，您可以在端点中使用简单的类型注释：repository: IngredientRepository。

感谢存储库，端点不必存储所有与ORM相关的负担。即使在更复杂的药水情况下，您需要做的就是同时使用两个存储库。

您可能会想知道初始化两个存储库是否将两次初始化会话初始化。答案是不。 FastAPI依赖项系统在单个请求中缓存了相同的依赖关系调用。这意味着会话初始化被缓存，并且两个存储库都使用完全相同的会话对象。 Sqlalchemy和Fastapi组合的另一个重要特征。

API功能齐全，具有可重复使用的高性能数据访问层。下一步是通过编写一些端到端测试来确保满足要求。

测试

测试在软件开发中起着至关重要的作用。项目可以包含单元，集成和端到端（E2E）测试。虽然通常最好进行大量有意义的单元测试，但也最好编写至少一些E2E测试以确保整个工作流程正常运行。

要为炼金术士应用程序创建一些E2E测试，需要其他两个库：

pytest实际创建和运行测试
httpx在测试中提出异步请求

安装了这些内容后，下一步就是将一个单独的测试数据库安装到位。您不希望对默认数据库进行污染或删除。由于炼金术士包含了Docker设置，因此仅需要一个简单的脚本来创建第二个数据库。从scripts/create_test_db.sh文件中查看代码：

#!/bin/bash

psql -U postgres
psql -c "CREATE DATABASE test"

为了执行脚本，必须将其作为卷中的卷添加到Postgres容器中。可以通过将其包括在docker-compose.yaml文件的volumes部分中来实现。

准备的最后一步是在tests/conftest.py文件中创建pytest灯具：

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创建引擎。
删除所有现有表以确保测试具有干净的数据库。
在数据库内创建所有表（与迁移脚本相同的代码）。
创建并产生一个会话对象。
完成测试后手动关闭会话。

尽管最后一步也可以作为上下文管理器实现，但在这种情况下，手册关闭效果很好。

剩下的两个固定装置应该相当不言自明：

test_app是alchemist/app.py文件的FastApi实例，get_db_session依赖项替换为db_session夹具
client是HTTPX AsyncClient，它将针对test_app提出API请求

所有这些都在设置后，最后可以编写实际的测试。为了简明性，下面的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逻辑。本文旨在使您对如何建立和维护这种惊人的组合有很好的了解。

来源

可以在此处找到炼金术士项目的源代码：link。