透過 FastAPI 中的非同步程式設計增強您的 API 效能-Python教學-PHP中文網

準備好將您的 API 遊戲提升到新的水平了嗎？

FastAPI 設計讓您的 API 更快、更能回應靈敏，並且能夠像專業人士一樣處理重負載。

在本文中，我們將向您展示如何利用 FastAPI 中的非同步程式設計來建立高效能 API。最後，您將具備實現非同步端點並有效測試它們的知識。

你將學到什麼

這是您將要掌握的內容：

非同步程式設計的基礎知識及其重要性。
如何建置FastAPI環境進行非同步開發。
使用真實範例編寫和測試非同步端點。
使用非同步程式庫進行 HTTP 請求、檔案處理和後台任務。

為什麼在 FastAPI 中使用非同步程式設計？

它是什麼？

非同步程式設計使任務能夠並發運作。這對於網路請求、資料庫查詢或文件操作等通常需要等待回應的任務特別有用。

為什麼它很重要？

在傳統的同步程式設計中，任務會依序運行，導致處理多個請求時出現延遲。透過非同步編程，您可以同時為多個使用者提供服務，最大限度地提高資源利用率並確保更好的使用者體驗。

設定您的 FastAPI 環境

現在，讓我們捲起袖子，創造一些令人驚嘆的東西！

首先，安裝所需的庫：

pip install "fastapi[standard]" httpx aiofiles pytest

登入後複製

守則

以下是一個完整的範例，示範了 FastAPI 中的非同步程式設計。程式碼的每個部分都有獨特的用途，我們將逐步解釋它們。

from fastapi import FastAPI, BackgroundTasks
import httpx
import aiofiles
import pytest
from fastapi.testclient import TestClient

app = FastAPI()

# API Endpoints
@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

# Testing Code
client = TestClient(app)

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

def test_read_file():
    # Create an example file for testing
    with open("example.txt", "w") as file:
        file.write("This is a test file content")

    response = client.get("/read-file")
    assert response.status_code == 200
    assert response.json() == {"content": "This is a test file content"}

def test_schedule_email():
    response = client.post("/send-email/?email=fogigav197@rabitex.com")
    assert response.status_code == 200
    assert response.json() == {"message": "Email scheduled"}

@pytest.mark.asyncio
async def test_call_external_api():
    async with httpx.AsyncClient() as async_client:
        response = await async_client.get("https://jsonplaceholder.typicode.com/posts/1")
    assert response.status_code == 200
    assert "id" in response.json()

登入後複製

分解它

API端點

1.簡單非同步端點

@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

登入後複製

此端點顯示如何使用 async def 定義基本非同步路由。它透過 ID 檢索項目。

2.呼叫外部API

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

登入後複製

這示範如何使用 httpx 發出非阻塞 HTTP 請求。在等待外部 API 回應時，您的應用程式可以處理其他請求。

3.非同步檔案讀取

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

登入後複製

這會非同步讀取文件，確保文件操作不會阻塞應用程式。

4.後台任務執行

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

登入後複製

此端點安排後台任務發送電子郵件，允許主執行緒處理其他請求。

測試程式碼

1.測試基本端點

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

登入後複製

這將驗證 /item/{item_id} 端點傳回預期的資料。

2.測試檔讀取

pip install "fastapi[standard]" httpx aiofiles pytest

登入後複製

這將建立一個測試檔案並檢查 /read-file 端點是否已正確讀取並傳回其內容。

3.測試後台任務

from fastapi import FastAPI, BackgroundTasks
import httpx
import aiofiles
import pytest
from fastapi.testclient import TestClient

app = FastAPI()

# API Endpoints
@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

@app.get("/external-api")
async def call_external_api():
    async with httpx.AsyncClient() as client:
        response = await client.get("https://jsonplaceholder.typicode.com/posts/1")
    return response.json()

@app.get("/read-file")
async def read_file():
    async with aiofiles.open("example.txt", mode="r") as file:
        content = await file.read()
    return {"content": content}

def send_email(email: str, message: str):
    print(f"Sending email to {email} with message: {message}")

@app.post("/send-email/")
async def schedule_email(background_tasks: BackgroundTasks, email: str):
    background_tasks.add_task(send_email, email, "Welcome!")
    return {"message": "Email scheduled"}

# Testing Code
client = TestClient(app)

def test_read_item():
    response = client.get("/item/1")
    assert response.status_code == 200
    assert response.json() == {"item_id": 1}

def test_read_file():
    # Create an example file for testing
    with open("example.txt", "w") as file:
        file.write("This is a test file content")

    response = client.get("/read-file")
    assert response.status_code == 200
    assert response.json() == {"content": "This is a test file content"}

def test_schedule_email():
    response = client.post("/send-email/?email=fogigav197@rabitex.com")
    assert response.status_code == 200
    assert response.json() == {"message": "Email scheduled"}

@pytest.mark.asyncio
async def test_call_external_api():
    async with httpx.AsyncClient() as async_client:
        response = await async_client.get("https://jsonplaceholder.typicode.com/posts/1")
    assert response.status_code == 200
    assert "id" in response.json()

登入後複製

這測試後台電子郵件任務是否成功排程。

4.測試外部 API 呼叫

@app.get("/item/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}

登入後複製

這可確保 /external-api 端點正確地從外部來源取得資料。

輸出

Supercharge Your API Performance with Asynchronous Programming in FastAPI

結論

透過提供的程式碼，您現在可以實際了解如何使用 FastAPI 建立和測試非同步 API。無論是處理檔案、呼叫外部 API 還是調度後台任務，非同步程式設計都可以讓您建立可輕鬆擴展的高效能應用程式。

準備好建立您的下一個 FastAPI 專案了嗎？讓我們開始吧！

感謝您的閱讀...
編碼快樂！

以上是透過 FastAPI 中的非同步程式設計增強您的 API 效能的詳細內容。更多資訊請關注PHP中文網其他相關文章！