在快節(jié)奏的 Web 開發(fā)世界中，性能至關重要。高效的緩存機制可以通過減少冗余計算和數(shù)據(jù)庫查詢來顯著增強 API 的響應能力。在本文中，我們將探討如何使用 SQLModel 和 Redis 將 py-cachify 庫集成到 FastAPI 應用程序中，以實現(xiàn)緩存和并發(fā)控制。

目錄：

• 簡介
• 項目設置
• 使用 SQLModel 創(chuàng)建數(shù)據(jù)庫模型
• 構建 FastAPI 端點
• 緩存端點結果
• 鎖定更新端點的執(zhí)行
• 運行應用程序
• 結論

介紹

緩存是一種強大的技術，通過存儲昂貴操作的結果并從快速訪問存儲中提供它們來提高 Web 應用程序的性能。借助 py-cachify，我們可以無縫地將緩存添加到 FastAPI 應用程序中，并利用 Redis 進行存儲。此外，py-cachify 提供并發(fā)控制工具，防止關鍵操作期間出現(xiàn)競爭情況。

在本教程中，我們將逐步在 FastAPI 應用程序中設置 py-cachify 庫，并使用用于 ORM 的 SQLModel 和用于緩存的 Redis。

項目設置

讓我們從設置項目環(huán)境開始。

先決條件

• 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 和數(shù)據(jù)驗證。
• Redis：用于與 Redis 交互的 Python 客戶端。
• py-cachify：緩存和鎖定實用程序。

初始化 py-cachify

在使用 py-cachify 之前，我們需要使用 Redis 客戶端對其進行初始化。我們將使用 FastAPI 的壽命參數(shù)來完成此操作。

# 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)，我們：

• 創(chuàng)建異步 Redis 客戶端。
• 使用此客戶端初始化 py-cachify。

使用 SQLModel 創(chuàng)建數(shù)據(jù)庫模型

我們將創(chuàng)建一個簡單的用戶模型來與我們的數(shù)據(jù)庫交互。

# 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ù)據(jù)庫引擎并在生命周期函數(shù)中創(chuàng)建表：

# 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 支持的任何數(shù)據(jù)庫。

構建 FastAPI 端點

讓我們創(chuàng)建端點來與我們的用戶模型交互。

# 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

緩存端點結果

現(xiàn)在，讓我們緩存 read_user 端點的結果，以避免不必要的數(shù)據(jù)庫查詢。

端點代碼將如下所示：

# 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 分鐘內(nèi)使用相同 user_id 調(diào)用此端點將返回緩存的結果。

更新時重置緩存

當用戶的數(shù)據(jù)更新時，我們需要重置緩存以確?？蛻舳耸盏阶钚碌男畔ⅰ榱藢崿F(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

通過調(diào)用 read_user.reset(user_id=user_id)，我們：

• 清除特定user_id的緩存數(shù)據(jù)。
• 確保后續(xù) GET 請求從數(shù)據(jù)庫獲取新數(shù)據(jù)。

在下面，緩存的裝飾器動態(tài)包裝您的函數(shù)，添加 .reset 方法。此方法模仿函數(shù)的簽名和類型，這樣根據(jù)原始函數(shù)，它將是同步或異步的，并且將接受相同的參數(shù)。

.reset 方法使用緩存裝飾器中定義的相同密鑰生成邏輯來識別要使哪個緩存條目無效。例如，如果您的緩存鍵模式是 user-{user_id}，則調(diào)用await read_user.reset(user_id=123) 將專門定位并刪除 user_id=123 的緩存條目。

鎖定更新端點的執(zhí)行

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

# 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

曾經(jīng)：

• 我們根據(jù)user_id鎖定功能。
• 如果另一個請求嘗試同時更新同一用戶，它將立即返回帶有 226 IM 已使用 狀態(tài)代碼的響應。
• 這可以防止同時更新導致數(shù)據(jù)不一致。

（可選）您可以配置 @once 以引發(fā)異?；蛟谝勋@取鎖的情況下返回特定值。

運行應用程序

現(xiàn)在是時候運行和測試我們的應用程序了！

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 應用程序。導航到項目的根目錄并執(zhí)行以下命令：

# 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 函數(shù)中添加延遲（例如，使用 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

并發(fā)和鎖定：同樣，在 update_user 函數(shù)中引入延遲，以觀察并發(fā)更新嘗試時鎖的行為。

示例：

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

這些延遲可以幫助您了解緩存和鎖定機制的有效性，因為由于緩存，后續(xù)讀取應該更快，并且應該通過鎖定有效管理對同一資源的并發(fā)寫入。

現(xiàn)在，您可以使用 Postman 等工具或訪問 http://127.0.0.1:8000/docs（當應用程序運行時?。﹣頊y試端點，并觀察性能改進和并發(fā)控制的實際情況。

享受使用增強型 FastAPI 應用程序進行實驗的樂趣！

結論

通過將 py-cachify 集成到我們的 FastAPI 應用程序中，我們釋放了眾多優(yōu)勢，增強了 API 的性能和可靠性。

讓我們回顧一下一些關鍵優(yōu)勢：

• 增強的性能：緩存重復的函數(shù)調(diào)用可以減少冗余計算和數(shù)據(jù)庫命中，從而大大縮短響應時間。
• 并發(fā)控制：通過內(nèi)置鎖定機制，py-cachify 可以防止競爭條件并確保數(shù)據(jù)一致性 - 對于高并發(fā)訪問的應用程序至關重要。
• 靈活性：無論您使用同步還是異步操作，py-cachify 都能無縫適應，使其成為現(xiàn)代 Web 應用程序的多功能選擇。
• 易于使用：該庫與 FastAPI 等流行的 Python 框架順利集成，讓您可以輕松上手。
• 完整類型注釋： py-cachify 具有完全類型注釋，有助于以最小的努力編寫更好、更易于維護的代碼。
• 最小設置： 如本教程所示，添加 py-cachify 只需要在現(xiàn)有設置之上添加幾行即可充分利用其功能。

對于那些渴望進一步探索的人，請查看py-cachify 的 GitHub 存儲庫和官方文檔以獲取更深入的指導、教程和示例。

您可以在 GitHub 此處訪問本教程的完整代碼。請隨意克隆存儲庫并嘗試實現(xiàn)以滿足您項目的需求。

如果您發(fā)現(xiàn) py-cachify 有益，請考慮在 GitHub 上給它一顆星來支持該項目！您的支持有助于推動進一步的改進和新功能。

編碼愉快！

以上是最大限度地提高 FastAPI 效率：使用 py-cachify 極快地實現(xiàn)緩存和鎖定的詳細內(nèi)容。更多信息請關注PHP中文網(wǎng)其他相關文章！