FastAPI の効率を最大化: py-cachify によるキャッシュとロックの超高速実装

ペースの速い Web 開発の世界では、パフォーマンスが最も重要です。効率的なキャッシュ メカニズムにより、冗長な計算とデータベース クエリが削減され、API の応答性が大幅に向上します。この記事では、SQLModel と Redis を使用して py-cachify ライブラリを FastAPI アプリケーションに統合し、キャッシュと同時実行制御を実装する方法を検討します。

目次：

• はじめに
• プロジェクトのセットアップ
• SQLModel を使用したデータベース モデルの作成
• FastAPI エンドポイントの構築
• エンドポイント結果のキャッシュ
• アップデートエンドポイントの実行をロック
• アプリケーションの実行
• 結論

導入

キャッシュは、高コストの操作の結果を保存し、それらを迅速にアクセスできるストレージから提供することにより、Web アプリケーションのパフォーマンスを向上させる強力な技術です。 py-cachify を使用すると、ストレージに Redis を利用して、FastAPI アプリケーションにキャッシュをシームレスに追加できます。さらに、py-cachify は同時実行制御のためのツールを提供し、重要な操作中の競合状態を防ぎます。

このチュートリアルでは、ORM 用の SQLModel とキャッシュ用の Redis を使用して、FastAPI アプリケーションで py-cachify ライブラリをセットアップする手順を説明します。

プロジェクトのセットアップ

プロジェクト環境をセットアップすることから始めましょう。

前提条件

• Python 3.12
• 詩 (好きなパッケージマネージャーを使用できます)
• ローカルで実行されている、またはリモートからアクセス可能な Redis サーバー

依存関係のインストール

詩を通じて新しいプロジェクトを開始します:

# create new project
poetry new --name app py-cachify-fastapi-demo

# enter the directory
cd py-cachify-fastapi-demo

# point poetry to use python3.12
poetry env use python3.12

# add dependencies
poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify

• FastAPI: API を構築するための Web フレームワーク。
• SQLModel aiosqlite: ORM とデータ検証のために SQLAlchemy と Pydantic を組み合わせます。
• Redis: Redis と対話するための Python クライアント。
• py-cachify: キャッシュおよびロック ユーティリティ。

py-cachify の初期化

py-cachify を使用する前に、Redis クライアントでそれを初期化する必要があります。これは、FastAPI のライフスパン パラメーターを使用して行います。

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        # Replace with your redis url if it differs
        async_client=from_url('redis://localhost:6379/0'),
    )
    yield


app = FastAPI(lifespan=lifespan)

寿命内で、私たちは次のことを行います:

• 非同期 Redis クライアントを作成します。
• このクライアントで py-cachify を初期化します。

SQLModel を使用したデータベース モデルの作成

データベースと対話するための単純な User モデルを作成します。

# app/db.py
from sqlmodel import Field, SQLModel


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

データベース エンジンをセットアップし、ライフスパン関数でテーブルを作成します。

# app/db.py

# Adjust imports
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine


# Add the following at the end of the file
sqlite_file_name = 'database.db'
sqlite_url = f'sqlite+aiosqlite:///{sqlite_file_name}'
engine = create_async_engine(sqlite_url, echo=True)
session_maker = async_sessionmaker(engine)


# app/main.py
# Adjust imports and lifespan function
from sqlmodel import SQLModel

from .db import engine


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        async_client=from_url('redis://localhost:6379/0'),
    )
    # Create SQL Model tables
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    yield

注: 簡単にするために SQLite を使用していますが、SQLAlchemy でサポートされている任意のデータベースを使用できます。

FastAPI エンドポイントの構築

User モデルと対話するためのエンドポイントを作成しましょう。

# create new project
poetry new --name app py-cachify-fastapi-demo

# enter the directory
cd py-cachify-fastapi-demo

# point poetry to use python3.12
poetry env use python3.12

# add dependencies
poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify

エンドポイント結果のキャッシュ

次に、不要なデータベース クエリを回避するために、read_user エンドポイントの結果をキャッシュしましょう。

エンドポイントのコードは次のようになります:

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        # Replace with your redis url if it differs
        async_client=from_url('redis://localhost:6379/0'),
    )
    yield


app = FastAPI(lifespan=lifespan)

@cached デコレータを使用する場合:

• user_id を使用して一意のキーを指定します。
• TTL (存続時間) を 5 分 (300 秒) に設定します。
• 同じ user_id を使用してこのエンドポイントを 5 分以内に呼び出すと、キャッシュされた結果が返されます。

更新時のキャッシュのリセット

ユーザーのデータが更新されると、クライアントが最新の情報を確実に受信できるようにキャッシュをリセットする必要があります。これを実現するには、update_user エンドポイントを変更しましょう。

# app/db.py
from sqlmodel import Field, SQLModel


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

read_user.reset(user_id=user_id) を呼び出すことで、次のことを行います。

• 特定の user_id のキャッシュ データをクリアします。
• 後続の GET リクエストがデータベースから新しいデータをフェッチするようにします。

その下では、キャッシュされたデコレータが関数を動的にラップし、.reset メソッドを追加します。このメソッドは関数のシグネチャと型を模倣します。この方法では、元の関数に応じて同期または非同期のいずれかになり、同じ引数を受け入れます。

.reset メソッドは、キャッシュされたデコレータで定義されているのと同じキー生成ロジックを使用して、どのキャッシュされたエントリを無効にするかを識別します。たとえば、キャッシュ キー パターンが user-{user_id} の場合、await read_user.reset(user_id=123) を呼び出すと、具体的に user_id=123 のキャッシュ エントリがターゲットになり、削除されます。

更新エンドポイントの実行のロック

更新中の競合状態を防ぐために、once デコレータを使用して更新エンドポイントの実行をロックします。

# app/db.py

# Adjust imports
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine


# Add the following at the end of the file
sqlite_file_name = 'database.db'
sqlite_url = f'sqlite+aiosqlite:///{sqlite_file_name}'
engine = create_async_engine(sqlite_url, echo=True)
session_maker = async_sessionmaker(engine)


# app/main.py
# Adjust imports and lifespan function
from sqlmodel import SQLModel

from .db import engine


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        async_client=from_url('redis://localhost:6379/0'),
    )
    # Create SQL Model tables
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    yield

1 回の場合:

• user_id に基づいて機能をロックします。
• 別のリクエストが同じユーザーを同時に更新しようとすると、すぐに 226 IM Used ステータス コードを含むレスポンスが返されます。
• これにより、データの不整合を引き起こす可能性のある同時更新が防止されます。

オプションで、ロックがすでに取得されている場合に例外を発生させるか、特定の値を返すように @once を構成できます。

アプリケーションの実行

今度はアプリを実行してテストします!

1) Redis サーバーを起動します:

Redis サーバーがローカルで実行されているか、リモートからアクセスできることを確認してください。 Docker を使用してローカル Redis サーバーを起動できます:

# app/main.py

# Adjust imports
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession

from .db import User, engine, session_maker


# Database session dependency
async def get_session():
    async with session_maker() as session:
        yield session


app = FastAPI(lifespan=lifespan)


@app.post('/users/')
async def create_user(user: User, session: AsyncSession = Depends(get_session)) -> User:
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user


@app.get('/users/{user_id}')
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)) -> User | None:
    return await session.get(User, user_id)


@app.put('/users/{user_id}')
async def update_user(user_id: int, new_user: User, session: AsyncSession = Depends(get_session)) -> User | None:
    user = await session.get(User, user_id)
    if not user:
        return None

    user.name = new_user.name
    user.email = new_user.email

    session.add(user)
    await session.commit()
    await session.refresh(user)

    return user

2) FastAPI アプリケーションを実行します:

すべてをセットアップしたら、Poetry を使用して FastAPI アプリケーションを起動できます。プロジェクトのルート ディレクトリに移動し、次のコマンドを実行します:

# app/main.py

# Add the import
from py_cachify import cached


@app.get('/users/{user_id}')
@cached('read_user-{user_id}', ttl=300)  # New decorator
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)) -> User | None:
    return await session.get(User, user_id)

3) キャッシュとロックのテストとプレイ:

Caching: 長時間実行される計算をシミュレートするために、read_user 関数に遅延を追加します (例: asyncio.sleep を使用)。結果がキャッシュされると応答時間が大幅に改善される様子を観察してください。

例:

# create new project
poetry new --name app py-cachify-fastapi-demo

# enter the directory
cd py-cachify-fastapi-demo

# point poetry to use python3.12
poetry env use python3.12

# add dependencies
poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify

同時実行性とロック: 同様に、update_user 関数に遅延を導入して、同時更新試行が行われたときのロックの動作を観察します。

例:

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        # Replace with your redis url if it differs
        async_client=from_url('redis://localhost:6379/0'),
    )
    yield


app = FastAPI(lifespan=lifespan)

これらの遅延は、動作中のキャッシュおよびロック メカニズムの有効性を確認するのに役立ちます。キャッシュによって後続の読み取りが高速化され、同じリソースへの同時書き込みはロックによって効果的に管理される必要があるためです。

これで、Postman などのツールを使用するか、http://127.0.0.1:8000/docs にアクセスして (アプリの実行中!) エンドポイントをテストし、実際のパフォーマンスの向上と同時実行制御を観察できます。

強化された FastAPI アプリを楽しんで実験してください!

結論

py-cachify を FastAPI アプリケーションに統合することにより、API のパフォーマンスと信頼性の両方を強化する多くの利点が解放されました。

主な強みのいくつかをまとめてみましょう:

• パフォーマンスの向上: 繰り返しの関数呼び出しをキャッシュすることで、冗長な計算とデータベース ヒットが削減され、応答時間が大幅に改善されます。
• 同時実行制御: 組み込みのロック メカニズムにより、py-cachify は競合状態を防止し、同時アクセスが多いアプリケーションにとって重要なデータの一貫性を確保します。
• 柔軟性: 同期操作でも非同期操作でも、py-cachify はシームレスに適応し、最新の Web アプリケーションにとって多用途の選択肢となります。
• 使いやすさ: このライブラリは、FastAPI などの一般的な Python フレームワークとスムーズに統合されているため、最小限の手間で使い始めることができます。
• 完全な型アノテーション: py-cachify は完全に型アノテーションが付けられており、最小限の労力でより優れた、より保守しやすいコードを作成するのに役立ちます。
• 最小限のセットアップ: このチュートリアルで示したように、py-cachify の機能を最大限に活用するには、既存のセットアップに数行追加するだけで済みます。

さらに詳しく知りたい方は、py-cachify の GitHub リポジトリ と 公式ドキュメント でさらに詳細なガイダンス、チュートリアル、例を確認してください。

このチュートリアルの完全なコードには、GitHub こちらからアクセスできます。自由にリポジトリのクローンを作成し、プロジェクトのニーズに合わせて実装を試してみてください。

py-cachify が有益だと思われる場合は、GitHub でスターを付けてプロジェクトをサポートすることを検討してください。あなたのサポートは、さらなる改善と新機能の推進に役立ちます。

コーディングを楽しんでください!

以上がFastAPI の効率を最大化: py-cachify によるキャッシュとロックの超高速実装の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。