知识小屋

    FastAPI 入门教程

    FastAPI 是一个现代、高性能的 Python Web 框架，基于标准 Python 类型提示构建，自带交互式文档，性能接近 Node.js 和 Go。

    官网：https://fastapi.tiangolo.com

    1、安装与启动

    pip install "fastapi[standard]"

    fastapi[standard] 会自动安装 Uvicorn 等常用依赖，建议用引号包裹确保终端兼容。

    最小示例：

    # main.py
from fastapi import FastAPI

app = FastAPI()

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

    启动开发服务器：

    fastapi dev

    启动后访问：

    fastapi dev 默认开启热重载。生产环境使用 fastapi run。

    2、路径参数（Path Parameters）

    路径参数使用花括号 {} 声明，配合类型注解实现自动校验和类型转换：

    @app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}
    • 访问 /items/3 → item_id 为 int 类型的 3
    • 访问 /items/foo → 自动返回 422 错误，提示不是合法整数

    枚举限定值

    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):
    return {"model_name": model_name}

    Swagger 文档会自动展示可选值下拉。

    路由顺序

    固定路径要写在路径参数前面，否则会被通配匹配：

    @app.get("/users/me")
async def read_user_me():
    return {"user_id": "current user"}

@app.get("/users/{user_id}")
async def read_user(user_id: str):
    return {"user_id": user_id}

    3、查询参数（Query Parameters）

    函数参数中不在路径里的声明，自动成为查询参数：

    @app.get("/items/{item_id}")
async def read_item(item_id: int, q: str | None = None, short: bool = False):
    result = {"item_id": item_id}
    if q:
        result["q"] = q
    if not short:
        result["description"] = "这是一个很长的描述"
    return result
    • /items/5?q=fastapi → 返回带 q 的结果
    • /items/5 → q 为 None，可选参数
    • /items/5?short=true → 布尔查询参数

    布尔参数：值为 true、1、True 等均为 True，其他为 False。

    4、请求体（Request Body）

    用 Pydantic 模型声明请求体，自动完成 JSON 解析、类型转换和校验：

    from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float
    tax: float | None = None

@app.post("/items/")
async def create_item(item: Item):
    return item

    发送请求体：

    {
    "name": "Foo",
    "price": 45.2,
    "description": "可选描述"
}
    • name 和 price 必传
    • description 和 tax 可选（有默认值 None）

    同时使用路径参数、查询参数和请求体

    FastAPI 会自动识别参数来源：

    @app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item, q: str | None = None):
    result = {"item_id": item_id, **item.model_dump()}
    if q:
        result["q"] = q
    return result
    • item_id → 路径参数
    • q → 查询参数（str 单值类型）
    • item → 请求体（Pydantic 模型）

    5、响应模型（Response Model）

    用 response_model 控制返回字段，自动过滤多余属性：

    class UserOut(BaseModel):
    id: int
    name: str

@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):
    return {"id": user_id, "name": "Alice", "password": "secret"}

    返回结果中不会包含 password 字段。

    6、表单与文件上传

    表单数据

    from fastapi import Form

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

    文件上传

    from fastapi import UploadFile, File

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    content = await file.read()
    return {"filename": file.filename, "size": len(content)}

    7、依赖注入（Dependency Injection）

    依赖注入是 FastAPI 的核心特性，用于复用公共逻辑：

    from typing import Annotated
from fastapi import Depends, FastAPI

app = FastAPI()

async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

@app.get("/users/")
async def read_users(commons: Annotated[dict, Depends(common_parameters)]):
    return commons

    数据库连接依赖

    from collections.abc import AsyncGenerator
from fastapi import Depends

async def get_db() -> AsyncGenerator:
    db = connect_db()
    try:
        yield db
    finally:
        await db.close()

@app.get("/users/")
async def list_users(db: AsyncGenerator = Depends(get_db)):
    return db.query_all("users")

    认证依赖

    from fastapi import HTTPException, Header

async def verify_token(token: str = Header(...)):
    if token != "secret":
        raise HTTPException(status_code=401, detail="Invalid token")
    return token

@app.get("/protected")
async def protected(token: str = Depends(verify_token)):
    return {"message": "ok"}

    8、异常处理

    HTTPException

    from fastapi import HTTPException

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

    自定义异常处理器

    from fastapi.responses import JSONResponse
from fastapi import Request

class BizError(Exception):
    def __init__(self, message: str):
        self.message = message

@app.exception_handler(BizError)
async def biz_error_handler(request: Request, exc: BizError):
    return JSONResponse(status_code=400, content={"error": exc.message})

    9、中间件

    import time
from fastapi import Request

@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

    10、跨域（CORS）

    from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],  # 生产环境指定域名
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

    11、路由分组（APIRouter）

    项目变大后，用 APIRouter 拆分路由：

    # routers/users.py
from fastapi import APIRouter

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

@router.get("/")
async def list_users():
    return []

@router.get("/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}
    # main.py
from fastapi import FastAPI
from routers.users import router as users_router

app = FastAPI()
app.include_router(users_router)

    12、静态文件挂载

    from fastapi.staticfiles import StaticFiles

app.mount("/static", StaticFiles(directory="static"), name="static")

    13、异步支持

    FastAPI 同时支持同步 def 和异步 async def：

    import httpx

@app.get("/proxy")
async def proxy():
    async with httpx.AsyncClient() as client:
        resp = await client.get("https://httpbin.org/get")
        return resp.json()
    • async def → 在事件循环中直接执行
    • def → 自动在线程池中运行，不会阻塞事件循环

    14、生命周期事件（Lifespan）

    推荐使用 lifespan 替代已废弃的 @app.on_event：

    from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时执行
    print("Starting up...")
    yield
    # 关闭时执行
    print("Shutting down...")

app = FastAPI(lifespan=lifespan)

    15、推荐项目结构

    以下结构基于 FastAPI 官方全栈模板 和企业级实践整理，按职责清晰分层：

    myproject/
├── backend/
│   ├── app/
│   │   ├── __init__.py
│   │   ├── main.py                 # 应用入口，注册路由、中间件、生命周期
│   │   │
│   │   ├── core/                   # 核心配置
│   │   │   ├── __init__.py
│   │   │   ├── config.py           # 环境变量、项目配置（Pydantic Settings）
│   │   │   ├── security.py         # JWT、密码哈希、认证逻辑
│   │   │   └── database.py         # 数据库引擎、Session 管理
│   │   │
│   │   ├── models/                 # 数据库模型（ORM/SQLModel）
│   │   │   ├── __init__.py
│   │   │   └── user.py             # 按业务实体拆分
│   │   │
│   │   ├── schemas/                # Pydantic 请求/响应模型
│   │   │   ├── __init__.py
│   │   │   └── user.py             # 与 models 一一对应
│   │   │
│   │   ├── crud/                   # 数据库操作（增删改查）
│   │   │   ├── __init__.py
│   │   │   └── crud_user.py        # 按实体拆分，封装所有 DB 操作
│   │   │
│   │   ├── api/                    # 路由层
│   │   │   ├── __init__.py
│   │   │   ├── deps.py             # 公共依赖（当前用户、分页等）
│   │   │   └── v1/                 # API 版本管理
│   │   │       ├── __init__.py
│   │   │       ├── router.py       # 聚合注册所有 v1 路由
│   │   │       ├── users.py        # 用户相关接口
│   │   │       └── items.py        # 商品相关接口
│   │   │
│   │   ├── services/               # 业务逻辑层（复杂业务放这里）
│   │   │   ├── __init__.py
│   │   │   └── user_service.py     # 纯业务逻辑，不含路由和 DB 直接操作
│   │   │
│   │   ├── middleware/             # 中间件
│   │   │   ├── __init__.py
│   │   │   └── logging.py          # 请求日志、耗时统计等
│   │   │
│   │   └── utils/                  # 工具函数
│   │       ├── __init__.py
│   │       └── email.py            # 邮件发送、文件处理等通用工具
│   │
│   ├── tests/                      # 测试
│   │   ├── __init__.py
│   │   ├── conftest.py             # pytest fixtures（测试数据库、客户端等）
│   │   ├── test_api/               # 按层测试
│   │   │   └── test_users.py
│   │   └── test_services/
│   │       └── test_user_service.py
│   │
│   ├── alembic/                    # 数据库迁移
│   │   ├── versions/
│   │   └── env.py
│   │
│   ├── alembic.ini                 # Alembic 配置
│   ├── pyproject.toml              # 项目依赖与配置
│   └── Dockerfile                  # 容器化部署
│
├── docker-compose.yml              # 本地开发环境编排
├── .env.example                    # 环境变量模板
└── README.md

    分层职责说明

    层级目录职责规则
    入口main.py创建 app、注册路由和中间件不写业务逻辑，只做组装
    配置core/config.py环境变量、数据库 URL、密钥等用 pydantic-settings 管理，按环境区分
    模型models/数据库表结构只定义字段和表关系，不放业务逻辑
    Schemaschemas/API 的请求/响应结构独立于 DB 模型，控制接口暴露的字段
    CRUDcrud/数据库增删改查只做数据操作，不包含业务判断
    路由api/接口定义、参数校验、调用 service不直接写 SQL，通过 crud 或 service 操作数据
    业务services/复杂业务逻辑、事务编排路由和 crud 之间的中间层，简单项目可省略
    测试tests/单元测试、接口测试用 pytest，按被测层分类

    关键设计原则

    1. 路由不碰数据库

    # ❌ 错误：路由直接操作 DB
@router.get("/users/")
async def list_users(db: Session = Depends(get_db)):
    return db.query(User).all()

# ✅ 正确：路由 → service/crud → DB
@router.get("/users/")
async def list_users(user_service: UserService = Depends())
    return user_service.get_all()

    2. models 与 schemas 分离

    # models/user.py — 数据库模型
from sqlmodel import SQLModel, Field

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    email: str
    hashed_password: str  # 数据库存哈希密码

# schemas/user.py — API 响应模型
from pydantic import BaseModel

class UserRead(BaseModel):
    id: int
    email: str
    # 不暴露 hashed_password

    3. API 版本化

    # api/v1/router.py
from fastapi import APIRouter
from .users import router as users_router

router = APIRouter(prefix="/api/v1")
router.include_router(users_router)

# main.py
from api.v1.router import router as v1_router
app.include_router(v1_router)

    4. 配置用环境变量，不要硬编码

    # core/config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: str
    SECRET_KEY: str
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 30
    ENVIRONMENT: str = "development"

    model_config = {"env_file": ".env"}

settings = Settings()

    pyproject.toml 配置入口

    [tool.fastapi]
entrypoint = "app.main:app"

    这样 fastapi dev 会自动找到 app 对象，无需每次指定文件路径。

    参考