FastAPI 基础入门

第一章：FastAPI 简介

什么是 FastAPI？

FastAPI 是一个现代、快速（高性能）的 Web 框架，用于基于 Python 3.7+ 构建 API，基于标准的 Python 类型提示。它由 Sebastián Ramírez 于 2018 年创建，迅速成为 Python 生态中最受欢迎的 Web 框架之一。

核心特性

1. 极致性能：FastAPI 基于 Starlette（用于 Web 部分）和 Pydantic（用于数据部分），是最快的 Python 框架之一，性能可与 NodeJS 和 Go 相媲美。

2. 开发效率：

   • 自动生成交互式 API 文档（Swagger UI 和 ReDoc）
   • 编辑器智能提示支持完善
   • 代码重复最小化

3. 类型安全：

   • 基于 Python 类型提示
   • 自动数据验证
   • 自动数据序列化

4. 标准兼容：

   • 完全兼容 OpenAPI（以前称为 Swagger）
   • 完全兼容 JSON Schema
   • 基于 ASGI 标准

FastAPI vs 其他框架

特性	FastAPI	Flask	Django	Express.js
异步支持	✅ 原生	❌ 需扩展	⚠️ 部分	✅ 原生
类型检查	✅ 内置	❌ 无	❌ 无	❌ 无
自动文档	✅ 内置	❌ 需扩展	❌ 需扩展	❌ 需扩展
数据验证	✅ 内置	❌ 需扩展	⚠️ 部分	❌ 需扩展
性能	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐	⭐⭐⭐⭐
学习曲线	中等	简单	较陡	简单

适用场景

• RESTful API 开发：构建高性能的 REST API
• 微服务架构：轻量级，启动快速
• 实时应用：WebSocket 支持
• 机器学习服务：与 ML 库完美集成
• 数据处理管道：异步处理大量数据

第二章：环境搭建与项目初始化

环境要求

• Python 3.7+（推荐 3.10+）
• pip 或 poetry 包管理器
• 虚拟环境（推荐）

安装 FastAPI

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Windows
.\venv\Scripts\activate
# Linux/macOS
source venv/bin/activate

# 安装 FastAPI 和 ASGI 服务器
pip install fastapi
pip install "uvicorn[standard]"

使用 Poetry 管理项目（推荐）

# 安装 Poetry
pip install poetry

# 创建新项目
poetry new my-fastapi-project
cd my-fastapi-project

# 添加依赖
poetry add fastapi
poetry add uvicorn[standard]

第一个 FastAPI 应用

创建 main.py 文件：

from fastapi import FastAPI

# 创建 FastAPI 实例
app = FastAPI(
    title="我的第一个 FastAPI 应用",
    description="这是一个示例 API",
    version="1.0.0"
)

# 定义根路由
@app.get("/")
async def root():
    """根路径端点"""
    return {"message": "Hello, FastAPI!"}

# 定义带参数的路由
@app.get("/hello/{name}")
async def say_hello(name: str):
    """向指定用户问好"""
    return {"message": f"Hello, {name}!"}

# 带查询参数的路由
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    """获取项目列表，支持分页"""
    return {"skip": skip, "limit": limit}

启动应用

# 开发模式启动（支持热重载）
uvicorn main:app --reload

# 生产模式启动
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

访问以下地址：

• API 根路径：http://127.0.0.1:8000
• Swagger UI 文档：http://127.0.0.1:8000/docs
• ReDoc 文档：http://127.0.0.1:8000/redoc
• OpenAPI JSON：http://127.0.0.1:8000/openapi.json

项目结构推荐

my-fastapi-project/
├── app/
│   ├── __init__.py
│   ├── main.py              # 应用入口
│   ├── config.py            # 配置管理
│   ├── dependencies.py      # 依赖注入
│   ├── api/
│   │   ├── __init__.py
│   │   ├── v1/
│   │   │   ├── __init__.py
│   │   │   ├── endpoints/
│   │   │   │   ├── __init__.py
│   │   │   │   ├── users.py
│   │   │   │   └── items.py
│   │   │   └── router.py
│   │   └── deps.py          # API 依赖
│   ├── core/
│   │   ├── __init__.py
│   │   ├── security.py      # 安全相关
│   │   └── config.py        # 核心配置
│   ├── models/
│   │   ├── __init__.py
│   │   ├── user.py          # 数据库模型
│   │   └── item.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── user.py          # Pydantic 模型
│   │   └── item.py
│   ├── crud/
│   │   ├── __init__.py
│   │   ├── base.py          # CRUD 基类
│   │   ├── user.py
│   │   └── item.py
│   ├── db/
│   │   ├── __init__.py
│   │   ├── base.py          # 数据库基础
│   │   └── session.py       # 数据库会话
│   └── utils/
│       ├── __init__.py
│       └── helpers.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py          # pytest 配置
│   └── test_api/
├── alembic/                  # 数据库迁移
│   └── versions/
├── .env                      # 环境变量
├── .gitignore
├── requirements.txt
├── pyproject.toml
└── README.md

第三章：HTTP 方法与基础路由

HTTP 方法概览

FastAPI 支持所有标准 HTTP 方法：

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items():
    """GET - 获取资源"""
    return {"method": "GET"}

@app.post("/items/")
async def create_item():
    """POST - 创建资源"""
    return {"method": "POST"}

@app.put("/items/{item_id}")
async def update_item(item_id: int):
    """PUT - 完整更新资源"""
    return {"method": "PUT", "item_id": item_id}

@app.patch("/items/{item_id}")
async def partial_update_item(item_id: int):
    """PATCH - 部分更新资源"""
    return {"method": "PATCH", "item_id": item_id}

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    """DELETE - 删除资源"""
    return {"method": "DELETE", "item_id": item_id}

@app.options("/items/")
async def options_items():
    """OPTIONS - 获取支持的方法"""
    return {"method": "OPTIONS"}

@app.head("/items/")
async def head_items():
    """HEAD - 获取响应头"""
    return {"method": "HEAD"}

路径参数

from fastapi import FastAPI, Path
from enum import Enum

app = FastAPI()

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

# 带验证的路径参数
@app.get("/items/{item_id}")
async def get_item(
    item_id: int = Path(
        ...,
        title="项目ID",
        description="要获取的项目的唯一标识符",
        ge=1,  # 大于等于 1
        le=1000  # 小于等于 1000
    )
):
    return {"item_id": item_id}

# 枚举路径参数
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 == ModelName.lenet:
        return {"model_name": model_name, "message": "LeCNN all the images"}
    return {"model_name": model_name, "message": "Have some residuals"}

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

查询参数

from fastapi import FastAPI, Query
from typing import Optional, List

app = FastAPI()

# 基础查询参数
@app.get("/items/")
async def read_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

# 可选查询参数
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: Optional[str] = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

# 带验证的查询参数
@app.get("/search/")
async def search_items(
    q: str = Query(
        ...,  # 必填
        min_length=3,
        max_length=50,
        regex="^[a-zA-Z0-9]+$",
        title="搜索关键词",
        description="用于搜索的关键词，仅支持字母和数字"
    ),
    page: int = Query(1, ge=1, description="页码"),
    size: int = Query(10, ge=1, le=100, description="每页数量")
):
    return {"q": q, "page": page, "size": size}

# 列表查询参数
@app.get("/items/multi/")
async def read_items_multi(
    q: List[str] = Query(
        default=["foo", "bar"],
        title="查询参数列表"
    )
):
    return {"q": q}

# 弃用参数
@app.get("/legacy/")
async def legacy_endpoint(
    old_param: Optional[str] = Query(
        None,
        deprecated=True,
        description="此参数已弃用，请使用 new_param"
    ),
    new_param: Optional[str] = None
):
    return {"old_param": old_param, "new_param": new_param}

请求体

from fastapi import FastAPI, Body
from pydantic import BaseModel, Field
from typing import Optional

app = FastAPI()

# 定义请求体模型
class Item(BaseModel):
    name: str = Field(..., min_length=1, max_length=100, example="示例项目")
    description: Optional[str] = Field(None, max_length=500)
    price: float = Field(..., gt=0, description="价格必须大于0")
    tax: Optional[float] = Field(None, ge=0)

    class Config:
        json_schema_extra = {
            "example": {
                "name": "示例商品",
                "description": "这是一个示例商品的描述",
                "price": 99.99,
                "tax": 10.0
            }
        }

# 单个请求体
@app.post("/items/")
async def create_item(item: Item):
    return {"item": item, "total": item.price + (item.tax or 0)}

# 多个请求体参数
class User(BaseModel):
    username: str
    email: str

@app.put("/items/{item_id}")
async def update_item(
    item_id: int,
    item: Item,
    user: User,
    importance: int = Body(..., ge=1, le=5)
):
    return {
        "item_id": item_id,
        "item": item,
        "user": user,
        "importance": importance
    }

# 嵌套请求体
class Image(BaseModel):
    url: str
    name: str

class ItemWithImage(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    images: Optional[List[Image]] = None

@app.post("/items-with-images/")
async def create_item_with_images(item: ItemWithImage):
    return item

第四章：响应处理

响应模型

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from typing import Optional, List

app = FastAPI()

# 用户输入模型（包含密码）
class UserCreate(BaseModel):
    username: str
    email: EmailStr
    password: str

# 用户响应模型（不包含密码）
class UserResponse(BaseModel):
    id: int
    username: str
    email: EmailStr

    class Config:
        from_attributes = True  # Pydantic V2

# 使用 response_model 过滤输出
@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
    # 模拟创建用户
    fake_user = {
        "id": 1,
        "username": user.username,
        "email": user.email,
        "password": user.password  # 这个字段会被过滤掉
    }
    return fake_user

# 响应模型排除未设置的字段
class ItemResponse(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None

@app.get("/items/{item_id}", response_model=ItemResponse, response_model_exclude_unset=True)
async def read_item(item_id: int):
    return {"name": "Foo", "price": 50.2}

响应状态码

from fastapi import FastAPI, status
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str

# 指定状态码
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    return item

# 使用数字状态码
@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
    return None

# 常用状态码
# HTTP_200_OK = 200
# HTTP_201_CREATED = 201
# HTTP_204_NO_CONTENT = 204
# HTTP_400_BAD_REQUEST = 400
# HTTP_401_UNAUTHORIZED = 401
# HTTP_403_FORBIDDEN = 403
# HTTP_404_NOT_FOUND = 404
# HTTP_422_UNPROCESSABLE_ENTITY = 422
# HTTP_500_INTERNAL_SERVER_ERROR = 500

自定义响应

from fastapi import FastAPI
from fastapi.responses import (
    JSONResponse,
    HTMLResponse,
    PlainTextResponse,
    RedirectResponse,
    StreamingResponse,
    FileResponse
)
from typing import Any
import json

app = FastAPI()

# JSON 响应
@app.get("/json/")
async def get_json():
    return JSONResponse(
        content={"message": "Hello"},
        status_code=200,
        headers={"X-Custom-Header": "custom-value"}
    )

# HTML 响应
@app.get("/html/", response_class=HTMLResponse)
async def get_html():
    return """
    <html>
        <head><title>Hello</title></head>
        <body><h1>Hello, World!</h1></body>
    </html>
    """

# 纯文本响应
@app.get("/text/", response_class=PlainTextResponse)
async def get_text():
    return "Hello, World!"

# 重定向响应
@app.get("/redirect/")
async def redirect():
    return RedirectResponse(url="/json/", status_code=302)

# 流式响应
async def generate_data():
    for i in range(10):
        yield f"data: {i}\n\n"

@app.get("/stream/")
async def stream():
    return StreamingResponse(generate_data(), media_type="text/event-stream")

# 文件响应
@app.get("/file/")
async def get_file():
    return FileResponse(
        path="./files/example.pdf",
        filename="download.pdf",
        media_type="application/pdf"
    )

# 自定义 JSON 编码器
class CustomJSONResponse(JSONResponse):
    def render(self, content: Any) -> bytes:
        return json.dumps(
            content,
            ensure_ascii=False,
            indent=2
        ).encode("utf-8")

@app.get("/custom-json/", response_class=CustomJSONResponse)
async def custom_json():
    return {"message": "你好，世界！", "emoji": "🚀"}

响应头和 Cookie

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

app = FastAPI()

# 设置响应头
@app.get("/headers/")
async def set_headers(response: Response):
    response.headers["X-Custom-Header"] = "custom-value"
    response.headers["X-Process-Time"] = "0.5"
    return {"message": "Check the headers!"}

# 设置 Cookie
@app.get("/set-cookie/")
async def set_cookie(response: Response):
    response.set_cookie(
        key="session_id",
        value="abc123",
        max_age=3600,  # 1小时
        httponly=True,
        secure=True,
        samesite="lax"
    )
    return {"message": "Cookie has been set"}

# 删除 Cookie
@app.get("/delete-cookie/")
async def delete_cookie(response: Response):
    response.delete_cookie(key="session_id")
    return {"message": "Cookie has been deleted"}

# 使用 JSONResponse 设置头和 Cookie
@app.get("/combined/")
async def combined():
    content = {"message": "Hello"}
    response = JSONResponse(content=content)
    response.headers["X-Custom"] = "value"
    response.set_cookie(key="token", value="xyz789")
    return response

第五章：自动文档与元数据

配置 API 元数据

from fastapi import FastAPI

# 详细的 API 元数据配置
app = FastAPI(
    title="我的 FastAPI 应用",
    description="""
# 欢迎使用我的 API

这是一个功能强大的 API，提供以下功能：

## 用户管理
* 创建用户
* 查询用户
* 更新用户信息

## 商品管理
* 商品 CRUD 操作
* 库存管理
""",
    version="2.0.0",
    terms_of_service="https://example.com/terms/",
    contact={
        "name": "API 支持团队",
        "url": "https://example.com/support",
        "email": "support@example.com"
    },
    license_info={
        "name": "MIT",
        "url": "https://opensource.org/licenses/MIT"
    },
    openapi_url="/api/v1/openapi.json",  # OpenAPI schema 路径
    docs_url="/docs",  # Swagger UI 路径
    redoc_url="/redoc"  # ReDoc 路径
)

标签分组

from fastapi import FastAPI

# 定义标签元数据
tags_metadata = [
    {
        "name": "users",
        "description": "用户相关操作。包括注册、登录、信息管理等。",
        "externalDocs": {
            "description": "用户文档",
            "url": "https://example.com/docs/users"
        }
    },
    {
        "name": "items",
        "description": "商品相关操作。",
    },
    {
        "name": "internal",
        "description": "内部 API，仅供内部使用。",
    }
]

app = FastAPI(openapi_tags=tags_metadata)

@app.get("/users/", tags=["users"])
async def get_users():
    return []

@app.post("/users/", tags=["users"])
async def create_user():
    return {}

@app.get("/items/", tags=["items"])
async def get_items():
    return []

@app.get("/internal/stats", tags=["internal"])
async def get_stats():
    return {"status": "ok"}

端点文档

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

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

@app.post(
    "/items/",
    summary="创建商品",
    description="""
创建一个新的商品，需要提供以下信息：

- **name**: 商品名称（必填）
- **price**: 商品价格（必填，大于0）
    """,
    response_description="返回创建的商品信息",
    tags=["items"],
    deprecated=False  # 标记为弃用
)
async def create_item(item: Item):
    """
    创建商品的详细说明：

    - 支持 JSON 格式的请求体
    - 返回创建的商品对象
    """
    return item

# 隐藏端点（不在文档中显示）
@app.get("/hidden/", include_in_schema=False)
async def hidden_endpoint():
    return {"message": "This endpoint is hidden from docs"}

禁用文档

from fastapi import FastAPI

# 完全禁用文档
app = FastAPI(
    docs_url=None,  # 禁用 Swagger UI
    redoc_url=None,  # 禁用 ReDoc
    openapi_url=None  # 禁用 OpenAPI schema
)

# 或者在生产环境禁用
import os

ENVIRONMENT = os.getenv("ENVIRONMENT", "development")

app = FastAPI(
    docs_url="/docs" if ENVIRONMENT == "development" else None,
    redoc_url="/redoc" if ENVIRONMENT == "development" else None
)

常见问题

Q1：FastAPI 和 Flask 有什么区别？

FastAPI 是异步优先的框架，内置类型验证和自动文档生成。Flask 是同步框架，更加轻量但需要更多扩展来实现相同功能。如果你需要高性能 API 和现代开发体验，选择 FastAPI；如果需要更多灵活性和成熟的生态，可以考虑 Flask。

Q2：如何处理跨域请求（CORS）？

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

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

Q3：FastAPI 支持 WebSocket 吗？

支持。FastAPI 完全支持 WebSocket：

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message: {data}")

Q4：如何在生产环境部署 FastAPI？

推荐使用 Gunicorn + Uvicorn workers：

gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

或者使用 Docker 部署，配合 Nginx 反向代理。

学习资源

1. 官方文档：https://fastapi.tiangolo.com/
2. GitHub 仓库：https://github.com/tiangolo/fastapi
3. Starlette 文档：https://www.starlette.io/
4. Pydantic 文档：https://docs.pydantic.dev/