1. 什么是 FastAPI？ #

FastAPI 是一个用于构建 Web API 的 Python 框架，特点是快、简单、自动生成文档。

你写几行 Python 代码，就能得到一个可以通过浏览器或程序访问的接口。比如你写一个「获取用户信息」的函数，FastAPI 会把它变成 GET /users/1 这样的网址，别人访问就能拿到数据。

打个比方：就像开一家「自助取餐窗口」——你定义好菜单（接口），顾客按编号点餐（发请求），窗口自动把菜端出来（返回数据）。FastAPI 帮你把「菜单」和「取餐流程」都标准化好了。

2. 前置知识：Web API 与 HTTP 方法 #

在学 FastAPI 之前，需要了解两个基础概念。

2.1 什么是 Web API？ #

API（Application Programming Interface）是程序之间通信的接口。Web API 就是通过网址（URL）来调用的接口。

例如：访问 https://api.example.com/users/1，服务器返回用户 1 的信息（通常是 JSON 格式）。你的程序发请求，服务器返回数据，这就是 Web API 的典型用法。

2.2 常用 HTTP 方法 #

方法	含义	常见用途
GET	获取	查询数据，如获取用户列表
POST	提交	创建数据，如新建用户
PUT	更新	整体替换，如修改用户信息
DELETE	删除	删除数据

FastAPI 用装饰器区分这些方法：@app.get()和@app.post()等。

2.3 HTTP 状态码与 JSON（简要） #

常见 HTTP 状态码含义：

状态码	含义	典型场景
200	成功	GET/PUT 成功返回数据
201	已创建	POST 创建资源成功
204	无内容	DELETE 成功且无返回体
400	请求错误	参数不合法
401	未认证	未登录或 Token 无效
403	禁止访问	无权限
404	未找到	资源不存在
422	无法处理	Pydantic 校验失败（FastAPI 自动返回）
500	服务器错误	程序内部异常

FastAPI 默认把字典、list、Pydantic 模型等序列化为 JSON，Content-Type 为 application/json。

---

3. 安装 #

FastAPI 需要配合 ASGI 服务器 才能运行，推荐使用 Uvicorn。

pip install fastapi uvicorn[standard]

若使用 uv 管理依赖：uv add fastapi "uvicorn[standard]"

3.1 ASGI、Starlette 与 FastAPI？ #

ASGI（Asynchronous Server Gateway Interface）是 Python 里服务 Web 应用的一种规范，支持异步、WebSocket 等，可以理解为「现代版的 WSGI」。

Uvicorn 是 ASGI 服务器：负责接收浏览器或客户端的 HTTP 请求，交给应用处理。

Starlette 是轻量级 ASGI 框架，提供路由、中间件、请求/响应对象等底层能力。

FastAPI 在 Starlette 之上构建，并深度集成 Pydantic，帮你自动生成文档、做数据校验。你可以只学 FastAPI；了解这一层关系有助于查官方文档或在 Starlette 文档里找「更底层」的行为说明。

4. 第一个 FastAPI 应用 #

# 导入 FastAPI 类
from fastapi import FastAPI
# 导入 uvicorn 用于启动服务器
import uvicorn
# 创建 FastAPI 应用实例，title 会显示在自动文档中
app = FastAPI(title="我的第一个 API")

# 使用 @app.get 定义 GET 请求的路由，"/" 表示根路径
@app.get("/")
def read_root():
    # 返回一个字典，FastAPI 会自动转为 JSON
    return {"message": "Hello, FastAPI!"}


# 定义 GET /items/{item_id}，item_id 是路径参数
@app.get("/items/{item_id}")
def read_item(item_id: int):
    # 路径参数会自动解析并做类型校验
    return {"item_id": item_id, "name": f"商品{item_id}"}


# 程序入口：直接运行此文件时执行
if __name__ == "__main__":

    # 启动服务器，host 和 port 为监听地址和端口
    uvicorn.run(app, host="127.0.0.1", port=8000)

运行方式：保存为 main.py，执行 python main.py，然后访问：

• http://127.0.0.1:8000 → 返回 {"message": "Hello, FastAPI!"}
• http://127.0.0.1:8000/items/1 → 返回 {"item_id": 1, "name": "商品1"}
• http://127.0.0.1:8000/docs → 自动生成的交互式文档

5. 路径参数与查询参数 #

5.1 路径参数 #

路径参数写在 URL 路径中，如 /items/123 里的 123。

# 路径参数：写在花括号 {param} 中，与函数参数同名
@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

5.2 查询参数 #

查询参数写在 URL 的 ? 后面，如 /items?skip=0&limit=10。

# 查询参数：有默认值的函数参数，未在路径中出现的即为查询参数
@app.get("/items")
def list_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

访问 http://127.0.0.1:8000/items?skip=5&limit=20 会得到 {"skip": 5, "limit": 20}。

5.3 查询参数校验：Query 与 Optional #

需要校验（如最小值、最大长度）或可选查询参数时，可用 Query()；可选类型用 typing.Optional。

# 导入 FastAPI、Query 与 Optional 类型
from fastapi import FastAPI, Query
from typing import Optional

# 创建应用
app = FastAPI()


# q 可选；skip 有默认值且必须 >= 0；limit 默认 10 且范围 1~100
@app.get("/search")
def search(
    q: Optional[str] = Query(None, max_length=50, description="关键词"),
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
):
    return {"q": q, "skip": skip, "limit": limit}


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

访问 /search?q=手机&skip=0&limit=20 合法；若 limit=200 会返回 422，并附带校验错误说明（在 /docs 里也能看到参数约束）。

---

6. 请求体与 Pydantic 模型 #

6.1 Pydantic 是什么？ #

Pydantic 是 FastAPI 用来做数据校验和序列化的库。你定义一个「模型」（类），指定每个字段的类型，Pydantic 会自动：

• 校验传入数据是否符合类型
• 把 Python 对象转成 JSON
• 生成 API 文档中的参数说明

6.2 定义模型并接收请求体 #

当客户端用 POST 发送 JSON 数据时，可以用 Pydantic 模型来接收和校验。

# 导入 FastAPI 和 Pydantic 的 BaseModel
from fastapi import FastAPI
from pydantic import BaseModel

# 创建应用
app = FastAPI()


# 定义请求体模型：继承 BaseModel，声明字段及类型
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = False


# POST 请求：item 会自动从请求体中解析，并做类型校验
@app.post("/items/")
def create_item(item: Item):
    return item


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

curl -i -X POST "http://127.0.0.1:8000/items/"  -H "Content-Type: application/json"   -d '{"name":"apple","price":12.5}'

6.3 字段校验 Field 与响应模型 response_model #

• Field：在模型字段上写 ge、le、min_length 等，增强校验与文档说明。
• response_model：指定返回给客户端的数据结构（可隐藏内部字段、只暴露安全字段）。

# 导入 FastAPI 与 Pydantic
from fastapi import FastAPI
from pydantic import BaseModel, Field

# 创建应用
app = FastAPI()


# 请求体：价格必须大于 0
class ItemIn(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0)


# 响应体：仅包含要暴露的字段
class ItemOut(BaseModel):
    name: str
    price: float
    id: int


@app.post("/items", response_model=ItemOut)
def create_item(item: ItemIn):
    # 仅示例：实际 id 来自数据库
    saved = ItemOut(id=1, name=item.name, price=item.price)
    return saved


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

路由上还可写 status_code=201，显式表示「已创建」（见7.1 同步与异步中的说明）。

---

7. 增删改查（CRUD） #

# 导入 FastAPI 和 Pydantic
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

# 创建应用
app = FastAPI(title="商品 API")

# 内存中模拟数据库（实际项目会用真实数据库）
items_db = {}
next_id = 1


# 定义商品模型
class Item(BaseModel):
    name: str
    price: float
    description: str = ""


# 获取所有商品
@app.get("/items")
def list_items():
    return {"items": [{"id": k, **v.model_dump()} for k, v in items_db.items()]}


# 根据 id 获取单个商品
@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品不存在")
    return {"id": item_id, **items_db[item_id].model_dump()}


# 创建商品（id 自动递增）
@app.post("/items")
def create_item(item: Item):
    global next_id
    items_db[next_id] = item
    result_id = next_id
    next_id += 1
    return {"message": "创建成功", "id": result_id, "item": item}


# 更新商品
@app.put("/items/{item_id}")
def update_item(item_id: int, item: Item):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品不存在")
    items_db[item_id] = item
    return {"message": "更新成功", "id": item_id, "item": item}


# 删除商品
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    if item_id not in items_db:
        raise HTTPException(status_code=404, detail="商品不存在")
    del items_db[item_id]
    return {"message": "删除成功"}


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

运行方式：保存为 main.py，执行 python main.py，通过 /docs 或 curl 测试各接口。

# 1) GET /items  获取所有商品
curl -i "http://127.0.0.1:8000/items"

# 2) GET /items/{item_id}  获取单个商品（示例 id=1）
curl -i "http://127.0.0.1:8000/items/1"

# 3) POST /items  创建商品
curl -i -X POST "http://127.0.0.1:8000/items" \
  -H "Content-Type: application/json" \
  -d '{"name":"apple","price":12.5,"description":"fresh"}'

# 4) PUT /items/{item_id}  更新商品（示例 id=1）
curl -i -X PUT "http://127.0.0.1:8000/items/1" \
  -H "Content-Type: application/json" \
  -d '{"name":"apple-pro","price":19.9,"description":"updated"}'

# 5) DELETE /items/{item_id}  删除商品（示例 id=1）
curl -i -X DELETE "http://127.0.0.1:8000/items/1"

补充：这组命令里 GET/PUT/DELETE 用到 id=1，所以建议先执行一次 POST 创建数据，再测其余 3 条。

7.1 同步与异步路由 #

路由函数可以是普通的 def（同步），也可以是 async def（异步）。

• 同步 def：在线程池中运行，适合 CPU 密集或大量同步 IO 的库（如部分数据库驱动）。
• 异步 async def：在事件循环中 await，适合 httpx.AsyncClient、aiosqlite 等高并发异步 IO 场景。

不要在 async def 路由里写长时间阻塞的同步代码（会卡住整个事件循环）；必要时仍用 def 交由线程池执行。

# 导入 FastAPI
from fastapi import FastAPI
from pydantic import BaseModel

# 创建应用
app = FastAPI()


class Item(BaseModel):
    name: str


# 同步路由：简单返回即可
@app.get("/sync")
def read_sync():
    return {"mode": "sync"}


# 异步路由：可在此 await 异步客户端、异步数据库等
@app.post("/async-demo", status_code=201)
async def create_async(item: Item):
    # 示例：这里没有 await，仅演示 async 路由与 201 状态码
    return {"created": True, "name": item.name}


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

上例中 `@app.post(..., status_code=201)` 表示成功创建时 HTTP 状态为 201 Created，与 200 OK 在语义上更贴切。

7.2 APIRouter：拆分路由、统一前缀 #

项目变大时，可用 APIRouter 把不同模块的路由分开，再用 include_router 挂到主应用上，并设置 prefix、tags（文档里分组显示）。

# 导入 FastAPI 与 APIRouter
from fastapi import FastAPI, APIRouter

# 主应用
app = FastAPI()

# 用户模块路由：前缀 /users，文档标签「用户」
user_router = APIRouter(prefix="/users", tags=["用户"])


@user_router.get("/")
def list_users():
    return [{"id": 1, "name": "张三"}]


@user_router.get("/{user_id}")
def get_user(user_id: int):
    return {"id": user_id, "name": "示例"}


# 挂载到主应用
app.include_router(user_router)


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

实际项目可建 routers/users.py 只放 user_router，在 main.py 里 include_router，便于维护。

7.3 文件上传与表单 #

浏览器提交「字段 + 附件」时，Content-Type 多为 multipart/form-data（与纯 JSON 的 application/json 不同）。FastAPI 中：

• Form(...)：普通表单字段（字符串、数字等）。
• File(...)：可配合 UploadFile 表示上传文件；也可声明为 bytes（整文件读入内存，只适合小文件）。

UploadFile 基于「临时文件/GC」流式读取，适合较大文件；用 await file.read() 按需读取，或用 file.file 得到类文件对象。

7.3.1 表单字段 + 可选单文件 #

# 导入 FastAPI、File、Form、UploadFile 模块
from fastapi import FastAPI, File, Form, UploadFile

# 创建 FastAPI 应用对象
app = FastAPI()

# 定义 /submit 路由，POST 方法，接收表单字段 name 和可选文件 avatar
@app.post("/submit")
async def submit(
    # 接收表单提交的 name 字段（类型为 str）
    name: str = Form(),
    # 接收上传文件 avatar，可以为 None，类型为 UploadFile
    avatar: UploadFile | None = File(None)
):
    # 如果上传了头像文件
    if avatar:
        # 返回用户名、文件原始名、文件类型
        return {
            "name": name,
            "file": avatar.filename,
            "content_type": avatar.content_type,
        }
    # 如果未上传头像，仅返回用户名和文件为 None
    return {"name": name, "file": None}

# 作为主程序运行时启动 Uvicorn 服务
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

curl 示例（Windows 可改用 Git Bash）：curl -X POST "http://127.0.0.1:8000/submit" -F "name=张三" -F "avatar=@./photo.png"

7.3.2 必填单文件 #

# 导入 FastAPI、File、UploadFile 模块
from fastapi import FastAPI, File, UploadFile

# 创建 FastAPI 应用对象
app = FastAPI()

# 定义 /upload 路由，POST 方法，接收一个必填文件参数 file
@app.post("/upload")
async def upload_one(
    # 声明 file 参数为必填文件（类型为 UploadFile），并加描述
    file: UploadFile = File(..., description="要上传的文件")
):
    # 异步读取上传的文件内容
    data = await file.read()
    # 获取文件字节数
    size = len(data)
    # 返回文件名和文件大小（字节）
    return {"filename": file.filename, "size_bytes": size}

# 判断是否为主程序入口，若是则运行 Uvicorn 启动服务
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

7.3.3 多文件上传 #

使用 list[UploadFile]（或 typing.List[UploadFile]）接收同一字段名下的多个文件，或拆开多个 File() 参数。

# 导入 FastAPI、File、UploadFile 模块
from fastapi import FastAPI, File, UploadFile

# 创建 FastAPI 应用实例
app = FastAPI()

# 定义上传多个文件的路由，POST 方法
@app.post("/upload-many")
async def upload_many(files: list[UploadFile] = File(...)):
    # 用于保存所有文件名
    names = []
    # 用于统计总字节数
    total = 0
    # 遍历每个上传的文件
    for f in files:
        # 异步读取当前文件的内容
        chunk = await f.read()
        # 累加当前文件内容的字节数量
        total += len(chunk)
        # 将当前文件名添加到列表中，没有文件名则为空字符串
        names.append(f.filename or "")
    # 返回上传文件的总数、所有文件名列表及总字节数
    return {"count": len(files), "filenames": names, "total_bytes": total}

# 判断是否作为主程序运行
if __name__ == "__main__":
    # 导入 uvicorn 作为 ASGI 服务器
    import uvicorn
    # 启动 FastAPI 服务，监听 127.0.0.1:8000
    uvicorn.run(app, host="127.0.0.1", port=8000)

curl 多文件：curl -X POST "http://127.0.0.1:8000/upload-many" -F "files=@a.txt" -F "files=@b.txt"

7.3.4 保存到本地目录 #

大文件若整段 read() 仍占内存，生产环境可分批读或使用云存储 SDK。下面演示小文件写入 uploads/ 目录：

# 导入 pathlib 库，用于操作文件和目录路径
from pathlib import Path
# 导入 FastAPI 框架及文件上传相关模块
from fastapi import FastAPI, File, UploadFile

# 定义文件上传保存的目标目录
UPLOAD_DIR = Path("uploads")
# 创建 FastAPI 应用实例
app = FastAPI()

# 定义 POST 路由 /save，用于接收并保存上传的小文件
@app.post("/save")
async def save_file(file: UploadFile = File(...)):
    # 确保上传保存目录存在，不存在则创建
    UPLOAD_DIR.mkdir(exist_ok=True)
    # 对上传文件名进行安全处理，只保留文件名部分，防止路径穿越
    safe_name = Path(file.filename or "unnamed").name
    # 拼接得到最终保存路径
    dest = UPLOAD_DIR / safe_name
    # 异步读取上传文件的全部内容为字节数据
    content = await file.read()
    # 将文件内容直接写入目标路径
    dest.write_bytes(content)
    # 返回保存后的文件路径和文件字节大小
    return {"saved_as": str(dest), "size": len(content)}

# 判断是否为主进程运行，如果是则启动 uvicorn 服务
if __name__ == "__main__":
    # 导入 uvicorn，用于启动 ASGI 服务器
    import uvicorn
    # 启动 FastAPI 服务，监听本地 127.0.0.1:8000
    uvicorn.run(app, host="127.0.0.1", port=8000)

在 /docs 里可对带 File 的接口使用「Try it out」选择本地文件测试。

7.3.5 小结 #

场景	写法
表单字符串	name: str = Form()
可选单文件	`f: UploadFile \	None = File(None)`
必填单文件	f: UploadFile = File(...)
多文件	files: list[UploadFile] = File(...)
读内容	await file.read()
小文件直读内存	data: bytes = File(...)（不推荐大文件）

仅 JSON 请求体时继续用 Pydantic Body 模型，不要用 Form/File。

7.4 Request 与 JSON 响应 #

需要读原始请求体、客户端 IP、请求头等时，可注入 Request（来自 Starlette）。返回非默认 JSON 时可用 JSONResponse 控制状态码与响应头。

# 导入 FastAPI、Request 与 JSONResponse
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

# 创建应用
app = FastAPI()


@app.get("/client-info")
def client_info(request: Request):
    # 客户端主机（代理环境下可能是代理地址）
    host = request.client.host if request.client else ""
    return {"host": host, "path": request.url.path}


@app.get("/custom-json")
def custom_json():
    # 自定义状态码与响应体
    return JSONResponse(status_code=401, content={"error": "未授权"})


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

---

8. 依赖注入（Depends） #

8.1 什么是依赖注入？ #

依赖注入 是把「获取某样东西」的逻辑单独写成函数，由 FastAPI 在每次请求时自动调用并传入路由。这样路由函数只关心业务逻辑，不必重复写「取参数、校验、获取数据库连接」等代码。

通俗理解：就像点外卖——你只说「我要一份盖浇饭」，平台自动帮你联系餐厅、安排骑手。你不需要自己打电话、自己取餐。Depends 就是 FastAPI 里的「自动安排」机制。

8.2 基础用法 #

用 Depends(函数名) 声明依赖，FastAPI 会在处理请求前先调用该函数，把返回值传给路由参数。

# 导入 FastAPI 和 Depends
from fastapi import FastAPI, Depends

# 创建应用
app = FastAPI()


# 定义依赖函数：返回公共的查询参数
def common_params(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}


# 使用 Depends 注入：commons 会自动调用 common_params 并传入结果
@app.get("/items")
def read_items(commons: dict = Depends(common_params)):
    return commons


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

访问 http://127.0.0.1:8000/items?skip=5&limit=20 会得到 {"skip": 5, "limit": 20}。

8.3 依赖链：依赖可以依赖其他依赖 #

依赖函数本身也可以依赖其他依赖，形成依赖链。FastAPI 会按依赖关系自动解析并调用。

# 导入 FastAPI 和 Depends
from fastapi import FastAPI, Depends

# 创建应用
app = FastAPI()


# 第一层依赖：分页参数
def pagination(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}


# 第二层依赖：依赖 pagination，在其基础上增加 keyword
def search_params(pg: dict = Depends(pagination), keyword: str = None):
    return {**pg, "keyword": keyword or ""}


# 路由依赖 search_params（内部会先调用 pagination）
@app.get("/items")
def list_items(params: dict = Depends(search_params)):
    return params


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

访问 http://127.0.0.1:8000/items?skip=2&limit=5&keyword=手机 会得到 {"skip": 2, "limit": 5, "keyword": "手机"}。

8.4 常见场景：模拟「当前用户校验」 #

依赖注入常用于校验请求头、Token 等，通过后再把用户信息传给路由。从请求头读取参数需使用 Header。

# 导入 FastAPI、Depends、Header 和 HTTPException
from fastapi import FastAPI, Depends, Header, HTTPException

# 创建应用
app = FastAPI()


# 从请求头 X-User-Id 读取用户 id，Header(None) 表示可选，缺省为 None
def get_current_user_id(x_user_id: str = Header(None)):
    if not x_user_id:
        raise HTTPException(status_code=401, detail="请提供 X-User-Id 请求头")
    return {"user_id": x_user_id}


# 需要「登录」的接口：依赖 get_current_user_id
@app.get("/me")
def read_me(user: dict = Depends(get_current_user_id)):
    return {"message": f"当前用户: {user['user_id']}"}


# 公开接口：不声明依赖
@app.get("/public")
def read_public():
    return {"message": "公开接口"}


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

测试方式：

# 不带请求头，返回 401
curl -i "http://127.0.0.1:8000/me"

# 带 X-User-Id，返回 200
curl -i -H "X-User-Id: alice" "http://127.0.0.1:8000/me"

8.5 鉴权 #

8.5.1 HTTPException #

HTTPException 用于在路由中主动返回错误响应（如 401 未授权、404 未找到）。抛出后 FastAPI 会将其转为对应的 HTTP 状态码和 JSON 响应。

# 导入 HTTPException
from fastapi import HTTPException

# 抛出 404，detail 会出现在响应体中
raise HTTPException(status_code=404, detail="资源不存在")

# 抛出 401，可自定义 headers
raise HTTPException(status_code=401, detail="未授权", headers={"WWW-Authenticate": "Bearer"})

8.5.2 APIKeyHeader #

APIKeyHeader 是 FastAPI 提供的安全方案之一，用于从请求头读取 API Key。 可把「校验 API Key」抽成依赖，供多个路由复用。

# 导入 FastAPI、Depends、HTTPException、Security
from fastapi import FastAPI, Depends, HTTPException, Security
# 导入 APIKeyHeader
from fastapi.security import APIKeyHeader

# 创建应用
app = FastAPI()

# 定义从请求头 X-API-Key 读取 API Key 的规则
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)


# 依赖函数：校验 API Key，无效则抛出 403
def verify_api_key(api_key: str = Security(api_key_header)):
    if not api_key or api_key != "secret-key-123":
        raise HTTPException(status_code=403, detail="无效的 API Key")
    return api_key


# 需要 API Key 的接口：依赖 verify_api_key
@app.get("/protected")
def read_protected(api_key: str = Depends(verify_api_key)):
    return {"message": "已通过校验", "key_prefix": api_key[:8] + "..."}


# 公开接口：不声明依赖
@app.get("/public")
def read_public():
    return {"message": "公开接口"}


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

测试方式：

# 不带 X-API-Key，返回 403
curl -i "http://127.0.0.1:8000/protected"

# 带错误 Key，返回 403
curl -i -H "X-API-Key: wrong" "http://127.0.0.1:8000/protected"

# 带正确 Key，返回 200
curl -i -H "X-API-Key: secret-key-123" "http://127.0.0.1:8000/protected"

小结：

组件	说明
HTTPException	抛出 HTTP 错误，指定 status_code 和 detail
APIKeyHeader	从指定请求头读取 API Key，如 X-API-Key
Security	声明「安全依赖」，与 Depends 类似，用于认证场景

8.5.3 Security 和 Depends 的区别 #

8.5.3.1. 本质关系 #

Security 和 Depends 在依赖注入上的行为是一样的：都会在请求前执行依赖函数，并把返回值注入到路由参数里。
区别主要在语义和文档生成上。

8.5.3.2. 主要区别 #

维度	Depends	Security
用途	通用依赖（分页、数据库、业务逻辑等）	安全相关依赖（认证、鉴权）
OpenAPI 文档	作为普通参数展示	作为安全要求展示，出现在 security 中
Swagger UI	在参数列表里显示	显示锁图标，可配置 API Key / OAuth 等
语义	普通依赖	表示“需要认证/鉴权”

8.5.3.3. 文档表现差异 #

使用 Depends：

def verify_key(api_key: str = Depends(api_key_header)):
    ...

在 /docs 里，api_key 会作为普通参数出现在接口参数列表中。

使用 Security：

def verify_key(api_key: str = Security(api_key_header)):
    ...

在 /docs 里会：

• 在接口旁显示锁图标
• 在页面顶部提供“Authorize”按钮
• 在 OpenAPI 的 security 字段中声明该接口需要认证

8.5.3.4. 底层实现 #

Security 可以理解为专门用于安全场景的 Depends，内部会设置 use_cache=False 等参数，并在生成 OpenAPI 时做特殊处理，把依赖标记为安全要求。

8.5.3.5. 使用建议 #

• 认证、鉴权（API Key、OAuth、JWT 等）→ 用 Security
• 其他依赖（数据库、分页、公共参数等）→ 用 Depends

8.6 常见场景：数据库连接 #

通过依赖注入管理数据库连接，每次请求获取连接、用完后自动关闭，避免连接泄漏。使用 yield 时，yield 之前的代码在请求开始时执行，之后的代码在请求结束后执行（类似 finally）。

# 导入 asynccontextmanager 上下文管理器用于 lifespan 生命周期管理
from contextlib import asynccontextmanager
# 从 FastAPI 导入 FastAPI 应用类和 Depends 依赖注入工具
from fastapi import FastAPI, Depends
# 导入 sqlite3 模块，用于操作 SQLite 数据库
import sqlite3

# 定义数据库文件路径
DB_PATH = "demo.db"


# 依赖函数：获取数据库连接，使用 yield 实现请求结束后自动关闭连接
def get_db():
    # 连接到 SQLite 数据库
    conn = sqlite3.connect(DB_PATH)
    # 设置查询返回的每一行以 dict 形式表现，支持通过列名获取数据
    conn.row_factory = sqlite3.Row
    try:
        # 通过 yield 暴露连接对象
        yield conn
    finally:
        # 请求结束后关闭连接，防止资源泄漏
        conn.close()


# 初始化数据库，创建 users 表并插入初始用户数据
def init_db():
    # 建立数据库连接
    conn = sqlite3.connect(DB_PATH)
    # 创建表（如果不存在），包含 id 和 name 字段
    conn.execute("CREATE TABLE IF NOT EXISTS users (id INT, name TEXT)")
    # 插入或更新样例数据：id=1 的张三、id=2 的李四
    conn.execute("INSERT OR REPLACE INTO users VALUES (1, '张三'), (2, '李四')")
    # 提交事务
    conn.commit()
    # 关闭连接
    conn.close()


# 声明异步 lifespan 生命周期上下文，在应用启动时初始化数据库
@asynccontextmanager
async def lifespan(app: FastAPI):
    # 在应用启动阶段调用数据库初始化函数
    init_db()
    # 进入主生命周期，后续事件继续执行
    yield


# 创建 FastAPI 应用实例，并指定 lifespan 生命周期管理
app = FastAPI(lifespan=lifespan)


# 定义获取单个用户信息的 GET 路由接口，依赖数据库连接
@app.get("/users/{user_id}")
def get_user(user_id: int, conn=Depends(get_db)):
    # 执行 SQL 查询，根据用户 id 查找用户
    cur = conn.execute("SELECT id, name FROM users WHERE id = ?", (user_id,))
    # 获取一条查询结果
    row = cur.fetchone()
    # 如果没有找到对应用户，返回错误信息
    if not row:
        return {"error": "用户不存在"}
    # 找到对应用户，返回 id 和 name 信息
    return {"id": row["id"], "name": row["name"]}


# 判断当前脚本是否为主程序入口
if __name__ == "__main__":
    # 导入 uvicorn 用于启动 FastAPI 应用
    import uvicorn
    # 启动 uvicorn 服务，监听本地 127.0.0.1:8000 端口
    uvicorn.run(app, host="127.0.0.1", port=8000)

运行方式：保存为 main.py，执行 python main.py，访问 http://127.0.0.1:8000/users/1 可得到 {"id": 1, "name": "张三"}。

8.7 小结 #

要点	说明
声明方式	参数 = Depends(依赖函数)
执行时机	每次请求前自动调用依赖函数
依赖链	依赖函数可以再依赖其他依赖
常见用途	分页参数、用户校验、数据库连接等

9. 后台任务（BackgroundTasks） #

9.1 什么是 BackgroundTasks？ #

BackgroundTasks 用于在响应返回之后执行一些耗时或非关键操作，如发邮件、写日志、清理缓存。客户端不用等这些任务完成，就能先收到响应。

通俗理解：就像餐厅——顾客点完餐，服务员先回复「好的，马上做」，然后去后厨下单。顾客不用站在柜台等菜做好。BackgroundTasks 就是「先回复，再在后台慢慢做」的机制。

9.2 基本用法 #

在路由中声明 BackgroundTasks 类型的参数，通过 add_task 添加要在响应返回后执行的任务。

# 导入 FastAPI Web 框架和后台任务模块
from fastapi import FastAPI, BackgroundTasks

# 创建一个 FastAPI 应用实例
app = FastAPI()

# 定义一个模拟发送邮件的后台任务函数
def send_email(email: str, content: str):
    # 打印后台邮件发送信息
    print(f"[后台] 发送邮件给 {email}: {content}")

# 定义 /notify 路由，POST 请求
@app.post("/notify")
# 路由处理函数，接收 email 参数和后台任务对象
def notify_user(email: str, background_tasks: BackgroundTasks):
    # 添加 send_email 任务到后台，响应先返回后再执行任务
    background_tasks.add_task(send_email, email, "您有一条新通知")
    # 立即返回响应内容（不等待邮件发送完成）
    return {"message": "通知已提交"}

# 仅当本文件作为主程序运行时才执行以下代码
if __name__ == "__main__":
    # 导入 uvicorn 服务器
    import uvicorn
    # 启动 FastAPI 应用，监听127.0.0.1:8000
    uvicorn.run(app, host="127.0.0.1", port=8000)

测试方式：curl -X POST "http://127.0.0.1:8000/notify?email=user@example.com"，会先收到 {"message": "通知已提交"}，终端随后打印 [后台] 发送邮件给 user@example.com: ...。

9.3 注册后发欢迎邮件 #

# 导入 FastAPI 应用类和后台任务工具
from fastapi import FastAPI, BackgroundTasks
# 导入 Pydantic 的 BaseModel，用于数据校验
from pydantic import BaseModel

# 创建 FastAPI 应用实例
app = FastAPI()

# 定义请求体的数据模型，包含 email 和 name 字段
class UserCreate(BaseModel):
    # 用户邮箱
    email: str
    # 用户名
    name: str

# 模拟：保存用户信息到数据库（这里只是打印，实际项目需要用数据库操作）
def save_user(email: str, name: str):
    # 输出正在保存的用户信息
    print(f"[后台] 保存用户: {email}, {name}")

# 模拟：发送欢迎邮件（这里只是打印，实际项目需要集成邮件服务）
def send_welcome_email(email: str, name: str):
    # 输出正在发送的欢迎邮件信息
    print(f"[后台] 发送欢迎邮件给 {name} <{email}>")

# 创建注册接口，POST 请求到 "/register"
@app.post("/register")
def register(user: UserCreate, background_tasks: BackgroundTasks):
    # 添加保存用户的后台任务
    background_tasks.add_task(save_user, user.email, user.name)
    # 添加发送欢迎邮件的后台任务
    background_tasks.add_task(send_welcome_email, user.email, user.name)
    # 立即返回注册成功信息
    return {"message": "注册成功", "email": user.email}

# 当该文件作为主程序运行时，启动 uvicorn 服务
if __name__ == "__main__":
    # 导入 uvicorn 启动 ASGI 服务器
    import uvicorn
    # 以 127.0.0.1:9000 运行 FastAPI 应用
    uvicorn.run(app, host="127.0.0.1", port=9000)

curl -X POST "http://127.0.0.1:9000/register" -H "Content-Type: application/json" -d '{"email":"zhangsan@qq.com","name":"zhangsan"}'

9.4 小结 #

要点	说明
用途	响应返回后执行耗时或非关键操作
声明	background_tasks: BackgroundTasks
添加任务	background_tasks.add_task(函数, 参数1, 参数2, ...)
执行顺序	按 add_task 调用顺序依次执行

10. CORS 跨域配置 #

当前端（如网页）从不同域名访问你的 API 时，浏览器会做跨域检查。若未配置 CORS，请求可能被拦截。

# 导入 FastAPI 类
from fastapi import FastAPI
# 导入用于支持跨域请求的 CORS 中间件
from fastapi.middleware.cors import CORSMiddleware

# 创建 FastAPI 应用实例
app = FastAPI()

# 配置并添加 CORS 中间件，允许所有来源的跨域请求
app.add_middleware(
    CORSMiddleware,
    # 允许所有来源访问，生产环境推荐填写具体域名
    allow_origins=["*"],
    # 允许客户端发送 cookies 等凭证信息
    allow_credentials=True,
    # 允许所有 HTTP 方法
    allow_methods=["*"],
    # 允许所有请求头
    allow_headers=["*"],
)

# 定义根路径的 GET 请求接口
@app.get("/")
def read_root():
    # 返回一条消息
    return {"message": "Hello"}

# 判断是否以脚本方式运行
if __name__ == "__main__":
    # 导入 uvicorn，用于运行 ASGI 服务器
    import uvicorn
    # 启动 FastAPI 应用，监听本地 127.0.0.1:8000
    uvicorn.run(app, host="127.0.0.1", port=8000)

11. 自动文档与 OpenAPI #

FastAPI 会根据路由、类型注解和 Pydantic 模型生成 OpenAPI（原 Swagger）规范，无需手写。

地址	说明
http://127.0.0.1:8000/docs	Swagger UI，可在线调试
http://127.0.0.1:8000/redoc	ReDoc，阅读式文档
http://127.0.0.1:8000/openapi.json	原始 OpenAPI JSON，可导入 Postman 等工具

11.1 增强文档：tags、summary、description #

在装饰器或 APIRouter 上使用 tags 可在 /docs 里分组；用 summary / description 补充接口说明。

# 导入 FastAPI
from fastapi import FastAPI

# 创建应用，可配置全局标题与版本
app = FastAPI(
    title="示例 API",
    version="0.1.0",
    description="教学用示例，实际项目可写更完整的说明。",
)


@app.get(
    "/health",
    tags=["系统"],
    summary="健康检查",
    description="用于负载均衡或监控探活，返回服务是否可用。",
)
def health():
    return {"status": "ok"}


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

生产环境若需关闭交互文档，可在创建应用时设置 docs_url=None, redoc_url=None（按需使用）。

12. 运行方式 #

方式一：在代码中调用 uvicorn（如前文示例）

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

方式二：命令行启动（推荐开发时使用 --reload 热重载）

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

其中 main 是文件名，app 是 FastAPI 实例的变量名。

13. 总结 #

• FastAPI 基于 Starlette（ASGI），配合 Pydantic 做校验与文档。
• 路径参数 写在 {param} 中；查询参数 可用 Query/Optional 做校验。
• 请求体 用 Pydantic 模型；Field / response_model 做强校验与响应裁剪。
• 同步 def 与异步 async def 按场景选用；APIRouter 便于拆分模块。
• Form / UploadFile 处理表单与上传；Request、JSONResponse 处理更底层需求。
• Depends 做依赖注入；HTTPException，Security + APIKeyHeader 做鉴权。
• BackgroundTasks 在响应返回后执行耗时任务。
• /docs、/redoc、/openapi.json 提供自动文档；tags/summary 可增强可读性。