Python FastAPI 框架入门

2025-11-17 约 2832 字预计阅读 6 分钟

学习目标：掌握 FastAPI 的核心特性，理解类型驱动开发的威力，能够构建高性能异步 API 并生成交互式文档。

1. FastAPI 简介

FastAPI 是一个现代、高性能的 Python Web 框架，基于 Starlette（ASGI）和 Pydantic（数据验证），专为构建 API 而生。

1.1 核心特性

极速性能：与 Node.js 和 Go 相当，是 Python 最快框架之一
类型驱动：基于 Python 类型提示，自动数据验证
自动文档：自动生成 OpenAPI 和 Swagger UI 文档
异步原生：原生支持 async/await
依赖注入：强大的依赖注入系统
数据验证：基于 Pydantic，自动序列化和反序列化

1.2 安装

        
pip install fastapi uvicorn

# 可选：标准依赖（包含 ORJSON、Jinja2 等）
pip install fastapi[standard]

2. 快速开始

2.1 最小应用

        
# main.py
from fastapi import FastAPI

app = FastAPI(title="我的 API", version="1.0")

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

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

2.2 运行服务

        
# 开发模式
uvicorn main:app --reload --host 0.0.0.0 --port 8000

# 生产模式
uvicorn main:app --workers 4

# 使用 Hypercorn（HTTP/2 支持）
pip install hypercorn
hypercorn main:app --bind 0.0.0.0:8000

2.3 自动文档

启动后访问：

Swagger UI：http://localhost:8000/docs
ReDoc：http://localhost:8000/redoc
OpenAPI JSON：http://localhost:8000/openapi.json

3. 路径操作

3.1 HTTP 方法

        
        
        
    
from fastapi import FastAPI

app = FastAPI()

@app.get("/users")
def list_users():
    """获取用户列表"""
    return [{"id": 1, "name": "张三"}]

@app.post("/users")
def create_user():
    """创建用户"""
    return {"id": 2, "created": True}

@app.put("/users/{user_id}")
def update_user(user_id: int):
    """更新用户"""
    return {"id": user_id, "updated": True}

@app.patch("/users/{user_id}")
def partial_update(user_id: int):
    """部分更新"""
    return {"id": user_id, "patched": True}

@app.delete("/users/{user_id}")
def delete_user(user_id: int):
    """删除用户"""
    return {"id": user_id, "deleted": True}

3.2 路径参数

        
        
        
    
from fastapi import FastAPI
from enum import Enum

app = FastAPI()

# 基础路径参数
@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

# 枚举限制
class Role(str, Enum):
    admin = "admin"
    user = "user"
    guest = "guest"

@app.get("/roles/{role}")
def get_role(role: Role):
    if role == Role.admin:
        return {"role": role, "permissions": "all"}
    return {"role": role, "permissions": "limited"}

# 路径包含斜杠
@app.get("/files/{file_path:path}")
def read_file(file_path: str):
    return {"file_path": file_path}

4. 查询参数与请求体

4.1 查询参数

        
        
        
    
from fastapi import FastAPI
from typing import Optional

app = FastAPI()

# 可选参数带默认值
@app.get("/items/")
def read_items(
    skip: int = 0,
    limit: int = 10,
    search: Optional[str] = None
):
    return {
        "skip": skip,
        "limit": limit,
        "search": search
    }

# 布尔参数
@app.get("/items/filter/")
def filter_items(
    published: bool = False,
    category: Optional[str] = None
):
    return {
        "published_only": published,
        "category": category
    }

# 必需查询参数（无默认值）
@app.get("/calculate/")
def calculate(a: int, b: int, operation: str = "add"):
    if operation == "add":
        result = a + b
    elif operation == "multiply":
        result = a * b
    else:
        result = None
    return {"result": result}

4.2 Pydantic 模型（请求体）

        
        
        
    
from fastapi import FastAPI
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
from datetime import datetime

app = FastAPI()

class UserBase(BaseModel):
    """用户基础模型"""
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    full_name: Optional[str] = None

class UserCreate(UserBase):
    """创建用户"""
    password: str = Field(..., min_length=6)

class UserResponse(UserBase):
    """用户响应"""
    id: int
    created_at: datetime
    is_active: bool = True
    
    class Config:
        from_attributes = True

class Item(BaseModel):
    """商品模型"""
    name: str = Field(..., description="商品名称")
    description: Optional[str] = Field(None, max_length=300)
    price: float = Field(..., gt=0, description="必须大于 0")
    tax: Optional[float] = None
    tags: list[str] = []

# 创建用户
@app.post("/users", response_model=UserResponse)
def create_user(user: UserCreate):
    """创建新用户
    
    - **username**: 用户名（3-50 字符）
    - **email**: 有效邮箱
    - **password**: 至少 6 位
    """
    # 模拟创建
    return {
        "id": 1,
        **user.model_dump(),
        "created_at": datetime.now(),
        "is_active": True
    }

# 创建商品
@app.post("/items")
def create_item(item: Item):
    """创建商品"""
    item_dict = item.model_dump()
    if item.tax:
        item_dict["total"] = item.price + item.tax
    return item_dict

# 更新商品（部分更新）
@app.patch("/items/{item_id}")
def update_item(item_id: int, item: Item):
    return {"item_id": item_id, **item.model_dump()}

5. 响应处理

5.1 自定义响应

        
        
        
    
from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse, HTMLResponse, PlainTextResponse

app = FastAPI()

# 自定义状态码
@app.post("/items", status_code=201)
def create_item():
    return {"message": "创建成功"}

# 显式响应模型
@app.get("/legacy/")
def get_legacy():
    data = {"message": "旧版数据"}
    return JSONResponse(
        content=data,
        headers={"X-API-Version": "1.0"}
    )

# HTML 响应
@app.get("/page", response_class=HTMLResponse)
def read_page():
    return "<h1>Hello HTML</h1>"

# 设置 Cookie
@app.get("/set-cookie")
def set_cookie(response: Response):
    response.set_cookie(key="session", value="abc123")
    return {"message": "Cookie 已设置"}

# 文件下载
from fastapi.responses import FileResponse

@app.get("/download")
def download_file():
    return FileResponse(
        path="report.pdf",
        filename="年度报告.pdf",
        media_type="application/pdf"
    )

6. 依赖注入

6.1 基础依赖

        
        
        
    
from fastapi import FastAPI, Depends, Header, HTTPException
from typing import Optional

app = FastAPI()

# 定义依赖函数
async def get_token_header(x_token: str = Header(...)):
    """验证 Token"""
    if x_token != "secret-token":
        raise HTTPException(status_code=400, detail="无效的 Token")
    return x_token

async def common_parameters(
    q: Optional[str] = None,
    skip: int = 0,
    limit: int = 100
):
    """通用分页参数"""
    return {"q": q, "skip": skip, "limit": limit}

# 使用依赖
@app.get("/items/")
def read_items(commons: dict = Depends(common_parameters)):
    return commons

@app.get("/admin/items/")
def read_admin_items(
    commons: dict = Depends(common_parameters),
    token: str = Depends(get_token_header)
):
    return {"items": [], "token": token}

6.2 类依赖

        
        
        
    
class CommonQueryParams:
    """通用查询参数类"""
    def __init__(
        self,
        q: Optional[str] = None,
        skip: int = 0,
        limit: int = 100
    ):
        self.q = q
        self.skip = skip
        self.limit = limit

@app.get("/books/")
def read_books(commons: CommonQueryParams = Depends()):
    return {
        "query": commons.q,
        "pagination": {
            "skip": commons.skip,
            "limit": commons.limit
        }
    }

6.3 数据库依赖

        
        
        
    
from fastapi import FastAPI, Depends
from contextlib import contextmanager

# 模拟数据库连接
class Database:
    def connect(self):
        print("数据库连接")
        return self
    
    def disconnect(self):
        print("数据库断开")
    
    def query(self, sql: str):
        return [{"id": 1, "name": "Test"}]

db = Database()

# 依赖注入获取数据库连接
async def get_db():
    conn = db.connect()
    try:
        yield conn
    finally:
        conn.disconnect()

@app.get("/users/")
def get_users(db_conn = Depends(get_db)):
    return db_conn.query("SELECT * FROM users")

7. 错误处理

        
        
        
    
from fastapi import FastAPI, HTTPException, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

# 自定义异常
class ItemNotFoundException(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

# 注册异常处理器
@app.exception_handler(ItemNotFoundException)
async def item_not_found_handler(request, exc: ItemNotFoundException):
    return JSONResponse(
        status_code=404,
        content={"detail": f"商品 {exc.item_id} 不存在"}
    )

@app.exception_handler(RequestValidationError)
async def validation_handler(request, exc):
    return JSONResponse(
        status_code=422,
        content={
            "detail": "数据验证失败",
            "errors": exc.errors()
        }
    )

# 使用 HTTPException
@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id < 1:
        raise HTTPException(
            status_code=400,
            detail="ID 必须大于 0"
        )
    
    # 模拟查找
    items = {1: "Apple", 2: "Banana"}
    if item_id not in items:
        raise ItemNotFoundException(item_id)
    
    return {"item": items[item_id]}

# 使用状态码常量
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    # ... 删除逻辑
    raise HTTPException(
        status_code=status.HTTP_403_FORBIDDEN,
        detail="无权删除此商品"
    )

8. 异步支持

        
        
        
    
from fastapi import FastAPI
import asyncio

app = FastAPI()

# 异步路由
@app.get("/async-data")
async def get_async_data():
    await asyncio.sleep(1)  # 模拟异步操作
    return {"data": "异步获取的数据"}

# 混合使用（同步函数在线程池中运行）
def cpu_intensive_task(n: int):
    """CPU 密集型任务"""
    return sum(range(n))

@app.get("/compute/{n}")
async def compute(n: int):
    # FastAPI 自动在线程池中运行同步函数
    result = cpu_intensive_task(n)
    return {"result": result}

# 异步数据库操作（示例）
async def fetch_from_db(query: str):
    await asyncio.sleep(0.1)
    return [{"id": 1, "data": "result"}]

@app.get("/users-async")
async def get_users_async():
    users = await fetch_from_db("SELECT * FROM users")
    return {"users": users}

9. 后台任务

        
        
        
    
from fastapi import FastAPI, BackgroundTasks
from typing import Optional
import time

app = FastAPI()

def send_email(email: str, message: str):
    """发送邮件（模拟）"""
    time.sleep(2)
    print(f"邮件已发送至 {email}: {message}")

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

@app.post("/send-notification/{email}")
def send_notification(
    email: str,
    background_tasks: BackgroundTasks
):
    """发送通知（后台执行）"""
    background_tasks.add_task(send_email, email, "感谢您的注册！")
    background_tasks.add_task(write_log, f"发送邮件至 {email}")
    
    return {"message": "通知将在后台发送"}

10. 中间件

        
        
        
    
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time

app = FastAPI()

# 使用装饰器创建中间件
@app.middleware("http")
async def add_process_time_header(request: 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

# 使用类创建中间件
class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        print(f"请求: {request.method} {request.url}")
        response = await call_next(request)
        print(f"响应: {response.status_code}")
        return response

app.add_middleware(LoggingMiddleware)

# CORS 中间件
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

11. 小结

特性	FastAPI 实现	说明
路由	`@app.get()` 等	方法装饰器
数据验证	Pydantic 模型	类型提示驱动
自动文档	`/docs`, `/redoc`	OpenAPI 自动生成
依赖注入	`Depends()`	函数/类依赖
异步	`async def`	原生 ASGI 支持
后台任务	`BackgroundTasks`	非阻塞执行
中间件	`@app.middleware()`	请求处理管道

12. 练习题

使用 FastAPI 创建一个完整的 TODO API，包含 CRUD 和过滤功能。
为上述 API 添加 JWT 认证（使用 fastapi-users 或手写）。
使用 Pydantic 模型实现复杂的数据验证（如邮箱格式、密码强度）。
对比 FastAPI 与 Flask/Django 的相同功能实现，总结各自优势。

下节预告：我们将对比 Django、Flask 和 FastAPI 三个主流框架，分析各自适用场景，帮助你做出技术选型决策。

目录

Python FastAPI 框架入门

1. FastAPI 简介

1.1 核心特性

1.2 安装

2. 快速开始

2.1 最小应用

2.2 运行服务

2.3 自动文档

3. 路径操作

3.1 HTTP 方法

3.2 路径参数

4. 查询参数与请求体

4.1 查询参数

4.2 Pydantic 模型（请求体）

5. 响应处理

5.1 自定义响应

6. 依赖注入

6.1 基础依赖

6.2 类依赖

6.3 数据库依赖

7. 错误处理

8. 异步支持

9. 后台任务

10. 中间件

11. 小结

12. 练习题