通过 FastAPI 中的异步编程增强您的 API 性能

准备好将您的 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 端点正确地从外部源获取数据。

输出

结论

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

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

感谢您的阅读...
编码快乐！

以上是通过 FastAPI 中的异步编程增强您的 API 性能的详细内容。更多信息请关注PHP中文网其他相关文章！