下载工作台
FastAPI 开发

路由、依赖注入与 lifespan

试读上半部分 · 解锁后可读全文

第 2 章 · 路由、依赖注入与 lifespan

本章目标:掌握 FastAPI 路由装饰器APIRouter 模块化拆分;理解 Depends 依赖注入机制,实现可复用的数据库会话、配置、鉴权占位;使用 lifespan 上下文管理器处理应用启动与关闭(连接池初始化、资源释放);将 svc-demo 从单文件 main.py 重构为可维护的多模块结构;能对照 Django URLConf 与 Flask Blueprint 说明差异。

学时建议:4~5 小时(含 1.5 小时跟练)

前置:完成 fastapi-web ch01;学过 flask-web ch02(蓝图)django-web ch01(urls.py) 更佳。


2.1 场景说明:svc-demo 接口越来越多

ch01 的 main.py 只有 3 个路由。随着 svc-demo 接入商品、订单、用户等域,需要:

需求ch01 现状本章目标
路由组织全写在 main.pyAPIRouter 按业务分包
公共逻辑Depends 注入配置、日志、DB 占位
启动初始化lifespan 打开连接池、预热缓存
关闭清理直接杀进程lifespan 优雅关闭连接
URL 前缀手写完整路径prefix="/api/v1/products"

对照其他框架:

FastAPIDjango(shop-demo)Flask(api-demo)
APIRouterinclude(urls)Blueprint
Depends(get_db)中间件 / 装饰器g / 装饰器
lifespanAppConfig.ready()@app.before_first_request(已废弃)

2.2 推荐目录结构

svc-demo/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 工厂 + lifespan
│   ├── config.py            # 配置(占位)
│   ├── dependencies.py      # 公共 Depends
│   └── routers/
│       ├── __init__.py
│       ├── health.py
│       └── products.py      # 商品路由(内存假数据)
├── requirements.txt
└── README.md

启动方式改为:

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

2.3 APIRouter 基础

2.3.1 健康检查路由 app/routers/health.py

from fastapi import APIRouter

router = APIRouter(tags=["健康检查"])


@router.get("/health")
async def health():
    return {"status": "healthy"}


@router.get("/ready")
async def ready():
    """就绪探针:可扩展为检查数据库连通性(ch05)"""
    return {"ready": True}

2.3.2 商品路由 app/routers/products.py

from fastapi import APIRouter, Depends, HTTPException

from app.dependencies import get_pagination, Pagination

router = APIRouter(prefix="/api/v1/products", tags=["商品"])

# 教学用内存数据,ch05 换为数据库
_FAKE_PRODUCTS = [
    {"id": 1, "sku": "SKU-001", "name": "无线鼠标", "price": 89.0},
    {"id": 2, "sku": "SKU-002", "name": "机械键盘", "price": 399.0},
    {"id": 3, "sku": "SKU-003", "name": "USB-C 集线器", "price": 129.0},
]


@router.get("/")
async def list_products(pagination: Pagination = Depends(get_pagination)):
    start = pagination.offset
    end = start + pagination.limit
    items = _FAKE_PRODUCTS[start:end]
    return {"total": len(_FAKE_PRODUCTS), "items": items}


@router.get("/{product_id}")
async def get_product(product_id: int):
    for p in _FAKE_PRODUCTS:
        if p["id"] == product_id:
            return p
    raise HTTPException(status_code=404, detail="商品不存在")

2.3.3 注册路由 app/main.py

from contextlib import asynccontextmanager

from fastapi import FastAPI

from app.routers import health, products


@asynccontextmanager
async def lifespan(app: FastAPI):
    # ── 启动 ──
    print("[lifespan] svc-demo 启动:初始化资源…")
    app.state.startup_message = "svc-demo is running"
    yield
    # ── 关闭 ──
    print("[lifespan] svc-demo 关闭:释放资源…")


def create_app() -> FastAPI:
    app = FastAPI(
        title="svc-demo API",
        version="0.2.0",
        lifespan=lifespan,
    )
    app.include_router(health.router)
    app.include_router(products.router)
    return app


app = create_app()
参数作用示例
prefix路由组统一前缀/api/v1/products
tagsSwagger 文档分组["商品"]
include_router挂载子路由类似 Flask register_blueprint

2.4 Depends 依赖注入

依赖注入(DI)让你把「获取数据库会话」「解析分页参数」「校验 API Key」等逻辑写成可复用函数,由框架在每次请求前自动调用。

2.4.1 分页依赖 app/dependencies.py

from dataclasses import dataclass

from fastapi import Query


@dataclass
class Pagination:
    offset: int
    limit: int


def get_pagination(
    offset: int = Query(0, ge=0, description="跳过条数"),
    limit: int = Query(20, ge=1, le=100, description="每页条数"),
) -> Pagination:
    return Pagination(offset=offset, limit=limit)

以下内容需解锁后阅读

试读已结束。解锁本章 ¥5.00,或开通年度会员畅读全部教程。
年度会员 ¥199.00/年; 小紫 AI 工作台有效会员 ¥99.00/年

正文仅在服务端鉴权后下发,未付费无法获取下半部分内容。