当前位置： 首页 > news >正文

《Python实战进阶》专栏 No 4：高效、简洁、强大之使用 FastAPI 构建高性能异步API

news 来源：原创 2025/5/6 1:33:00

《Python实战进阶》专栏 第4集：高效、简洁、强大之使用 FastAPI 构建高性能异步API

引言

在现代 Web 开发中，构建高性能、可扩展的 API 是一项核心任务。FastAPI 是一个基于 Python 的现代化框架，以其高性能和易用性迅速成为开发者的首选工具之一。它结合了异步编程的强大功能与 Pydantic 的数据验证能力，特别适合构建实时应用和微服务。

本集将带你深入了解 FastAPI 的核心特性，并通过一个实际项目——构建一个支持异步操作的 RESTful API，展示如何利用 FastAPI 实现高性能的后端服务。

在这里插入图片描述

一、FastAPI 的核心特性

1.1 高性能

FastAPI 基于 Starlette（用于处理 HTTP 请求）和 Pydantic（用于数据验证），其性能接近 Node.js 和 Go 等语言的框架。根据官方基准测试，FastAPI 的性能仅次于 Starlette 和 Uvicorn。

1.2 异步支持

FastAPI 天然支持异步编程，允许开发者轻松实现非阻塞 I/O 操作。这对于需要处理大量并发请求的应用（如实时聊天或流媒体服务）尤为重要。

1.3 自动生成文档

FastAPI 内置 Swagger UI 和 ReDoc，能够自动生成交互式 API 文档，极大简化了开发和调试过程。

1.4 数据验证与类型提示

通过 Pydantic，FastAPI 提供强大的数据验证功能，并支持 Python 类型提示，确保输入数据的正确性和一致性。

二、实战项目：构建一个高性能的异步 API

我们将通过一个简单的任务管理系统来演示 FastAPI 的使用。该系统支持以下功能：

  • 创建任务
  • 查询任务列表
  • 获取单个任务详情
  • 更新任务状态
  • 删除任务

2.1 环境准备

首先，安装 FastAPI 和 Uvicorn（ASGI 服务器）：

pip install fastapi uvicorn

2.2 定义数据模型

使用 Pydantic 定义任务的数据模型：

from pydantic import BaseModel
from typing import Optional
from datetime import datetime

class Task(BaseModel):
    id: int
    title: str
    description: Optional[str] = None
    status: str = "pending"
    created_at: datetime = datetime.now()

class TaskCreate(BaseModel):
    title: str
    description: Optional[str] = None

2.3 构建 API

以下是完整的 FastAPI 应用代码, 将文件名存为 main.py ：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetime

app = FastAPI()

# 数据模型
class Task(BaseModel):
    id: int
    title: str
    description: Optional[str] = None
    status: str = "pending"
    created_at: datetime = datetime.now()

class TaskCreate(BaseModel):
    title: str
    description: Optional[str] = None

# 模拟数据库（初始为空）
tasks_db = []

# 初始化模拟数据
def initialize_mock_data():
    global tasks_db
    tasks_db = [
        Task(id=1, title="Learn FastAPI", description="Build a high-performance API", status="in_progress"),
        Task(id=2, title="Write Blog Post", description="Publish an article on FastAPI", status="pending"),
        Task(id=3, title="Deploy to Production", description="Set up CI/CD pipeline", status="completed"),
    ]

# 调用初始化函数
initialize_mock_data()

# 创建任务
@app.post("/tasks/", response_model=Task)
async def create_task(task: TaskCreate):
    new_task = Task(id=len(tasks_db) + 1, title=task.title, description=task.description)
    tasks_db.append(new_task)
    return new_task

# 获取任务列表
@app.get("/tasks/", response_model=List[Task])
async def get_tasks():
    return tasks_db

# 获取单个任务
@app.get("/tasks/{task_id}", response_model=Task)
async def get_task(task_id: int):
    if task_id < 1 or task_id > len(tasks_db):
        raise HTTPException(status_code=404, detail="Task not found")
    return tasks_db[task_id - 1]

# 更新任务状态
@app.put("/tasks/{task_id}", response_model=Task)
async def update_task(task_id: int, status: str):
    if task_id < 1 or task_id > len(tasks_db):
        raise HTTPException(status_code=404, detail="Task not found")
    task = tasks_db[task_id - 1]
    task.status = status
    return task

# 删除任务
@app.delete("/tasks/{task_id}")
async def delete_task(task_id: int):
    if task_id < 1 or task_id > len(tasks_db):
        raise HTTPException(status_code=404, detail="Task not found")
    tasks_db.pop(task_id - 1)
    return {"message": "Task deleted successfully"}

2.4 启动服务

运行以下命令启动 API 服务，其中main为主文件名，如果你保存的不是main，请相应更改下：

uvicorn main:app --reload
访问 http://127.0.0.1:8000/docs，你将看到自动生成的 Swagger UI 文档界面。

在这里插入图片描述

代码关键点解读

  1. 延迟初始化模拟数据：

    • 将模拟数据的初始化放在一个单独的函数 initialize_mock_data 中，并在 Task 类定义完成后调用该函数。
    • 这样可以确保 Task 类在使用之前已经定义完成。

  2. 全局变量的使用：

    • 使用 global tasks_db 确保模拟数据能够正确更新到全局变量 tasks_db 中。

运行步骤

  1. 保存代码：
    将上述代码保存为 main.py 文件。

  2. 启动服务：
    在终端中运行以下命令启动 FastAPI 服务：

    uvicorn main:app --reload

  3. 访问 Swagger UI：
    打开浏览器并访问 http://127.0.0.1:8000/docs，你应该能看到完整的 API 文档界面。

  4. 测试 API：

    • 使用 /tasks/ 的 GET 方法查看模拟数据。
    • 使用 /tasks/ 的 POST 方法创建新任务。
    • 使用 /tasks/{task_id} 的 PUT 和 DELETE 方法更新和删除任务。

预期结果

1. 查看任务列表

  • 访问 /tasks/ 的 GET 方法，在浏览器输入：http://127.0.0.1:8000/tasks/?get，返回的模拟数据如下：
    [
  {
    "id": 1,
    "title": "Learn FastAPI",
    "description": "Build a high-performance API",
    "status": "in_progress",
    "created_at": "2023-10-01T12:00:00"
  },
  {
    "id": 2,
    "title": "Write Blog Post",
    "description": "Publish an article on FastAPI",
    "status": "pending",
    "created_at": "2023-10-01T12:05:00"
  },
  {
    "id": 3,
    "title": "Deploy to Production",
    "description": "Set up CI/CD pipeline",
    "status": "completed",
    "created_at": "2023-10-01T12:10:00"
  }
]

2. 创建新任务

  • 使用 /tasks/ 的 POST 方法创建任务，例如：
    {
  "title": "Test Task",
  "description": "This is a test task"
}

也可以直接使用 Swagger UI界面直接进行 Post 调试
在 Swagger UI页面点击 /tasks/ 的 GET 方法，点击 “Try it out”，然后点击 “Execute”。
在这里插入图片描述

  • 返回结果：
    {
  "id": 4,
  "title": "Test Task",
  "description": "This is a test task",
  "status": "pending",
  "created_at": "2023-10-01T12:15:00"
}

** Swagger UI页面反馈**
在这里插入图片描述

3. 更新任务状态

  • 使用 /tasks/{task_id} 的 PUT 方法更新任务状态，例如：
    {
  "status": "completed"
}
  • 返回结果：
    {
  "id": 1,
  "title": "Learn FastAPI",
  "description": "Build a high-performance API",
  "status": "completed",
  "created_at": "2023-10-01T12:00:00"
}

4. 删除任务

  • 使用 /tasks/{task_id} 的 DELETE 方法删除任务，例如 task_id=1。
  • 返回结果：
    {
  "message": "Task deleted successfully"
}

三、异步编程的优势

为了展示 FastAPI 的异步能力，我们可以在任务创建时模拟一个耗时操作（如调用外部 API）。以下是修改后的代码片段：

import asyncio

@app.post("/tasks/", response_model=Task)
async def create_task(task: TaskCreate):
    # 模拟耗时操作
    await asyncio.sleep(2)
    new_task = Task(id=len(tasks_db) + 1, title=task.title, description=task.description)
    tasks_db.append(new_task)
    return new_task

优势：

  • 在同步框架中，耗时操作会阻塞整个线程，而 FastAPI 的异步支持允许其他请求在此期间继续处理。

四、性能测试

我们可以使用 Apache Benchmark (ab) 或 Locust 测试 FastAPI 的性能。以下是一个简单的测试命令：

ab -n 1000 -c 100 http://127.0.0.1:8000/tasks/

结果分析：

  • FastAPI 能够轻松处理高并发请求，响应时间极短。
  • 相比传统的同步框架（如 Flask），FastAPI 的吞吐量显著提升。

五、总结与展望

通过本集的学习，我们了解了 FastAPI 的核心特性，并通过一个实战项目展示了如何构建高性能的异步 API。FastAPI 的易用性和高性能使其成为现代 Web 开发的理想选择，特别是在需要处理大量并发请求的场景中。

希望这篇文章能为你提供清晰的学习路径！如果你有任何问题或想法，欢迎在评论区留言讨论。

下集预告：第5集将聚焦于 GraphQL vs RESTful API 的对比与实现，探讨两种 API 设计范式的优劣及适用场景。

附录：Apache Benchmark (ab) 的安装和使用

Apache Benchmark (ab) 是 Apache HTTP Server 自带的一个简单而强大的工具，用于对 Web 服务器进行压力测试和性能评估。以下是安装和使用 ab 的详细步骤。

一、安装 Apache Benchmark

1. 在 Linux 系统上安装

在大多数 Linux 发行版中，ab 是通过安装 Apache HTTP Server 工具包提供的。你可以使用以下命令安装：

Ubuntu/Debian
sudo apt update
sudo apt install apache2-utils
CentOS/RHEL
sudo yum install httpd-tools

安装完成后，你可以通过以下命令验证是否安装成功：

ab -V

输出示例：

This is ApacheBench, Version 2.3 <$Revision: 1843412 $>
Copyright 1996 Adam Twiss, Zeus Technology Ltd, http://www.zeustech.net/
Licensed to The Apache Software Foundation, http://www.apache.org/

2. 在 macOS 上安装

macOS 用户可以通过 Homebrew 安装 ab：

brew install httpd

安装完成后，ab 命令可以直接使用。

3. 在 Windows 上安装

Windows 用户可以通过安装 XAMPP 或单独安装 Apache HTTP Server 来获取 ab 工具。

方法 1：通过 XAMPP 安装
  1. 下载并安装 XAMPP。
  2. 安装完成后，ab 可执行文件通常位于 C:\xampp\apache\bin\ab.exe。
  3. 将该路径添加到系统环境变量中，以便直接使用 ab 命令。
方法 2：单独安装 Apache
  1. 下载 Apache HTTP Server 的 Windows 版本（可以从 Apache Lounge 获取）。
  2. 解压后找到 ab.exe，将其路径添加到系统环境变量中。

二、使用 Apache Benchmark 进行性能测试

1. 基本语法

ab 的基本语法如下：

ab [选项] [URL]

常用选项：

  • -n requests：总请求数。
  • -c concurrency：并发用户数。
  • -t timelimit：测试的最长时间（秒）。
  • -k：启用 HTTP KeepAlive（复用连接）。
  • -H header：自定义请求头。

2. 示例：测试 FastAPI 服务

假设你的 FastAPI 服务运行在 http://127.0.0.1:8000/tasks/，你可以使用以下命令进行测试：

ab -n 1000 -c 100 http://127.0.0.1:8000/tasks/

参数说明：

  • -n 1000：发送 1000 个请求。
  • -c 100：每次并发 100 个请求。

3. 测试结果分析

运行上述命令后，ab 会输出详细的性能报告，例如：

Server Software:        uvicorn
Server Hostname:        127.0.0.1
Server Port:            8000

Document Path:          /tasks/
Document Length:        312 bytes

Concurrency Level:      100
Time taken for tests:   2.567 seconds
Complete requests:      1000
Failed requests:        0
Requests per second:    389.56 [#/sec] (mean)
Time per request:       256.700 [ms] (mean)
Transfer rate:          120.45 [Kbytes/sec] received

关键指标：

  • Requests per second：每秒处理的请求数（吞吐量），越高越好。
  • Time per request：每个请求的平均响应时间，越低越好。
  • Failed requests：失败的请求数，理想情况下应为 0。

三、高级用法

1. 启用 HTTP KeepAlive

HTTP KeepAlive 可以复用 TCP 连接，减少连接建立的开销。使用 -k 参数启用：

ab -n 1000 -c 100 -k http://127.0.0.1:8000/tasks/

2. 自定义请求头

如果你的 API 需要特定的请求头（如 Authorization），可以使用 -H 参数：

ab -n 1000 -c 100 -H "Authorization: Bearer YOUR_TOKEN" http://127.0.0.1:8000/tasks/

3. 测试 POST 请求

默认情况下，ab 只支持 GET 请求。如果需要测试 POST 请求，可以结合 curl 或其他工具生成负载数据。

四、注意事项

  1. 避免生产环境测试：

    • 不要在生产环境中直接运行 ab，以免对真实用户造成影响。
    • 建议在本地或测试环境中进行性能测试。

  2. 调整系统限制：

    • 如果并发数较高（如 -c 1000），可能需要调整操作系统的文件描述符限制：
      ulimit -n 65535

  3. 监控资源使用：

    • 使用 top 或 htop 监控 CPU 和内存使用情况，确保测试不会耗尽系统资源。

五、总结

通过安装和使用 Apache Benchmark (ab)，你可以轻松对 Web 服务进行性能测试，并分析其吞吐量和响应时间等关键指标。对于 FastAPI 等高性能框架，ab 是一个简单而有效的工具，可以帮助你评估服务的性能瓶颈。

希望这篇文章能帮助你顺利安装和使用 ab！

相关文章：

  • 请简述一下Prefab（预制体）的本质是什么？
  • 大白话TypeScript第三章高级特性学习
  • 使用Socket编写超牛的http服务器和客户端(一）
  • L1-6 剪切粘贴（字符串增删查改）
  • INTJ人格的本质
  • 【JavaScript】《JavaScript高级程序设计 (第4版) 》笔记-Chapter21-错误处理与调试
  • Linux 命令大全完整版（12）
  • apply的用法
  • 自动驾驶中planning为什么要把横纵向分开优化？
  • AI知识架构之RAG
  • 如何在VMware虚拟机的window10系统中安装网易mumu模拟器
  • Linux权限（一）
  • 【Java】求绝对值
  • hive开窗函数边界值ROWS BETWEEN 和 RANGE BETWEEN区别
  • 【带你 langchain 双排系列教程】6.LangChain多模态输入与自定义输出实战指南
  • 机器学习数学通关指南——链式法则
  • Three.js轮廓线、边框线、选中效果实现的几种实现方法以及性能评估
  • Lock-Free环形队列C++实现
  • C进阶 自定义类型
  • 计算机性能指标
  • 外卖大战之外，缝隙中的校园到寝外卖和那些送餐的大学生们
  • 视觉周刊｜劳动开创未来
  • 新华每日电讯“关爱青年成长”三连评：青春应有多样的精彩
  • 9米长林肯车开进“皖南川藏线”致拥堵数小时，车主回应称将配合调查
  • 陈逸飞《黄河颂》人物造型与借鉴影像意义
  • 五一假期，这些短剧值得一刷