最大限度地提高 FastAPI 效率：使用 py-cachify 極快實現快取和鎖定

在快節奏的 Web 開發世界中，效能至關重要。高效的快取機制可以透過減少冗餘計算和資料庫查詢來顯著增強 API 的回應能力。在本文中，我們將探討如何使用 SQLModel 和 Redis 將 py-cachify 程式庫整合到 FastAPI 應用程式中，以實現快取和並發控制。

目錄：

• 簡介
• 項目設定
• 使用 SQLModel 建立資料庫模型
• 建置 FastAPI 端點
• 快取端點結果
• 鎖定更新端點的執行
• 運行應用程式
• 結論

介紹

快取是一種強大的技術，透過儲存昂貴操作的結果並從快速存取儲存中提供它們來提高 Web 應用程式的效能。透過 py-cachify，我們可以無縫地將快取添加到 FastAPI 應用程式中，並利用 Redis 進行儲存。此外，py-cachify 提供並發控制工具，防止關鍵操作期間出現競爭情況。

在本教學中，我們將逐步在 FastAPI 應用程式中設定 py-cachify 函式庫，並使用用於 ORM 的 SQLModel 和用於快取的 Redis。

項目設定

讓我們從設定專案環境開始。

先決條件

• 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：結合 SQLAlchemy 和 Pydantic 進行 ORM 和資料驗證。
• 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 建立資料庫模型

我們將建立一個簡單的使用者模型來與我們的資料庫互動。

# 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 端點

讓我們建立端點來與我們的使用者模型互動。

# 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 秒）。
• 5 分鐘內使用相同 user_id 呼叫此端點將傳回快取的結果。

更新時重置快取

當使用者的資料更新時，我們需要重置快取以確保客戶端收到最新的資訊。為了實現這一點，讓我們修改 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 的快取條目。

鎖定更新端點的執行

為了防止更新期間的競爭條件，我們將使用一次裝飾器來鎖定更新端點的執行。

# 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

曾經：

• 我們根據user_id鎖定功能。
• 如果另一個請求嘗試同時更新同一用戶，它將立即傳回帶有 226 IM 已使用 狀態代碼的回應。
• 這可以防止同時更新導致資料不一致。

（可選）您可以設定 @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) 測試和使用快取和鎖定：

快取： 在 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中文網其他相關文章！