准备好将您的 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中文网其他相关文章！
