FastAPI：高性能 Web 开发的现代选择

你是否有过这样的困扰——想把自己训练的机器学习模型对外提供 API 服务，却发现传统框架性能捉襟见肘？或者你正在开发一个小型应用，希望快速搭建一套高并发、自带文档的 RESTful API？FastAPI 正是为解决这些痛点而生的现代 Python Web 框架。它诞生于 2018 年，专门应对“高并发 + 自动生成接口文档”两大核心需求，尤其在 AI 模型部署和微服务领域表现卓越。FastAPI 深度集成 Python 的类型注解，基于 Starlette（异步 Web 框架）和 Pydantic（高性能数据校验），实测中单节点 QPS 可达 3000+，是传统同步框架的 5-8 倍。无论是开发电商后台的订单处理接口，还是为移动 App 提供实时数据更新服务，FastAPI 都能以极高的性能和极其优雅的代码风格，让你专注于业务逻辑，而不是框架本身。

截至 2026 年，FastAPI 已在 GitHub 上积累了超过 80k Star，增长速度在所有 Python Web 框架中保持领先，并被微软、Netflix、滴滴等多家知名公司广泛采用。

安装 FastAPI

安装 FastAPI 极为简单，核心依赖只有 fastapi 本身和 ASGI 服务器 uvicorn：

bash

pip install fastapi uvicorn[standard]

建议在虚拟环境中执行上述命令，以隔离项目依赖。uvicorn 是一个基于 uvloop 和 httptools 构建的高性能 ASGI 服务器，负责处理 HTTP 请求并运行 FastAPI 应用。

基本用法

第一步：创建应用实例

创建一个名为 main.py 的文件，导入 FastAPI 并实例化应用对象：

python

from fastapi import FastAPI

app = FastAPI(title="My First FastAPI App")

第二步：定义 GET 路由

使用 @app.get() 装饰器将函数与根路径的 GET 请求绑定：

python

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

第三步：运行服务

在终端执行以下命令启动服务：

bash

uvicorn main:app --reload --host 0.0.0.0 --port 8000

其中 --reload 参数会在代码变更时自动重启，便于开发调试。访问 http://localhost:8000 即可看到返回的 JSON 数据。

第四步：探索自动生成的文档

FastAPI 最强大的特性之一就是自动生成交互式 API 文档。启动服务后，访问以下两个地址：

• Swagger UI：http://localhost:8000/docs

• ReDoc：http://localhost:8000/redoc

你编写的每一个路由都会自动出现在文档中，并且可以直接在页面上测试接口。

高级用法

依赖注入系统

FastAPI 内置了极其灵活的依赖注入系统。通过声明式参数，你可以轻松实现认证校验、数据库会话管理、配置读取等横切关注点的复用。例如：

python

from fastapi import Depends, HTTPException, status

def verify_token(token: str):
    # 这里编写 token 校验逻辑
    if not is_valid(token):
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"user_id": 123}

@app.get("/protected")
async def protected_route(user_data: dict = Depends(verify_token)):
    return {"message": f"Welcome user {user_data['user_id']}"}

WebSocket 实时通信

FastAPI 原生支持 WebSocket，非常适合开发聊天室、实时通知等应用场景。WebSocket 连接管理是高频应用的核心模式，建议配合连接管理器来统一处理广播消息和断连清理：

python

from fastapi import WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"Client {client_id} says: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"Client {client_id} left the chat")

实际应用场景案例

场景一：局域网内 P2P 文件分享与实时聊天

在企业内部或宿舍局域网中，集中式文件传输需要服务器中转所有文件，容易产生存储瓶颈和隐私问题。借助 FastAPI 的 WebSocket 和异步 HTTP 端点，我们可以实现一个混合架构：中央服务器仅负责用户发现和消息路由，文件传输直接在客户端之间完成（P2P），传输速度可达 LAN 级别的 100+ MB/s。

下面是一个简化版的 P2P 文件分享服务实现：

python

import hashlib
import uuid
from pathlib import Path
from typing import Dict, Set
from fastapi import FastAPI, UploadFile, File, HTTPException, Depends, WebSocket
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel

app = FastAPI(title="P2P File Sharing Service")

security = HTTPBearer()
active_users: Dict[str, dict] = {}  # user_id -> {websocket, ip, port, files}
pending_shares: Dict[str, dict] = {}  # share_id -> {from_user, file_name, size, p2p_url}

class FileMetadata(BaseModel):
    file_name: str
    size: int
    p2p_url: str

def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    # 实际生产中应对 token 进行校验并解析用户信息
    if not token or len(token) < 10:
        raise HTTPException(status_code=401, detail="Invalid token")
    return {"user_id": token[:8]}

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    user_id = None
    try:
        data = await websocket.receive_json()
        user_id = data.get("user_id")
        if not user_id:
            await websocket.close(code=1008)
            return
        active_users[user_id] = {"websocket": websocket, "files": set()}
        await websocket.send_json({"type": "connected", "message": "Welcome!"})

        while True:
            msg = await websocket.receive_json()
            msg_type = msg.get("type")

            if msg_type == "share_file":
                # 用户分享文件：生成分享ID，通知在线用户
                share_id = str(uuid.uuid4())[:8]
                pending_shares[share_id] = {
                    "from_user": user_id,
                    "file_name": msg["file_name"],
                    "size": msg["size"],
                    "p2p_url": msg["p2p_url"]
                }
                await broadcast_to_all({"type": "new_share", "share_id": share_id, **pending_shares[share_id]})

            elif msg_type == "request_file":
                share_id = msg["share_id"]
                share_info = pending_shares.get(share_id)
                if share_info:
                    # 将 P2P 地址直接返回给请求者
                    await websocket.send_json({"type": "file_info", "p2p_url": share_info["p2p_url"]})
    except Exception as e:
        print(f"WebSocket error: {e}")
    finally:
        if user_id and user_id in active_users:
            del active_users[user_id]

async def broadcast_to_all(message: dict):
    for user_id, info in active_users.items():
        try:
            await info["websocket"].send_json(message)
        except:
            pass

@app.post("/upload")
async def upload_file(file: UploadFile = File(...), user: dict = Depends(get_current_user)):
    file_content = await file.read()
    file_hash = hashlib.md5(file_content).hexdigest()
    save_path = Path(f"storage/{user['user_id']}_{file_hash}.bin")
    save_path.parent.mkdir(exist_ok=True)
    with open(save_path, "wb") as f:
        f.write(file_content)
    return {"file_id": file_hash, "size": len(file_content), "p2p_url": f"http://192.168.1.100:8001/download/{file_hash}"}

@app.get("/download/{file_id}")
async def download_file(file_id: str, user: dict = Depends(get_current_user)):
    for f in Path("storage").glob(f"*_{file_id}.bin"):
        return FileResponse(f, filename=file_id + ".bin")
    raise HTTPException(status_code=404, detail="File not found")

场景二：机器学习模型服务部署

FastAPI 已成为机器学习模型服务部署的首选框架。将训练好的模型通过 FastAPI 封装为 REST API，既可以利用其异步特性处理高并发推理请求，又能通过自动生成的文档让下游开发者快速接入。FastAPI 2026 年的数据显示，38% 的 Python 开发者已在其生产环境中使用 FastAPI，这一比例在 ML 工程领域尤其突出。

python

import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib

app = FastAPI(title="House Price Prediction API")

model = joblib.load("house_price_model.pkl")

class HouseFeatures(BaseModel):
    area: float
    bedrooms: int
    location_score: float = 8.0

class PredictionResponse(BaseModel):
    predicted_price: float
    confidence: float

@app.post("/predict", response_model=PredictionResponse)
async def predict_price(house: HouseFeatures):
    features = np.array([[house.area, house.bedrooms, house.location_score]])
    try:
        prediction = model.predict(features)[0]
        confidence = 0.92  # 示例置信度，实际应通过模型评估获取
        return {"predicted_price": float(prediction), "confidence": confidence}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Prediction error: {str(e)}")

---

FastAPI 以极致的性能、优雅的类型安全设计以及自动生成的交互式文档，彻底改变了 Python Web 开发体验。无论你是在构建微服务架构、部署机器学习模型，还是开发实时通信应用，FastAPI 都堪称不二之选。其 80k+ GitHub Star 和高速增长的社区充分证明了它在开发者群体中的认可度。希望这篇介绍能帮你快速上手并应用到实际项目中。

如果你已经使用过 FastAPI，欢迎在评论区分享你的项目经验；如果还在观望，不妨从今天开始，用十分钟搭建你的第一个高性能 API——相信我，你会爱上这种“写代码即写文档”的畅快感！