路由与请求处理

第一章：路由系统深入理解

APIRouter 模块化路由

在大型项目中，将所有路由定义在一个文件中会导致代码难以维护。FastAPI 提供了 APIRouter 来实现路由模块化。

# app/api/v1/endpoints/users.py
from fastapi import APIRouter, HTTPException, status
from typing import List
from pydantic import BaseModel

router = APIRouter(
    prefix="/users",  # 路由前缀
    tags=["users"],  # 文档标签
    responses={404: {"description": "用户不存在"}}  # 通用响应
)

class User(BaseModel):
    id: int
    username: str
    email: str

# 模拟数据库
fake_users_db = {}

@router.get("/", response_model=List[User])
async def list_users():
    """获取所有用户列表"""
    return list(fake_users_db.values())

@router.get("/{user_id}", response_model=User)
async def get_user(user_id: int):
    """根据 ID 获取用户"""
    if user_id not in fake_users_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    return fake_users_db[user_id]

@router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)
async def create_user(user: User):
    """创建新用户"""
    if user.id in fake_users_db:
        raise HTTPException(status_code=400, detail="用户已存在")
    fake_users_db[user.id] = user
    return user

@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: int):
    """删除用户"""
    if user_id not in fake_users_db:
        raise HTTPException(status_code=404, detail="用户不存在")
    del fake_users_db[user_id]

# app/api/v1/endpoints/items.py
from fastapi import APIRouter
from typing import Optional

router = APIRouter(
    prefix="/items",
    tags=["items"]
)

@router.get("/")
async def list_items(
    skip: int = 0,
    limit: int = 10,
    search: Optional[str] = None
):
    """获取商品列表"""
    return {"skip": skip, "limit": limit, "search": search}

@router.get("/{item_id}")
async def get_item(item_id: int):
    """获取单个商品"""
    return {"item_id": item_id}

# app/api/v1/router.py
from fastapi import APIRouter
from app.api.v1.endpoints import users, items

api_router = APIRouter()
api_router.include_router(users.router)
api_router.include_router(items.router)

# app/main.py
from fastapi import FastAPI
from app.api.v1.router import api_router

app = FastAPI(title="我的应用")

# 包含 v1 版本的所有路由
app.include_router(api_router, prefix="/api/v1")

# 也可以直接在 app 上定义路由
@app.get("/health")
async def health_check():
    return {"status": "healthy"}

路由优先级和顺序

FastAPI 按照定义顺序匹配路由，需要注意固定路由应该放在动态路由之前：

from fastapi import FastAPI

app = FastAPI()

# 正确的顺序
@app.get("/users/me")  # 固定路由优先
async def get_current_user():
    return {"user": "current"}

@app.get("/users/{user_id}")  # 动态路由在后
async def get_user(user_id: str):
    return {"user_id": user_id}

# 错误的顺序会导致 /users/me 永远匹配不到
# @app.get("/users/{user_id}")  # 这会先匹配
# @app.get("/users/me")  # 这永远不会被访问

路由回调和钩子

from fastapi import APIRouter, Request, Response
from typing import Callable

# 创建带回调的路由
def create_router_with_callbacks(
    on_startup: Callable = None,
    on_shutdown: Callable = None
):
    router = APIRouter()

    # 路由级别的中间件
    @router.middleware("http")
    async def log_requests(request: Request, call_next):
        print(f"Request: {request.method} {request.url}")
        response = await call_next(request)
        print(f"Response: {response.status_code}")
        return response

    return router

router = create_router_with_callbacks()

@router.get("/")
async def root():
    return {"message": "Hello"}

第二章：请求数据处理

Header 参数

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

app = FastAPI()

@app.get("/items/")
async def read_items(
    # 标准 header（自动转换下划线为连字符）
    user_agent: Optional[str] = Header(None),
    # 自定义 header
    x_token: Optional[str] = Header(None, alias="X-Token"),
    # 禁用自动转换
    accept_encoding: Optional[str] = Header(None, convert_underscores=False)
):
    return {
        "User-Agent": user_agent,
        "X-Token": x_token,
        "Accept-Encoding": accept_encoding
    }

# 重复的 header
@app.get("/items-multi-header/")
async def read_items_multi_header(
    x_token: List[str] = Header(None)
):
    return {"X-Token values": x_token}

Cookie 参数

from fastapi import FastAPI, Cookie
from typing import Optional

app = FastAPI()

@app.get("/items/")
async def read_items(
    session_id: Optional[str] = Cookie(None),
    tracking_id: Optional[str] = Cookie(None, alias="tracking-id")
):
    return {
        "session_id": session_id,
        "tracking_id": tracking_id
    }

表单数据处理

from fastapi import FastAPI, Form, File, UploadFile
from typing import List

app = FastAPI()

# 安装依赖：pip install python-multipart

# 基础表单
@app.post("/login/")
async def login(
    username: str = Form(...),
    password: str = Form(...)
):
    return {"username": username}

# 文件上传
@app.post("/upload/")
async def upload_file(
    file: UploadFile = File(...)
):
    contents = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "size": len(contents)
    }

# 多文件上传
@app.post("/upload-multiple/")
async def upload_files(
    files: List[UploadFile] = File(...)
):
    return {
        "filenames": [f.filename for f in files],
        "count": len(files)
    }

# 表单 + 文件混合
@app.post("/create-with-file/")
async def create_with_file(
    name: str = Form(...),
    description: str = Form(None),
    file: UploadFile = File(...)
):
    return {
        "name": name,
        "description": description,
        "filename": file.filename
    }

# 大文件分块处理
@app.post("/upload-large/")
async def upload_large_file(file: UploadFile = File(...)):
    total_size = 0
    chunk_size = 1024 * 1024  # 1MB

    while True:
        chunk = await file.read(chunk_size)
        if not chunk:
            break
        total_size += len(chunk)
        # 处理每个块...

    return {"total_size": total_size}

Request 对象直接访问

from fastapi import FastAPI, Request
from typing import Dict, Any

app = FastAPI()

@app.post("/items/")
async def create_item(request: Request):
    # 获取原始请求体
    body = await request.body()

    # 获取 JSON 数据
    json_data = await request.json()

    # 获取表单数据
    # form_data = await request.form()

    # 获取查询参数
    query_params = dict(request.query_params)

    # 获取路径参数
    path_params = dict(request.path_params)

    # 获取 headers
    headers = dict(request.headers)

    # 获取 cookies
    cookies = dict(request.cookies)

    # 获取客户端信息
    client_host = request.client.host
    client_port = request.client.port

    return {
        "url": str(request.url),
        "method": request.method,
        "client": f"{client_host}:{client_port}",
        "query_params": query_params
    }

# 获取请求 URL 各部分
@app.get("/url-info/")
async def url_info(request: Request):
    return {
        "full_url": str(request.url),
        "scheme": request.url.scheme,  # http 或 https
        "host": request.url.hostname,
        "port": request.url.port,
        "path": request.url.path,
        "query": request.url.query,
        "fragment": request.url.fragment
    }

第三章：复杂请求体处理

嵌套模型

from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl
from typing import Optional, List, Set
from datetime import datetime

app = FastAPI()

# 地址模型
class Address(BaseModel):
    street: str
    city: str
    state: str
    country: str = "中国"
    zip_code: Optional[str] = None

# 图片模型
class Image(BaseModel):
    url: HttpUrl
    name: str
    description: Optional[str] = None

# 商品模型（嵌套）
class Item(BaseModel):
    name: str
    description: Optional[str] = None
    price: float
    tax: Optional[float] = None
    tags: Set[str] = set()  # 使用 Set 去重
    images: Optional[List[Image]] = None

# 订单模型（深度嵌套）
class OrderItem(BaseModel):
    item: Item
    quantity: int

class Order(BaseModel):
    order_id: str
    customer_name: str
    shipping_address: Address
    billing_address: Optional[Address] = None
    items: List[OrderItem]
    created_at: datetime = None
    notes: Optional[str] = None

@app.post("/orders/")
async def create_order(order: Order):
    # 如果账单地址未提供，使用发货地址
    if order.billing_address is None:
        order.billing_address = order.shipping_address

    # 计算总价
    total = sum(
        item.item.price * item.quantity
        for item in order.items
    )

    return {
        "order_id": order.order_id,
        "total": total,
        "items_count": len(order.items)
    }

纯列表请求体

from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

app = FastAPI()

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

# 接收列表作为请求体
@app.post("/items/bulk/")
async def create_items(items: List[Item]):
    return {
        "count": len(items),
        "items": items
    }

# 接收任意字典
@app.post("/data/")
async def process_data(data: dict):
    return {"received_keys": list(data.keys())}

多种内容类型支持

from fastapi import FastAPI, Request, Body
from pydantic import BaseModel
from typing import Union
import json

app = FastAPI()

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

# 支持多种内容类型
@app.post("/items/")
async def create_item(request: Request):
    content_type = request.headers.get("content-type", "")

    if "application/json" in content_type:
        data = await request.json()
    elif "application/x-www-form-urlencoded" in content_type:
        form = await request.form()
        data = dict(form)
    elif "multipart/form-data" in content_type:
        form = await request.form()
        data = dict(form)
    else:
        body = await request.body()
        data = {"raw": body.decode()}

    return {"content_type": content_type, "data": data}

# 嵌入式请求体参数
@app.post("/items-embedded/")
async def create_item_embedded(
    item: Item = Body(..., embed=True)  # {"item": {...}}
):
    return item

额外的请求体字段

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

app = FastAPI()

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

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),
    notes: Optional[str] = Body(None, max_length=500)
):
    """
    请求体格式：
    {
        "item": {"name": "xxx", "price": 100},
        "user": {"username": "xxx", "email": "xxx@example.com"},
        "importance": 3,
        "notes": "一些备注"
    }
    """
    return {
        "item_id": item_id,
        "item": item,
        "user": user,
        "importance": importance,
        "notes": notes
    }

第四章：参数验证进阶

自定义验证器

from fastapi import FastAPI, Query, Path
from pydantic import BaseModel, field_validator, model_validator
from typing import Optional
import re

app = FastAPI()

class UserCreate(BaseModel):
    username: str
    email: str
    password: str
    password_confirm: str

    @field_validator("username")
    @classmethod
    def username_alphanumeric(cls, v):
        if not v.isalnum():
            raise ValueError("用户名必须是字母和数字")
        if len(v) < 3:
            raise ValueError("用户名至少3个字符")
        return v

    @field_validator("email")
    @classmethod
    def email_valid(cls, v):
        pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
        if not re.match(pattern, v):
            raise ValueError("邮箱格式不正确")
        return v.lower()

    @field_validator("password")
    @classmethod
    def password_strength(cls, v):
        if len(v) < 8:
            raise ValueError("密码至少8个字符")
        if not re.search(r"[A-Z]", v):
            raise ValueError("密码必须包含大写字母")
        if not re.search(r"[a-z]", v):
            raise ValueError("密码必须包含小写字母")
        if not re.search(r"\d", v):
            raise ValueError("密码必须包含数字")
        return v

    @model_validator(mode='after')
    def passwords_match(self):
        if self.password != self.password_confirm:
            raise ValueError("两次密码输入不一致")
        return self

@app.post("/users/")
async def create_user(user: UserCreate):
    return {"username": user.username, "email": user.email}

高级 Query 参数

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

app = FastAPI()

class SortOrder(str, Enum):
    asc = "asc"
    desc = "desc"

class Category(str, Enum):
    electronics = "electronics"
    clothing = "clothing"
    books = "books"

@app.get("/products/")
async def list_products(
    # 必填参数，带正则验证
    q: str = Query(
        ...,
        min_length=2,
        max_length=100,
        regex="^[a-zA-Z0-9\s]+$",
        title="搜索关键词",
        description="产品搜索关键词"
    ),
    # 分页参数
    page: int = Query(1, ge=1, le=1000, description="页码"),
    page_size: int = Query(20, ge=1, le=100, description="每页数量"),
    # 排序参数
    sort_by: str = Query("created_at", regex="^(created_at|price|name)$"),
    sort_order: SortOrder = Query(SortOrder.desc),
    # 过滤参数
    category: Optional[Category] = Query(None),
    min_price: Optional[float] = Query(None, ge=0),
    max_price: Optional[float] = Query(None, ge=0),
    # 多选参数
    tags: List[str] = Query([], description="标签过滤"),
    # 布尔参数
    in_stock: bool = Query(True, description="只显示有货商品")
):
    # 验证价格范围
    if min_price and max_price and min_price > max_price:
        from fastapi import HTTPException
        raise HTTPException(400, "最低价格不能高于最高价格")

    return {
        "q": q,
        "page": page,
        "page_size": page_size,
        "sort": f"{sort_by} {sort_order.value}",
        "filters": {
            "category": category,
            "price_range": [min_price, max_price],
            "tags": tags,
            "in_stock": in_stock
        }
    }

参数别名和弃用

from fastapi import FastAPI, Query
from typing import Optional

app = FastAPI()

@app.get("/items/")
async def read_items(
    # 使用别名处理特殊字符
    item_query: Optional[str] = Query(
        None,
        alias="item-query",  # 实际 URL 中使用 ?item-query=xxx
        title="商品查询"
    ),
    # 弃用参数（仍然有效，但在文档中标记为弃用）
    old_filter: Optional[str] = Query(
        None,
        deprecated=True,
        description="已弃用，请使用 new_filter"
    ),
    new_filter: Optional[str] = Query(None)
):
    return {
        "item_query": item_query,
        "filter": new_filter or old_filter
    }

第五章：路由高级配置

路由依赖和回调

from fastapi import FastAPI, APIRouter, Depends, HTTPException
from typing import Callable

app = FastAPI()

# 公共依赖
async def verify_token(token: str = None):
    if not token:
        raise HTTPException(status_code=401, detail="未授权")
    return token

async def verify_admin(token: str = Depends(verify_token)):
    # 模拟管理员验证
    if token != "admin-token":
        raise HTTPException(status_code=403, detail="需要管理员权限")
    return token

# 带依赖的路由器
admin_router = APIRouter(
    prefix="/admin",
    tags=["admin"],
    dependencies=[Depends(verify_admin)],  # 所有路由都需要管理员权限
    responses={
        401: {"description": "未授权"},
        403: {"description": "权限不足"}
    }
)

@admin_router.get("/dashboard")
async def admin_dashboard():
    return {"message": "管理员仪表盘"}

@admin_router.get("/users")
async def admin_users():
    return {"users": []}

# 公开路由器（无依赖）
public_router = APIRouter(prefix="/public", tags=["public"])

@public_router.get("/info")
async def public_info():
    return {"message": "公开信息"}

app.include_router(admin_router)
app.include_router(public_router)

路由回调（Callbacks）

from fastapi import FastAPI, APIRouter
from pydantic import BaseModel, HttpUrl
from typing import Optional

app = FastAPI()

# 定义回调模型
class Invoice(BaseModel):
    id: str
    amount: float
    customer: str

class InvoiceEvent(BaseModel):
    event_type: str
    invoice: Invoice

class InvoiceCallback(BaseModel):
    callback_url: HttpUrl

# 回调路由（文档用途）
invoices_callback_router = APIRouter()

@invoices_callback_router.post(
    "{$callback_url}",
    response_model=dict
)
async def invoice_callback(body: InvoiceEvent):
    """
    当发票状态变化时，系统会调用此回调 URL
    """
    pass

# 主路由，关联回调
@app.post(
    "/invoices/",
    callbacks=invoices_callback_router.routes
)
async def create_invoice(
    invoice: Invoice,
    callback: Optional[InvoiceCallback] = None
):
    """
    创建发票。

    如果提供了 callback_url，当发票状态变化时会通知该 URL。
    """
    return {"id": invoice.id, "status": "created"}

路由装饰器选项

from fastapi import FastAPI, Response
from pydantic import BaseModel
from typing import Union

app = FastAPI()

class Item(BaseModel):
    name: str

class Message(BaseModel):
    message: str

@app.post(
    "/items/",
    # 响应相关
    response_model=Item,
    response_model_exclude_unset=True,
    response_model_exclude_none=True,
    response_model_exclude={"internal_id"},
    response_model_include={"name", "price"},

    # 状态码
    status_code=201,

    # 文档相关
    summary="创建商品",
    description="创建一个新的商品记录",
    response_description="创建成功返回商品信息",
    tags=["items"],

    # 文档示例
    responses={
        201: {
            "description": "创建成功",
            "content": {
                "application/json": {
                    "example": {"name": "示例商品", "price": 99.99}
                }
            }
        },
        400: {
            "model": Message,
            "description": "无效请求"
        },
        422: {
            "description": "验证错误"
        }
    },

    # 其他
    deprecated=False,
    operation_id="create_item_v1",
    include_in_schema=True
)
async def create_item(item: Item):
    return item

动态路由生成

from fastapi import FastAPI, APIRouter
from typing import Type, Callable
from pydantic import BaseModel

app = FastAPI()

# 通用 CRUD 路由生成器
def create_crud_router(
    model: Type[BaseModel],
    prefix: str,
    tags: list = None
) -> APIRouter:
    router = APIRouter(prefix=prefix, tags=tags or [prefix.strip("/")])

    # 存储数据
    items = {}

    @router.get("/")
    async def list_items():
        return list(items.values())

    @router.get("/{item_id}")
    async def get_item(item_id: int):
        if item_id not in items:
            from fastapi import HTTPException
            raise HTTPException(404, "Item not found")
        return items[item_id]

    @router.post("/")
    async def create_item(item: model):
        item_id = len(items) + 1
        items[item_id] = item
        return {"id": item_id, **item.model_dump()}

    @router.delete("/{item_id}")
    async def delete_item(item_id: int):
        if item_id in items:
            del items[item_id]
        return {"status": "deleted"}

    return router

# 定义模型
class Product(BaseModel):
    name: str
    price: float

class Category(BaseModel):
    name: str
    description: str = None

# 动态生成路由
app.include_router(create_crud_router(Product, "/products", ["products"]))
app.include_router(create_crud_router(Category, "/categories", ["categories"]))

第六章：错误处理

内置异常处理

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id < 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Item ID must be positive"
        )
    if item_id == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Item not found",
            headers={"X-Error": "Item not found"}  # 自定义响应头
        )
    return {"item_id": item_id}

自定义异常类

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from typing import Optional

app = FastAPI()

# 自定义异常基类
class AppException(Exception):
    def __init__(
        self,
        message: str,
        code: str = "UNKNOWN_ERROR",
        status_code: int = 500,
        details: Optional[dict] = None
    ):
        self.message = message
        self.code = code
        self.status_code = status_code
        self.details = details or {}

# 具体异常类
class NotFoundError(AppException):
    def __init__(self, resource: str, resource_id: any):
        super().__init__(
            message=f"{resource} with id {resource_id} not found",
            code="NOT_FOUND",
            status_code=404,
            details={"resource": resource, "id": resource_id}
        )

class ValidationError(AppException):
    def __init__(self, field: str, message: str):
        super().__init__(
            message=f"Validation error: {message}",
            code="VALIDATION_ERROR",
            status_code=422,
            details={"field": field, "error": message}
        )

class AuthenticationError(AppException):
    def __init__(self, message: str = "Authentication required"):
        super().__init__(
            message=message,
            code="AUTHENTICATION_ERROR",
            status_code=401
        )

class PermissionError(AppException):
    def __init__(self, action: str):
        super().__init__(
            message=f"Permission denied for action: {action}",
            code="PERMISSION_DENIED",
            status_code=403,
            details={"action": action}
        )

# 注册异常处理器
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "error": {
                "code": exc.code,
                "message": exc.message,
                "details": exc.details
            }
        }
    )

# 使用示例
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # 模拟用户不存在
    if user_id == 404:
        raise NotFoundError("User", user_id)
    return {"user_id": user_id}

@app.post("/login")
async def login(username: str, password: str):
    if not username:
        raise ValidationError("username", "用户名不能为空")
    if password != "secret":
        raise AuthenticationError("用户名或密码错误")
    return {"token": "xxx"}

全局异常处理

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
import traceback
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

# 处理 HTTP 异常
@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "error": {
                "code": f"HTTP_{exc.status_code}",
                "message": exc.detail
            }
        }
    )

# 处理验证错误
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })

    return JSONResponse(
        status_code=422,
        content={
            "success": False,
            "error": {
                "code": "VALIDATION_ERROR",
                "message": "请求参数验证失败",
                "details": errors
            }
        }
    )

# 处理所有未捕获的异常
@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    # 记录错误日志
    logger.error(
        f"Unhandled exception: {exc}\n"
        f"Request: {request.method} {request.url}\n"
        f"Traceback: {traceback.format_exc()}"
    )

    return JSONResponse(
        status_code=500,
        content={
            "success": False,
            "error": {
                "code": "INTERNAL_ERROR",
                "message": "服务器内部错误"
            }
        }
    )

常见问题

Q1：如何处理文件上传的大小限制？

from fastapi import FastAPI, File, UploadFile, HTTPException

app = FastAPI()

MAX_FILE_SIZE = 10 * 1024 * 1024  # 10MB

@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
    # 检查文件大小
    contents = await file.read()
    if len(contents) > MAX_FILE_SIZE:
        raise HTTPException(400, f"文件大小不能超过 {MAX_FILE_SIZE // 1024 // 1024}MB")

    # 重置文件指针以便后续使用
    await file.seek(0)

    return {"size": len(contents)}

Q2：如何实现请求超时？

import asyncio
from fastapi import FastAPI, HTTPException

app = FastAPI()

async def long_running_task():
    await asyncio.sleep(10)
    return "完成"

@app.get("/slow/")
async def slow_endpoint():
    try:
        result = await asyncio.wait_for(
            long_running_task(),
            timeout=5.0
        )
        return {"result": result}
    except asyncio.TimeoutError:
        raise HTTPException(408, "请求超时")

Q3：如何处理跨域请求？

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

app = FastAPI()

# 配置 CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "https://example.com"],
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
    allow_headers=["*"],
    expose_headers=["X-Custom-Header"],
    max_age=600  # 预检请求缓存时间
)

学习资源

1. FastAPI 路由文档：https://fastapi.tiangolo.com/tutorial/first-steps/
2. Starlette 路由：https://www.starlette.io/routing/
3. OpenAPI 规范：https://swagger.io/specification/