使用 FastAPI 中的异步编程增强你的 API 性能

# Web 开发 # 教程 # Python

准备好将您的 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.**测试文件读取**

def test_read_file():
    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"}

这将创建一个测试文件并检查 `/read-file` 端点是否正确读取并返回其内容。

3.**测试后台任务**

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

这将测试后台电子邮件任务是否成功安排。

4.**测试外部 API 调用**

@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()

这确保 `/external-api` 端点正确地从外部源获取数据。

输出

结论

通过提供的代码，您现在已经对如何使用 FastAPI 构建和测试异步 API 有了实际的了解。无论是处理文件、调用外部 API 还是安排后台任务，异步编程都可以让您轻松创建可扩展的高性能应用程序。

准备好构建你的下一个 FastAPI 项目了吗？让我们开始吧！

感谢阅读...

**编码愉快！**

CLIS.CC