準備好將您的 API 遊戲提升到新的水平了嗎？
FastAPI 設計讓您的 API 更快、更能回應靈敏，並且能夠像專業人士一樣處理重負載。
在本文中，我們將向您展示如何利用 FastAPI 中的非同步程式設計來建立高效能 API。最後，您將具備實現非同步端點並有效測試它們的知識。
這是您將要掌握的內容：
非同步程式設計使任務能夠並發運作。這對於網路請求、資料庫查詢或文件操作等通常需要等待回應的任務特別有用。
在傳統的同步程式設計中，任務會依序運行，導致處理多個請求時出現延遲。透過非同步編程，您可以同時為多個使用者提供服務，最大限度地提高資源利用率並確保更好的使用者體驗。
現在，讓我們捲起袖子，創造一些令人驚嘆的東西！
首先，安裝所需的庫：
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()
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 端點正確地從外部來源取得資料。
透過提供的程式碼，您現在可以實際了解如何使用 FastAPI 建立和測試非同步 API。無論是處理檔案、呼叫外部 API 還是調度後台任務，非同步程式設計都可以讓您建立可輕鬆擴展的高效能應用程式。
準備好建立您的下一個 FastAPI 專案了嗎？讓我們開始吧！
感謝您的閱讀...
編碼快樂！
以上是透過 FastAPI 中的非同步程式設計增強您的 API 效能的詳細內容。更多資訊請關注PHP中文網其他相關文章！
