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

目次：

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

導(dǎo)入

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

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

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

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

前提條件

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

依存関係のインストール

詩(shī)を通じて新しいプロジェクトを開(kāi)始します:

# 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 を構(gòu)築するための 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)

壽命內(nèi)で、私たちは次のことを行います:

• 非同期 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

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

# 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

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

FastAPI エンドポイントの構(gòu)築

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

エンドポイント結(jié)果のキャッシュ

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

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

# 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 デコレータを使用する場(chǎng)合:

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

更新時(shí)のキャッシュのリセット

ユーザーのデータが更新されると、クライアントが最新の情報(bào)を確実に受信できるようにキャッシュをリセットする必要があります。これを?qū)g現(xiàn)するには、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 リクエストがデータベースから新しいデータをフェッチするようにします。

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

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

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

更新中の競(jìng)合狀態(tài)を防ぐために、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 回の場(chǎng)合:

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

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

アプリケーションの実行

今度はアプリを?qū)g行してテストします!

1) Redis サーバーを起動(dòng)します:

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

# 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 アプリケーションを?qū)g行します:

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

# 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: 長(zhǎng)時(shí)間実行される計(jì)算をシミュレートするために、read_user 関數(shù)に遅延を追加します (例: asyncio.sleep を使用)。結(jié)果がキャッシュされると応答時(shí)間が大幅に改善される様子を観察してください。

例:

# 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

同時(shí)実行性とロック: 同様に、update_user 関數(shù)に遅延を?qū)毪筏?、同時(shí)更新試行が行われたときのロックの動(dòng)作を観察します。

例:

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

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

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

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

結(jié)論

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

主な強(qiáng)みのいくつかをまとめてみましょう:

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

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

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

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

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

以上がF(xiàn)astAPI の効率を最大化: py-cachify によるキャッシュとロックの超高速実裝の詳細(xì)內(nèi)容です。詳細(xì)については、PHP 中國(guó)語(yǔ) Web サイトの他の関連記事を參照してください。