FastAPI で SQLAlchemy 2.0 を使用するためのパターンと実践

多くの Python エンジニアにとって Django と Flask が依然として最初の選択肢ですが、 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 を簡単に組み合わせることができるようになります。さらに、適切に構造化され、堅牢でパフォーマンスの高いアプリケーションを作成するためのベスト プラクティスとパターンについても理解できます。

前提条件

この記事に含まれているコード例は、材料とポーションのオブジェクトを作成および読み取るための基本的な API であるalchemistプロジェクトからのものです。この記事の主な焦点は、FastAPI と SQLAlchemy の組み合わせを探ることです。次のような他のトピックについては説明しません。

• Docker セットアップの構成
• ユビコーンサーバーを起動しています
• リンティングの設定

これらのトピックに興味がある場合は、コードベースを調べて自分で調べることができます。 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 で公開されるオブジェクトとエンティティを表します。 Alchemist アプリの場合、材料とポーションという 2 つのエンティティがあります。 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 型チェッカーと互換性のあるデータベース モデルを構築できます。

すべてのモデルに必要なフィールドを含む抽象モデル (この場合はBaseクラス) を作成することも推奨されます。これらのフィールドには、すべてのオブジェクトの一意の識別子である主キーが含まれます。抽象モデルには、多くの場合、オブジェクトの作成日と更新日も格納されます。これらの日付は、オブジェクトが作成または更新されるときに自動的に設定されます。ただし、 Baseモデルはシンプルなままになります。

Ingredientモデルに進むと、 __tablename__属性はデータベース テーブルの名前を指定しますが、 nameフィールドは新しい SQLAlchemy 構文を使用し、モデル フィールドを型注釈で宣言できるようにします。この簡潔で最新のアプローチは、 nameフィールドを文字列として認識するため、型チェッカーや IDE にとって強力かつ有利です。

Potionモデルでは事態はさらに複雑になります。 __tablename__属性とname属性も含まれますが、それに加えて、材料との関係も保存されます。 Mapped[list["Ingredient"]]を使用すると、ポーションに複数の成分を含めることができ、この場合、関係は多対多 (M2M) であることを示します。これは、単一の材料を複数のポーションに割り当てることができることを意味します。

M2M には追加の構成が必要で、これには通常、2 つのエンティティ間の接続を保存する関連付けテーブルの作成が含まれます。この場合、 potion_ingredient_associationオブジェクトには材料とポーションの識別子のみが格納されますが、ポーションに必要な特定の材料の量などの追加の属性も含めることもできます。

relationship関数は、ポーションとその成分の間の関係を構成します。 lazy引数は、関連項目をロードする方法を指定します。言い換えれば、ポーションを取得するときに、SQLAlchemy は関連する材料に対して何をすべきかということです。これをselectinに設定すると、成分がポーションとともにロードされ、コード内で追加のクエリが必要なくなることを意味します。

ORM を使用する場合、適切に設計されたモデルを構築することが重要です。これが完了したら、次のステップはデータベースとの接続を確立することです。

セッションハンドラー

データベースを操作するとき、特に SQLAlchemy を使用するときは、次の概念を理解することが重要です。

• 方言
• エンジン
• 繋がり
• 接続プール
• セッション

これらすべての用語の中で、最も重要なものはエンジンです。 SQLAlchemy のドキュメントによると、エンジン オブジェクトはPoolとDialectを接続してデータベースの接続と動作を容易にする役割を果たします。簡単に言うと、エンジン オブジェクトはデータベース接続のソースであり、接続はSQL ステートメントの実行、トランザクションの管理、データベースからの結果の取得などの高レベルの機能を提供します。

セッションは、関連する操作を 1 つのトランザクション内にグループ化する作業単位です。これは、基礎となるデータベース接続を抽象化し、接続とトランザクション動作を効率的に管理します。

Dialectは、特定のデータベース バックエンドのサポートを提供するコンポーネントです。 SQLAlchemy とデータベースの間の仲介者として機能し、通信の詳細を処理します。 Alchemist プロジェクトはデータベースとして 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関数の最初の 2 行は、データベース エンジンとセッションを作成します。ただし、セッション オブジェクトはコンテキスト マネージャーとしても使用できます。これにより、潜在的な例外と成功した結果をより詳細に制御できるようになります。

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())

簡単に言うと、 migrate_tables関数はモデルの構造を読み取り、SQLAlchemy エンジンを使用してデータベース内にそれを再作成します。このスクリプトを実行するには、 python scripts/migrate.pyコマンドを使用します。

モデルはコードとデータベースの両方に存在し、 get_db_sessionによってデータベースとの対話が容易になります。これで、API ロジックの作業を開始できます。

ORMを使用したAPI

前述したように、材料とポーションの API は 3 つの操作をサポートすることを目的としています。

• オブジェクトの作成
• オブジェクトのリスト表示
• 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 では、3 つのルートが利用可能です。 POST エンドポイントは、以前に作成されたモデルとデータベース セッションからオブジェクトとして成分ペイロードを受け取ります。 get_db_sessionジェネレーター関数はセッションを初期化し、データベースの対話を可能にします。

実際の関数本体では、次の 5 つのステップが実行されます。

1. 成分オブジェクトは受信ペイロードから作成されます。
2. セッション オブジェクトのaddメソッドは、成分オブジェクトをセッション追跡システムに追加し、データベースへの挿入が保留中としてマークします。
3. セッションはコミットされます。
4. 成分オブジェクトが更新されて、その属性がデータベースの状態と一致することが確認されます。
5. データベース コンポーネント インスタンスは、 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 で非常に優れたパフォーマンスを発揮します。

残りの 2 つのエンドポイントはそれほど複雑ではなく、類似点を共有しています。より深い説明に値する部分の 1 つは、 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)

ポーションの GET エンドポイントは、材料の GET エンドポイントと同じです。ただし、POST 関数には追加のコードが必要です。これは、ポーションの作成には少なくとも 1 つの材料 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 とポーション 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 、抽象データベース モデルにバインドされたModelジェネリック タイプを持つ Python ジェネリックを使用します。これにより、リポジトリ クラスは静的型チェックの恩恵をさらに受けられるようになります。特定のモデルで使用すると、リポジトリ メソッドの戻り値の型はこの特定のモデルを反映します。

リポジトリはデータベース セッションを使用する必要があるため、 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 を確認してください。ここには POST エンドポイントのみが示されていますが、コードがどのように改善されるかをより明確に理解するには十分です。

 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 関連の負担をすべて保存する必要がなくなります。より複雑なポーションの場合でも、必要なのは 2 つのリポジトリを同時に使用することだけです。

2 つのリポジトリを初期化するとセッションが 2 回初期化されるのではないかと疑問に思われるかもしれません。答えはいいえだ。 FastAPI 依存関係システムは、単一のリクエスト内の同じ依存関係呼び出しをキャッシュします。これは、セッションの初期化がキャッシュされ、両方のリポジトリがまったく同じセッション オブジェクトを使用することを意味します。 SQLAlchemy と FastAPI の組み合わせのもう 1 つの優れた機能。

API は完全に機能し、再利用可能でパフォーマンスの高いデータ アクセス レイヤーを備えています。次のステップでは、いくつかのエンドツーエンドのテストを作成して、要件が満たされていることを確認します。

テスト

テストはソフトウェア開発において重要な役割を果たします。プロジェクトには、単体テスト、統合テスト、エンドツーエンド (E2E) テストを含めることができます。通常は、意味のある単体テストを多数作成することが最善ですが、ワークフロー全体が正しく機能していることを確認するために、少なくともいくつかの E2E テストを作成することも推奨されます。

Alchemist アプリの E2E テストを作成するには、2 つの追加ライブラリが必要です。

• 実際にテストを作成して実行するには pytest
• httpx はテスト内で非同期リクエストを作成します

これらをインストールしたら、次のステップは、別のテスト データベースを配置することです。デフォルトのデータベースが汚染されたり削除されたりすることは望ましくありません。 Alchemist には Docker セットアップが含まれているため、2 番目のデータベースを作成するために必要なのは簡単なスクリプトだけです。 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

テストで変更する必要がある点の 1 つは、アプリがデータベースとどのように対話するかです。これには、データベース URL を変更するだけでなく、空のデータベースから開始して各テストを確実に分離することも含まれます。

db_sessionフィクスチャは、これらの両方の目標を達成します。その本体は次の手順を実行します。

1. 変更したデータベース URL を使用してエンジンを作成します。
2. 既存のテーブルをすべて削除して、テストにクリーンなデータベースが含まれていることを確認します。
3. データベース内にすべてのテーブルを作成します (移行スクリプトと同じコード)。
4. セッション オブジェクトを作成して生成します。
5. テストが完了したら、セッションを手動で閉じます。

最後のステップはコンテキスト マネージャーとして実装することもできますが、この場合は手動で閉じると問題なく機能します。

残りの 2 つのフィクスチャは、一目瞭然です。

• test_app 、 alchemist/app.pyファイルの FastAPI インスタンスであり、 get_db_session依存関係がdb_sessionフィクスチャに置き換えられています。
• client test_appに対して API リクエストを行う httpx AsyncClientです。

ここまでの設定が完了したら、いよいよ実際のテストを作成できます。簡潔にするために、 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 ベースのプロジェクトに最適なオプションの 1 つとなります。開発者がベスト プラクティスとパターンに従えば、データベース操作と API ロジックを簡単に処理できる、パフォーマンスが高く、堅牢で、適切に構造化されたアプリケーションを作成できます。この記事は、この素晴らしい組み合わせを設定および維持する方法をよく理解していただくことを目的としています。

情報源

Alchemist プロジェクトのソース コードは、リンクにあります。