深入浅出FastAPI：现代Python Web开发的利器

匿名上传

发布时间:2026-01-30 13:00:01

诸神缄默不语-个人技术博文与视频目录

官方文档：FastAPI

fastapi · PyPI

什么是FastAPI？

FastAPI是一个现代、快速（高性能）的Python Web框架，专门用于构建API。它基于标准Python类型提示，使用Python 3.6+版本的类型提示特性，让API开发变得简单而强大。

FastAPI的核心优势

极快的性能：与NodeJS和Go相当，是Python领域最快的框架之一
高效的编码：开发速度提升约200%-300%
更少的bug：减少约40%的人为错误
直观的API：强大的编辑器支持，自动补全无处不在
完整的标准：基于并完全兼容开放的API标准（OpenAPI和JSON Schema）

安装FastAPI

pip install "fastapi[standard]"

第一个FastAPI应用

让我们从一个简单的"Hello World"开始：

from fastapi import FastAPI

# 创建FastAPI实例
app = FastAPI()

# 定义根路径的路由
@app.get("/")
async def read_root():
    return {"message": "Hello World"}

# 带路径参数的路由
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

保存为 try1.py，然后运行：

python -m uvicorn try1:app --reload

访问，你将看到JSON响应：{"message": "Hello World"}

uvicorn也可以使用Python脚本运行：创建一个 try2.py 文件：

import uvicorn

if __name__ == "__main__":
    uvicorn.run("try1:app", host="127.0.0.1", port=8000, reload=True)

然后运行：

python try2.py

自动API文档

FastAPI最酷的特性之一就是自动生成交互式API文档！

Swagger UI:
ReDoc:

这些文档是完全交互的，你可以直接在浏览器中测试API！

由于网络问题，可能打开后无法显示内容，可以在F12中看到需要从外网下js：

深入FastAPI特性

1. 类型提示和数据验证

from typing import Optional
from pydantic import BaseModel

# 定义数据模型
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

@app.post("/items/")
async def create_item(item: Item):
    item_dict = item.dict()
    if item.tax:
        price_with_tax = item.price + item.tax
        item_dict.update({"price_with_tax": price_with_tax})
    return item_dict

2. 路径参数和查询参数

from enum import Enum

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
    if model_name == ModelName.alexnet:
        return {"model_name": model_name, "message": "Deep Learning FTW!"}
    
    if model_name.value == "lenet":
        return {"model_name": model_name, "message": "LeCNN all the images"}
    
    return {"model_name": model_name, "message": "Have some residuals"}

@app.get("/users/")
async def read_users(skip: int = 0, limit: int = 10, q: Optional[str] = None):
    # 模拟从数据库获取用户
    users = [{"id": i, "name": f"User {i}"} for i in range(100)]
    
    if q:
        users = [user for user in users if q.lower() in user["name"].lower()]
    
    return users[skip: skip + limit]

3. 请求体和复杂数据模型

from datetime import datetime
from typing import List

class User(BaseModel):
    id: int
    name: str = "John Doe"
    signup_ts: Optional[datetime] = None
    friends: List[int] = []

class ComplexData(BaseModel):
    user: User
    items: List[Item]

@app.put("/complex-data/{data_id}")
async def update_complex_data(data_id: int, data: ComplexData):
    return {
        "data_id": data_id,
        "data": data.dict(),
        "received_at": datetime.now()
    }

4. 依赖注入系统

from fastapi import Depends, HTTPException, status

# 简单的依赖
async def common_parameters(
    skip: int = 0, 
    limit: int = 100,
    q: Optional[str] = None
):
    return {"skip": skip, "limit": limit, "q": q}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return commons

# 更复杂的依赖注入示例
class Database:
    async def get_user(self, user_id: int):
        # 模拟数据库查询
        return {"user_id": user_id, "name": f"User {user_id}"}

async def get_database():
    return Database()

async def get_current_user(
    user_id: int, 
    db: Database = Depends(get_database)
):
    user = await db.get_user(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user

@app.get("/users/me")
async def read_current_user(current_user: dict = Depends(get_current_user)):
    return current_user

5. 错误处理

from fastapi import HTTPException
from fastapi.responses import JSONResponse

# 自定义异常处理器
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": exc.detail, "error": True}
    )

# 抛出HTTP异常
@app.get("/protected-route")
async def protected_route(token: str = None):
    if not token or token != "secret-token":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return {"message": "Access granted"}

6. 中间件和CORS

from fastapi.middleware.cors import CORSMiddleware
import time

# 添加CORS中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 生产环境中应该指定具体域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 自定义中间件：添加处理时间头
@app.middleware("http")
async def add_process_time_header(request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

完整示例：待办事项API

让我们构建一个完整的待办事项管理API：

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
from uuid import UUID, uuid4

app = FastAPI(title="Todo API", version="1.0.0")

# 数据模型
class TodoItem(BaseModel):
    id: Optional[UUID] = None
    title: str
    description: Optional[str] = None
    completed: bool = False

# 模拟数据库
class TodoDB:
    def __init__(self):
        self.todos: List[TodoItem] = []
    
    def get_all(self) -> List[TodoItem]:
        return self.todos
    
    def get(self, todo_id: UUID) -> Optional[TodoItem]:
        for todo in self.todos:
            if todo.id == todo_id:
                return todo
        return None
    
    def create(self, todo: TodoItem) -> TodoItem:
        todo.id = uuid4()
        self.todos.append(todo)
        return todo
    
    def update(self, todo_id: UUID, updated_todo: TodoItem) -> Optional[TodoItem]:
        for idx, todo in enumerate(self.todos):
            if todo.id == todo_id:
                updated_todo.id = todo_id
                self.todos[idx] = updated_todo
                return updated_todo
        return None
    
    def delete(self, todo_id: UUID) -> bool:
        for idx, todo in enumerate(self.todos):
            if todo.id == todo_id:
                del self.todos[idx]
                return True
        return False

# 依赖注入
def get_db():
    return TodoDB()

# API路由
@app.get("/todos", response_model=List[TodoItem])
async def list_todos(db: TodoDB = Depends(get_db)):
    return db.get_all()

@app.get("/todos/{todo_id}", response_model=TodoItem)
async def get_todo(todo_id: UUID, db: TodoDB = Depends(get_db)):
    todo = db.get(todo_id)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo

@app.post("/todos", response_model=TodoItem)
async def create_todo(todo: TodoItem, db: TodoDB = Depends(get_db)):
    return db.create(todo)

@app.put("/todos/{todo_id}", response_model=TodoItem)
async def update_todo(todo_id: UUID, updated_todo: TodoItem, db: TodoDB = Depends(get_db)):
    todo = db.update(todo_id, updated_todo)
    if not todo:
        raise HTTPException(status_code=404, detail="Todo not found")
    return todo

@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: UUID, db: TodoDB = Depends(get_db)):
    if not db.delete(todo_id):
        raise HTTPException(status_code=404, detail="Todo not found")
    return {"message": "Todo deleted successfully"}

部署FastAPI应用

使用Uvicorn部署：

# 开发环境（带热重载）
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# 生产环境
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

使用Docker + uvicorn部署：

FROM python:3.9

WORKDIR /code

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]

常见bug

Ctrl+C无法关闭进程，无法返回终端prompt 如果运行uv run uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload就无法正常关闭，只能用Python脚本来运行。就直接在入口脚本（uvicorn.run()的脚本）上直接加：

def receive_signal(signalNumber, frame):
    print('nReceived signal:', signalNumber)
    print('正在关闭FastAPI服务...')
    # 强制退出，不等待子进程
    os._exit(0)

# 注册信号处理
signal.signal(signal.SIGINT, receive_signal)  # 处理Ctrl+C
signal.signal(signal.SIGTERM, receive_signal)  # 处理终止信号

参考资料：stackoverflow.com/questions/7…