在快節(jié)奏的 Web 開(kāi)發(fā)世界中，效能至關(guān)重要。高效的快取機(jī)制可以透過(guò)減少冗餘計(jì)算和資料庫(kù)查詢來(lái)顯著增強(qiáng) API 的回應(yīng)能力。在本文中，我們將探討如何使用 SQLModel 和 Redis 將 py-cachify 程式庫(kù)整合到 FastAPI 應(yīng)用程式中，以實(shí)現(xiàn)快取和並發(fā)控制。

目錄：

• 簡(jiǎn)介
• 項(xiàng)目設(shè)定
• 使用 SQLModel 建立資料庫(kù)模型
• 建置 FastAPI 端點(diǎn)
• 快取端點(diǎn)結(jié)果
• 鎖定更新端點(diǎn)的執(zhí)行
• 運(yùn)行應(yīng)用程式
• 結(jié)論

介紹

快取是一種強(qiáng)大的技術(shù)，透過(guò)儲(chǔ)存昂貴操作的結(jié)果並從快速存取儲(chǔ)存中提供它們來(lái)提高 Web 應(yīng)用程式的效能。透過(guò) py-cachify，我們可以無(wú)縫地將快取添加到 FastAPI 應(yīng)用程式中，並利用 Redis 進(jìn)行儲(chǔ)存。此外，py-cachify 提供並發(fā)控制工具，防止關(guān)鍵操作期間出現(xiàn)競(jìng)爭(zhēng)情況。

在本教學(xué)中，我們將逐步在 FastAPI 應(yīng)用程式中設(shè)定 py-cachify 函式庫(kù)，並使用用於 ORM 的 SQLModel 和用於快取的 Redis。

項(xiàng)目設(shè)定

讓我們從設(shè)定專案環(huán)境開(kāi)始。

先決條件

• Python 3.12
• 詩(shī)歌（您可以使用任何您喜歡的套件管理器）
• 本地運(yùn)作或遠(yuǎn)端存取的Redis伺服器

安裝依賴項(xiàng)

透過(guò)詩(shī)歌開(kāi)始一個(gè)新專案：

# 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：結(jié)合 SQLAlchemy 和 Pydantic 進(jìn)行 ORM 和資料驗(yàn)證。
• Redis：用於與 Redis 互動(dòng)的 Python 用戶端。
• py-cachify：快取和鎖定實(shí)用程式。

初始化 py-cachify

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

# 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 建立資料庫(kù)模型

我們將建立一個(gè)簡(jiǎn)單的使用者模型來(lái)與我們的資料庫(kù)互動(dòng)。

# 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è)定資料庫(kù)引擎並在生命週期函數(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)單起見(jiàn)，我們使用 SQLite，但您可以使用 SQLAlchemy 支援的任何資料庫(kù)。

建構(gòu) FastAPI 端點(diǎn)

讓我們建立端點(diǎn)來(lái)與我們的使用者模型互動(dò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

快取端點(diǎn)結(jié)果

現(xiàn)在，讓我們快取 read_user 端點(diǎn)的結(jié)果，以避免不必要的資料庫(kù)查詢。

端點(diǎn)程式碼將如下所示：

# 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 指定一個(gè)唯一的鍵。
• 將 TTL（存活時(shí)間）設(shè)定為 5 分鐘（300 秒）。
• 5 分鐘內(nèi)使用相同 user_id 呼叫此端點(diǎn)將傳回快取的結(jié)果。

更新時(shí)重置快取

當(dāng)使用者的資料更新時(shí)，我們需要重置快取以確?？蛻舳耸盏阶钚碌馁Y訊。為了實(shí)現(xiàn)這一點(diǎn)，讓我們修改 update_user 端點(diǎn)。

# 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

透過(guò)呼叫 read_user.reset(user_id=user_id)，我們：

• 清除特定user_id的快取資料。
• 確保後續(xù) GET 請(qǐng)求從資料庫(kù)取得新資料。

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

.reset 方法使用快取裝飾器中定義的相同金鑰產(chǎn)生邏輯來(lái)識(shí)別要使哪個(gè)快取條目無(wú)效。例如，如果您的快取鍵模式是 user-{user_id}，則呼叫await read_user.reset(user_id=123) 將專門定位並刪除 user_id=123 的快取條目。

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

為了防止更新期間的競(jìng)爭(zhēng)條件，我們將使用一次裝飾器來(lái)鎖定更新端點(diǎn)的執(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鎖定功能。
• 如果另一個(gè)請(qǐng)求嘗試同時(shí)更新同一用戶，它將立即傳回帶有 226 IM 已使用 狀態(tài)代碼的回應(yīng)。
• 這可以防止同時(shí)更新導(dǎo)致資料不一致。

（可選）您可以設(shè)定 @once 以引發(fā)異?；蛟谝讶〉面i定的情況下傳回特定值。

運(yùn)行應(yīng)用程式

現(xiàn)在是時(shí)候運(yùn)行和測(cè)試我們的應(yīng)用程式了！

1) 啟動(dòng) Redis 伺服器：

確保您的 Redis 伺服器在本地運(yùn)行或可遠(yuǎn)端存取。您可以使用 Docker 啟動(dòng)本機(jī) 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) 執(zhí)行 FastAPI 應(yīng)用程式：

一切設(shè)定完畢後，您可以使用 Poetry 啟動(dòng) FastAPI 應(yīng)用程式。導(dǎo)航到專案的根目錄並執(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) 測(cè)試和使用快取和鎖定：

快?。?/strong> 在 read_user 函數(shù)中加入延遲（例如，使用 asyncio.sleep）以模擬長(zhǎng)時(shí)間運(yùn)行的計(jì)算。觀察結(jié)果緩存後反應(yīng)時(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

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