FastAPIとSQLModel、Tortoise ORMによる非同期データベース操作

はじめに

急速に進化するバックエンド開発の状況において、高性能でスケーラブルなWebサービスを構築することは最重要です。非同期プログラミングは、アプリケーションをブロックすることなく多数の同時リクエストを処理できるようにすることで、そのような目標を達成するための基盤として登場しました。FastAPIは、非同期操作へのネイティブサポートと直観的な設計により、モダンAPIの構築にすぐに利用できるフレームワークになりました。しかし、データベースのやり取りが同期的なままである場合、非同期Webフレームワークの利点は著しく損なわれる可能性があります。このボトルネックは、まさに非同期ORMおよびデータベースライブラリが解決しようとしているものです。この記事では、2つの著名な非同期データベースツール（SQLModelとTortoise ORM）をFastAPIと統合し、真にノンブロッキングなデータベース操作をアンロックし、それによってバックエンドサービスの全体的な効率と応答性を向上させる方法を検討します。

コアコンセプトと原則

実装の詳細に入る前に、FastAPIにおける非同期データベース操作を理解するために不可欠ないくつかの基本的な用語を明確にしましょう。

• 非同期プログラミング: プログラム全体をフリーズすることなく、長時間実行タスクを実行できるプログラミングパラダイム。Pythonでは、主にasync / awaitキーワードとasyncioライブラリを使用して実現され、協調的なマルチタスクを可能にします。
• ORM（オブジェクトリレーショナルマッピング）: 開発者がオブジェクト指向パラダイムを使用してデータベースと対話できるようにする技術。生のSQLクエリを記述する代わりに、ORMを使用すると、好みのプログラミング言語でオブジェクトとしてデータベースレコードを操作できます。この抽象化により、データベース操作が簡素化され、コードの可読性が向上し、SQLインジェクションを防ぐことでセキュリティが向上することがよくあります。
• FastAPI: PythonType Hintに基づいた標準Python Type Hintに基づいた、API構築のためのモダンで高速（高性能）なWebフレームワーク。asyncioと深く統合されており、非同期Webサービスに理想的です。
• SQLModel: SQLデータベースとの対話のためのPythonライブラリで、「SQLとPythonファースト」になるように設計されています。PydanticとSQLAlchemyの上に構築されており、データ検証のためのPydanticモデルとデータベース操作のためのSQLAlchemyモデルの両方として機能するモデルを定義するための、よりシンプルで直観的なエクスペリエンスを提供することを目指しています。SQLAlchemyの非同期エンジンを通じて、非同期操作をネイティブにサポートします。
• Tortoise ORM: asyncioおよびuvloop専用に設計された、Python向けの使いやすい非同期ORM。モデルの定義、クエリの実行、データベースマイグレーションの管理のためのシンプルなAPIを提供し、すべて非同期性を維持します。

FastAPIで非同期ORMを使用する原則は単純です。FastAPIアプリケーションがデータベースと通信する必要がある場合、データベース応答を同期的に待機する代わりに、ORMはアプリケーションが別のタスクに切り替えることを許可します。データベース操作が完了すると、アプリケーションは元のリクエストの処理を再開できます。このノンブロッキング動作は、データベース呼び出しのようなI/Oバウンド操作では不可欠であり、重い負荷の下でサーバーが応答しなくなるのを防ぎます。

非同期データベース操作の実装

SQLModelとTortoise ORMをFastAPIアプリケーションに統合して非同期データベース対話を行う方法を説明します。デモンストレーションのために、シンプルな「Hero」モデルを使用します。

SQLModelの使用

SQLModelはPydanticモデルとSQLAlchemyをブレンドし、データ定義のための強力でエレガントな方法を提供します。

セットアップ

まず、必要なパッケージをインストールします。

pip install fastapi "uvicorn[standard]" sqlmodel "psycopg2-binary" # または asyncpg を非同期用に使用

PostgreSQLをデータベースとして使用します。PostgreSQLの非同期操作には、psycopg2-binaryよりもasyncpgが好まれることが多いですが、psycopg2-binaryは非同期ラッパーで使用できます。ネイティブな非同期エクスペリエンスのためにasyncpgを使用しましょう。

pip install fastapi "uvicorn[standard]" sqlmodel asyncpg

データベース構成とモデル定義

from typing import Optional, List
from sqlmodel import Field, SQLModel, Session, create_engine
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends, HTTPException, status

# データベースURL、必要に応じて調整してください
DATABASE_URL = "postgresql+asyncpg://user:password@host:port/dbname"

class HeroBase(SQLModel):
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = Field(default=None, index=True)

class Hero(HeroBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)

class HeroCreate(HeroBase):
    pass

class HeroPublic(HeroBase):
    id: int

# 非同期エンジン
engine = create_engine(DATABASE_URL, echo=True)

async def create_db_and_tables():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

# データベースセッションを取得するための依存関係
async def get_session():
    async with Session(engine) as session:
        yield session

# FastAPIアプリケーションセットアップ
@asynccontextmanager
async def lifespan(app: FastAPI):
    await create_db_and_tables()
    yield

app = FastAPI(lifespan=lifespan)

FastAPIエンドポイント

Heroモデルの基本的なCRUDエンドポイントをいくつか作成しましょう。

from sqlmodel import select # select をインポート

@app.post("/heroes/", response_model=HeroPublic)
async def create_hero(*, session: Session = Depends(get_session), hero: HeroCreate):
    db_hero = Hero.model_validate(hero)
    session.add(db_hero)
    await session.commit()
    await session.refresh(db_hero)
    return db_hero

@app.get("/heroes/", response_model=List[HeroPublic])
async def read_heroes(offset: int = 0, limit: int = Field(default=100, le=100), session: Session = Depends(get_session)):
    heroes = (await session.exec(select(Hero).offset(offset).limit(limit))).all()
    return heroes

@app.get("/heroes/{hero_id}", response_model=HeroPublic)
async def read_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = (await session.get(Hero, hero_id))
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    return hero

@app.put("/heroes/{hero_id}", response_model=HeroPublic)
async def update_hero(*, session: Session = Depends(get_session), hero_id: int, hero: HeroCreate):
    db_hero = (await session.get(Hero, hero_id))
    if not db_hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    
    hero_data = hero.model_dump(exclude_unset=True)
    for key, value in hero_data.items():
        setattr(db_hero, key, value)
    
    session.add(db_hero)
    await session.commit()
    await session.refresh(db_hero)
    return db_hero

@app.delete("/heroes/{hero_id}")
async def delete_hero(*, session: Session = Depends(get_session), hero_id: int):
    hero = (await session.get(Hero, hero_id))
    if not hero:
        raise HTTPException(status_code=404, detail="Hero not found")
    await session.delete(hero)
    await session.commit()
    return {"ok": True}

ここでの重要な側面は次のとおりです。

• 非同期ドライバーのためのpostgresql+asyncpgを使用したcreate_engine。
• テーブルを非同期的に作成するためのasync with engine.begin()およびawait conn.run_sync()。
• 非同期データベースセッションを管理するためのasync with Session(engine) as session:。
• 非同期クエリのためのawait session.exec()およびawait session.get()。
• 変更を保存し、オブジェクトを更新するためのawait session.commit()およびawait session.refresh()。

Tortoise ORMの使用

Tortoise ORMは最初から非同期操作のために設計されており、独自のクエリ構文でより伝統的なORMエクスペリエンスを提供します。

セットアップ

Tortoise ORMと選択した非同期データベースドライバー（例：PostgreSQLのasyncpg）をインストールします。

pip install fastapi "uvicorn[standard]" tortoise-orm asyncpg

データベース構成とモデル定義

from typing import Optional, List
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from tortoise import fields, models
from tortoise.contrib.fastapi import register_tortoise
from tortoise.exceptions import DoesNotExist

# データベース構成
TORTOISE_CONFIG = {
    "connections": {"default": "postgresql://user:password@host:port/dbname"},
    "apps": {
        "models": {
            "models": ["main"], # mainファイルにモデルがあると仮定
            "default_connection": "default",
        }
    }
}

# Tortoise ORM Heroモデル
class Hero(models.Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=255, unique=True, index=True)
    secret_name = fields.CharField(max_length=255)
    age = fields.IntField(null=True, index=True)

    class Meta:
        table = "heroes"

    def __str__(self):
        return self.name

# リクエスト/レスポンス検証のためのPydanticモデル
class HeroIn(BaseModel):
    name: str
    secret_name: str
    age: Optional[int] = None

class HeroOut(BaseModel):
    id: int
    name: str
    secret_name: str
    age: Optional[int] = None

async def init_db(app: FastAPI):
    register_tortoise(
        app,
        config=TORTOISE_CONFIG,
        generate_schemas=True, # スキーマが存在しない場合はテーブルを作成します
        add_exception_handlers=True,
    )

app = FastAPI()

# アプリ起動中にTortoise ORMを登録
@app.on_event("startup")
async def startup_event():
    await init_db(app)

FastAPIエンドポイント

@app.post("/heroes/", response_model=HeroOut)
async def create_hero(hero_in: HeroIn):
    hero = await Hero.create(**hero_in.model_dump())
    return await HeroOut.from_tortoise_orm(hero)

@app.get("/heroes/", response_model=List[HeroOut])
async def get_heroes(offset: int = 0, limit: int = 100):
    heroes = await Hero.all().offset(offset).limit(limit)
    return [await HeroOut.from_tortoise_orm(hero) for hero in heroes]

@app.get("/heroes/{hero_id}", response_model=HeroOut)
async def get_hero(hero_id: int):
    try:
        hero = await Hero.get(id=hero_id)
        return await HeroOut.from_tortoise_orm(hero)
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Hero not found")

@app.put("/heroes/{hero_id}", response_model=HeroOut)
async def update_hero(hero_id: int, hero_in: HeroIn):
    try:
        hero = await Hero.get(id=hero_id)
        await hero.update_from_dict(hero_in.model_dump(exclude_unset=True))
        await hero.save()
        return await HeroOut.from_tortoise_orm(hero)
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Hero not found")

@app.delete("/heroes/{hero_id}", status_code=204)
async def delete_hero(hero_id: int):
    try:
        hero = await Hero.get(id=hero_id)
        await hero.delete()
        return {"message": "Hero deleted successfully"}
    except DoesNotExist:
        raise HTTPException(status_code=404, detail="Hero not found")

Tortoise ORMを使用する場合：

• register_tortoiseはデータベース接続とスキーマ生成を処理します。
• モデルはmodels.Modelを継承し、属性を定義するためにfieldsを使用します。
• await Hero.create(), await Hero.all(), await Hero.get(), await hero.save(), await hero.delete()はいずれも非同期操作です。
• HeroOut.from_tortoise_orm()は、ORMインスタンスをPydanticモデルに変換するためにtortoise.contrib.pydanticによって提供される便利なメソッドです。

アプリケーションシナリオ

SQLModelとTortoise ORMは、以下のようなシナリオで優れています。

• 高い同時実行性が期待される場合: 大量のユーザーまたはリクエストは、効率的なI/O処理を必要とします。
• マイクロサービスアーキテクチャ: 分離されたサービスは、それぞれのデータベースとの高速でノンブロッキングな通信から恩恵を受けることがよくあります。
• リアルタイムアプリケーション: リアルタイムデータまたは更新を提供するAPIは、低レイテンシのデータベース操作を必要とします。
• モダンPythonバックエンド: FastAPIのようなフレームワークとの統合は、それらの非同期機能を自然に活用します。

SQLModelの強みは、Pydanticとの緊密な統合にあり、データ検証とシリアライゼーションが重要であり、データモデルの単一の信頼できるソースが欲しいプロジェクトに最適です。SQLAlchemyの堅牢な基盤の上に構築されており、高度なクエリ機能を提供します。

Tortoise ORMの強みは、asyncio専用に設計されたシンプルなAPIと、非同期ORMの初心者にとってしばしばより穏やかな学習曲線です。特に、独自のクエリ構文を持つスタンドアロンORMソリューションを好む場合は魅力的です。

結論

SQLModelまたはTortoise ORMのようなORMを使用してFastAPIアプリケーションに非同期データベース操作を統合することは、パフォーマンスが高くスケーラブルなWebサービスを構築するための重要なステップです。どちらのツ​​ールも堅牢な非同期機能を提供し、I/Oボトルネックを排除して、アプリケーションが高負荷下でも応答性を維持することを保証します。SQLModelはPydanticとSQLAlchemyのパワーを備えた統合モデル定義を提供し、Tortoise ORMは簡潔でasyncioネイティブなエクスペリエンスを提供します。どちらを選択するかは、多くの場合、特定のプロジェクト要件、チームの習熟度、およびそれぞれのAPIスタイルへの好みによって決まりますが、どちらもFastAPIアプリケーションのデータベース対話効率を大幅に向上させるでしょう。非同期ORMを採用することで、FastAPIの機能を最大限に活用し、真にノンブロッキングで高性能なバックエンドを提供できます。