国产av日韩一区二区三区精品,成人性爱视频在线观看,国产,欧美,日韩,一区,www.成色av久久成人,2222eeee成人天堂

首頁 後端開發(fā) Python教學 最大限度地提高 FastAPI 效率:使用 py-cachify 極快實現(xiàn)快取和鎖定

最大限度地提高 FastAPI 效率:使用 py-cachify 極快實現(xiàn)快取和鎖定

Dec 05, 2024 am 05:47 AM

Maximize Your FastAPI Efficiency: Blazingly Fast Implementation of Caching and Locking with py-cachify

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

目錄:

  • 簡介
  • 項目設定
  • 使用 SQLModel 建立資料庫模型
  • 建置 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 和資料驗證。
  • 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),我們:

  • 建立非同步 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

設定資料庫引擎並在生命週期函數(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

注意:為了簡單起見,我們使用 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

快取端點結果

現(xiàn)在,讓我們快取 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 分鐘內(nèi)使用相同 user_id 呼叫此端點將傳回快取的結果。

更新時重置快取

當使用者的資料更新時,我們需要重置快取以確保客戶端收到最新的資訊。為了實現(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的快取資料。
  • 確保後續(xù) GET 請求從資料庫取得新資料。

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

.reset 方法使用快取裝飾器中定義的相同金鑰產(chǎn)生邏輯來識別要使哪個快取條目無效。例如,如果您的快取鍵模式是 user-{user_id},則呼叫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)代碼的回應。
  • 這可以防止同時更新導致資料不一致。

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

運行應用程式

現(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) 執(zhí)行 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ù)呼叫可以減少冗餘計算和資料庫命中,從而大大縮短回應時間。
  • 並發(fā)控制:透過內(nèi)建鎖定機制,py-cachify 可以防止競爭條件並確保資料一致性 - 對於高並發(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)其他相關文章!

本網(wǎng)站聲明
本文內(nèi)容由網(wǎng)友自願投稿,版權歸原作者所有。本站不承擔相應的法律責任。如發(fā)現(xiàn)涉嫌抄襲或侵權的內(nèi)容,請聯(lián)絡admin@php.cn

熱AI工具

Undress AI Tool

Undress AI Tool

免費脫衣圖片

Undresser.AI Undress

Undresser.AI Undress

人工智慧驅(qū)動的應用程序,用於創(chuàng)建逼真的裸體照片

AI Clothes Remover

AI Clothes Remover

用於從照片中去除衣服的線上人工智慧工具。

Clothoff.io

Clothoff.io

AI脫衣器

Video Face Swap

Video Face Swap

使用我們完全免費的人工智慧換臉工具,輕鬆在任何影片中換臉!

熱工具

記事本++7.3.1

記事本++7.3.1

好用且免費的程式碼編輯器

SublimeText3漢化版

SublimeText3漢化版

中文版,非常好用

禪工作室 13.0.1

禪工作室 13.0.1

強大的PHP整合開發(fā)環(huán)境

Dreamweaver CS6

Dreamweaver CS6

視覺化網(wǎng)頁開發(fā)工具

SublimeText3 Mac版

SublimeText3 Mac版

神級程式碼編輯軟體(SublimeText3)

Python的UNITDEST或PYTEST框架如何促進自動測試? Python的UNITDEST或PYTEST框架如何促進自動測試? Jun 19, 2025 am 01:10 AM

Python的unittest和pytest是兩種廣泛使用的測試框架,它們都簡化了自動化測試的編寫、組織和運行。 1.二者均支持自動發(fā)現(xiàn)測試用例並提供清晰的測試結構:unittest通過繼承TestCase類並以test\_開頭的方法定義測試;pytest則更為簡潔,只需以test\_開頭的函數(shù)即可。 2.它們都內(nèi)置斷言支持:unittest提供assertEqual、assertTrue等方法,而pytest使用增強版的assert語句,能自動顯示失敗詳情。 3.均具備處理測試準備與清理的機制:un

如何將Python用於數(shù)據(jù)分析和與Numpy和Pandas等文庫進行操作? 如何將Python用於數(shù)據(jù)分析和與Numpy和Pandas等文庫進行操作? Jun 19, 2025 am 01:04 AM

pythonisidealfordataanalysisionduetonumpyandpandas.1)numpyExccelSatnumericalComputationswithFast,多dimensionalArraysAndRaysAndOrsAndOrsAndOffectorizedOperationsLikenp.sqrt()

什麼是動態(tài)編程技術,如何在Python中使用它們? 什麼是動態(tài)編程技術,如何在Python中使用它們? Jun 20, 2025 am 12:57 AM

動態(tài)規(guī)劃(DP)通過將復雜問題分解為更簡單的子問題並存儲其結果以避免重複計算,來優(yōu)化求解過程。主要方法有兩種:1.自頂向下(記憶化):遞歸分解問題,使用緩存存儲中間結果;2.自底向上(表格化):從基礎情況開始迭代構建解決方案。適用於需要最大/最小值、最優(yōu)解或存在重疊子問題的場景,如斐波那契數(shù)列、背包問題等。在Python中,可通過裝飾器或數(shù)組實現(xiàn),並應注意識別遞推關係、定義基準情況及優(yōu)化空間複雜度。

如何使用__ITER__和__NEXT __在Python中實現(xiàn)自定義迭代器? 如何使用__ITER__和__NEXT __在Python中實現(xiàn)自定義迭代器? Jun 19, 2025 am 01:12 AM

要實現(xiàn)自定義迭代器,需在類中定義__iter__和__next__方法。 ①__iter__方法返回迭代器對象自身,通常為self,以兼容for循環(huán)等迭代環(huán)境;②__next__方法控制每次迭代的值,返回序列中的下一個元素,當無更多項時應拋出StopIteration異常;③需正確跟蹤狀態(tài)並設置終止條件,避免無限循環(huán);④可封裝複雜邏輯如文件行過濾,同時注意資源清理與內(nèi)存管理;⑤對簡單邏輯可考慮使用生成器函數(shù)yield替代,但需結合具體場景選擇合適方式。

Python編程語言及其生態(tài)系統(tǒng)的新興趨勢或未來方向是什麼? Python編程語言及其生態(tài)系統(tǒng)的新興趨勢或未來方向是什麼? Jun 19, 2025 am 01:09 AM

Python的未來趨勢包括性能優(yōu)化、更強的類型提示、替代運行時的興起及AI/ML領域的持續(xù)增長。首先,CPython持續(xù)優(yōu)化,通過更快的啟動時間、函數(shù)調(diào)用優(yōu)化及擬議中的整數(shù)操作改進提升性能;其次,類型提示深度集成至語言與工具鏈,增強代碼安全性與開發(fā)體驗;第三,PyScript、Nuitka等替代運行時提供新功能與性能優(yōu)勢;最後,AI與數(shù)據(jù)科學領域持續(xù)擴張,新興庫推動更高效的開發(fā)與集成。這些趨勢表明Python正不斷適應技術變化,保持其領先地位。

如何使用插座在Python中執(zhí)行網(wǎng)絡編程? 如何使用插座在Python中執(zhí)行網(wǎng)絡編程? Jun 20, 2025 am 12:56 AM

Python的socket模塊是網(wǎng)絡編程的基礎,提供低級網(wǎng)絡通信功能,適用於構建客戶端和服務器應用。要設置基本TCP服務器,需使用socket.socket()創(chuàng)建對象,綁定地址和端口,調(diào)用.listen()監(jiān)聽連接,並通過.accept()接受客戶端連接。構建TCP客戶端需創(chuàng)建socket對像後調(diào)用.connect()連接服務器,再使用.sendall()發(fā)送數(shù)據(jù)和??.recv()接收響應。處理多個客戶端可通過1.線程:每次連接啟動新線程;2.異步I/O:如asyncio庫實現(xiàn)無阻塞通信。注意事

如何在Python中切片列表? 如何在Python中切片列表? Jun 20, 2025 am 12:51 AM

Python列表切片的核心答案是掌握[start:end:step]語法並理解其行為。 1.列表切片的基本格式為list[start:end:step],其中start是起始索引(包含)、end是結束索引(不包含)、step是步長;2.省略start默認從0開始,省略end默認到末尾,省略step默認為1;3.獲取前n項用my_list[:n],獲取後n項用my_list[-n:];4.使用step可跳過元素,如my_list[::2]取偶數(shù)位,負step值可反轉(zhuǎn)列表;5.常見誤區(qū)包括end索引不

如何使用DateTime模塊在Python中使用日期和時間? 如何使用DateTime模塊在Python中使用日期和時間? Jun 20, 2025 am 12:58 AM

Python的datetime模塊能滿足基本的日期和時間處理需求。 1.可通過datetime.now()獲取當前日期和時間,也可分別提取.date()和.time()。 2.能手動創(chuàng)建特定日期時間對象,如datetime(year=2025,month=12,day=25,hour=18,minute=30)。 3.使用.strftime()按格式輸出字符串,常見代碼包括%Y、%m、%d、%H、%M、%S;用strptime()將字符串解析為datetime對象。 4.利用timedelta進行日期運

See all articles