深入解读 FastAPI：现代、高性能的 Python Web 框架

用户8589624

发布于 2025-11-15 12:27:05

1100

文章被收录于专栏：nginxnginx

深入解读 FastAPI：现代、高性能的 Python Web 框架

FastAPI 是基于 Python 的现代 Web 框架，以其高性能、易用性和自动生成的交互式 API 文档而备受关注。它适用于需要快速开发和高吞吐量的应用场景，如微服务、实时数据处理和机器学习服务。

1. 什么是 FastAPI？

FastAPI 是一个高性能的 Web 框架，基于 Python 3.7+ 构建，依赖于 Starlette（用于网络框架核心）和 Pydantic（用于数据校验）。它以 OpenAPI 和 JSON Schema 标准为核心，为开发者提供自动化的文档生成和强大的数据验证功能。

1.1 FastAPI 的主要特点

高性能：性能接近 Node.js 和 Go，基于异步支持（asyncio）。
易用性：开发者体验极佳，支持类型提示。
自动文档：内置自动生成 Swagger UI 和 ReDoc 文档。
内置校验：使用 Pydantic 进行数据校验，保证输入输出的准确性。
支持异步编程：通过 async 和 await 实现高并发。

2. 快速上手

2.1 安装 FastAPI 和 Uvicorn

FastAPI 通常与 Uvicorn 一起使用，后者是一个轻量级的 ASGI 服务器。

pip install fastapi uvicorn

2.2 构建第一个 FastAPI 应用

创建文件 main.py：

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "Hello, FastAPI!"}

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

运行服务器：

uvicorn main:app --reload

访问以下地址即可看到结果：

http://127.0.0.1:8000：返回 {"message": "Hello, FastAPI!"}。
http://127.0.0.1:8000/items/42?q=example：返回 {"item_id": 42, "q": "example"}。

自动生成的文档：

Swagger UI: http://127.0.0.1:8000/docs
ReDoc: http://127.0.0.1:8000/redoc

3. FastAPI 核心功能详解

3.1 路由

FastAPI 支持 GET、POST、PUT、DELETE 等 HTTP 方法。示例：

from fastapi import FastAPI

app = FastAPI()

@app.post("/items/")
async def create_item(name: str, price: float):
    return {"name": name, "price": price}

3.2 请求体和数据校验

利用 Pydantic 模型定义请求体：

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    description: str = None
    price: float
    tax: float = None

@app.post("/items/")
async def create_item(item: Item):
    return {"item_name": item.name, "item_price": item.price}

传入 JSON 数据：

{
  "name": "Laptop",
  "description": "A high-end laptop",
  "price": 1500.99,
  "tax": 100.0
}

3.3 查询参数

查询参数具有自动校验和类型转换功能：

@app.get("/search/")
async def search_items(keyword: str, limit: int = 10):
    return {"keyword": keyword, "limit": limit}

3.4 路径参数

路径参数可以通过类型提示自动校验：

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

3.5 响应模型

通过 response_model 指定返回数据结构：

from typing import List

class ItemOut(BaseModel):
    name: str
    price: float

@app.get("/items/", response_model=List[ItemOut])
async def list_items():
    return [
        {"name": "Phone", "price": 699.99, "tax": 50},
        {"name": "Laptop", "price": 1500.99}
    ]

4. 构建实战项目：Todo 应用

4.1 项目需求

用户可以创建、查看、更新和删除任务。
数据保存在内存中（可扩展为数据库）。

4.2 定义 Pydantic 模型

from pydantic import BaseModel

class TodoItem(BaseModel):
    id: int
    title: str
    completed: bool

4.3 实现路由

from fastapi import FastAPI, HTTPException
from typing import List

app = FastAPI()

todos = []

@app.post("/todos/", response_model=TodoItem)
async def create_todo(todo: TodoItem):
    todos.append(todo)
    return todo

@app.get("/todos/", response_model=List[TodoItem])
async def list_todos():
    return todos

@app.get("/todos/{todo_id}", response_model=TodoItem)
async def get_todo(todo_id: int):
    for todo in todos:
        if todo.id == todo_id:
            return todo
    raise HTTPException(status_code=404, detail="Todo not found")

@app.put("/todos/{todo_id}", response_model=TodoItem)
async def update_todo(todo_id: int, updated_todo: TodoItem):
    for index, todo in enumerate(todos):
        if todo.id == todo_id:
            todos[index] = updated_todo
            return updated_todo
    raise HTTPException(status_code=404, detail="Todo not found")

@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: int):
    for index, todo in enumerate(todos):
        if todo.id == todo_id:
            todos.pop(index)
            return {"message": "Todo deleted"}
    raise HTTPException(status_code=404, detail="Todo not found")

4.4 运行测试

启动服务器后，可以通过 Swagger UI 测试以下功能：

创建任务。
查看任务列表。
更新任务状态。
删除任务。

5. 高级功能

5.1 中间件

FastAPI 支持中间件，用于处理全局逻辑（如请求日志记录）：

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
)

5.2 后台任务

FastAPI 提供 BackgroundTasks 支持异步任务处理：

from fastapi import BackgroundTasks

async def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(f"{message}\n")

@app.post("/log/")
async def log_message(message: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, message)
    return {"message": "Task submitted"}

5.3 集成数据库

使用 SQLAlchemy 集成数据库：

pip install sqlalchemy databases

配置示例：

from databases import Database
from sqlalchemy import create_engine, MetaData

DATABASE_URL = "sqlite:///./test.db"
database = Database(DATABASE_URL)
engine = create_engine(DATABASE_URL)
metadata = MetaData()