FastApi框架源码赏析-核心应用模块

 
    def __init__(
        self: AppType,
        debug: bool = False,
        routes: Sequence[BaseRoute] | None = None,
        middleware: Sequence[Middleware] | None = None,
        exception_handlers: Mapping[Any, ExceptionHandler] | None = None,
        on_startup: Sequence[Callable[[], Any]] | None = None,
        on_shutdown: Sequence[Callable[[], Any]] | None = None,
        lifespan: Lifespan[AppType] | None = None,
    ) -> None:
        """初始化应用.
 
        debug：出错时是否返回调试堆栈。
        routes：HTTP/WebSocket 路由列表。
        middleware：全局中间件列表；框架会自动在首尾加上 ServerErrorMiddleware 和 ExceptionMiddleware。
        exception_handlers：状态码或异常类 → 处理函数映射，函数签名为 handler(request, exc) -> response。
        on_startup：启动时执行的函数列表（可同步/异步）。
        on_shutdown：关闭前执行的函数列表（可同步/异步）。
        lifespan：一次性上下文函数，替代 on_startup/on_shutdown，两者只能选其一。
        """
        # 断言：lifespan / on_startup + on_shutdown 二选一
        assert lifespan is None or (on_startup is None and on_shutdown is None), (
            "Use either 'lifespan' or 'on_startup'/'on_shutdown', not both."
        )
 
        self.debug = debug
        self.state = State()
        self.router = Router(routes, on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan)
        self.exception_handlers = {} if exception_handlers is None else dict(exception_handlers)
        self.user_middleware = [] if middleware is None else list(middleware)
        self.middleware_stack: ASGIApp | None = None

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        scope["app"] = self
        # 没有则创建中间件栈
        if self.middleware_stack is None:
            self.middleware_stack = self.build_middleware_stack()
        await self.middleware_stack(scope, receive, send)

def build_middleware_stack(self) -> ASGIApp:
    """
    把用户配置转换成一条 ASGI 洋葱式中间件链，顺序（由外到内）：
    1. ServerErrorMiddleware     —— 捕获未处理异常，返回 500 或调试页面
    2. 用户中间件列表            —— 按声明顺序依次包裹
    3. ExceptionMiddleware       —— 处理已声明的异常/状态码
    4. Router                    —— 真正的路由分发
    """
    debug = self.debug
    error_handler = None                     # 专门给 ServerErrorMiddleware 用
    exception_handlers: dict[Any, ExceptionHandler] = {}
 
    # 把 500 或 Exception 的处理器挑出来给最外层，其余留给 ExceptionMiddleware
    for key, value in self.exception_handlers.items():
        if key in (500, Exception):
            error_handler = value
        else:
            exception_handlers[key] = value
 
    # 组装最终中间件列表
    middleware = (
        [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
        + self.user_middleware
        + [Middleware(ExceptionMiddleware, handlers=exception_handlers, debug=debug)]
    )
 
    # 反向包裹：从 Router 开始，把中间件一层层包在外面
    app = self.router
    for cls, args, kwargs in reversed(middleware):
        app = cls(app, *args, **kwargs)
    return app

Starlette
├─ 字段
│  ├─ debug
│  ├─ state
│  ├─ router → .routes
│  ├─ exception_handlers
│  └─ middleware_stack（惰性构建）
├─ 初始化
│  └─ __init__ / __call__
├─ 构建
│  └─ build_middleware_stack
├─ 动态注册
│  ├─ add_middleware
│  ├─ add_exception_handler
│  ├─ add_route / add_websocket_route
│  ├─ mount / host
│  └─ add_event_handler
├─ URL 反向
│  └─ url_path_for
└─ 弃用装饰器
   ├─ exception_handler
   ├─ route
   ├─ websocket_route
   └─ middleware
 

def __init__(
    self: AppType,
    *,
    debug: bool = False,  # 是否返回调试堆栈
    routes: Optional[List[BaseRoute]] = None,  # 路由列表
    title: str = "FastAPI",  # API 的标题
    summary: Optional[str] = None,  # API 的简短摘要
    description: str = "",  # API 的描述，支持 Markdown
    version: str = "0.1.0",  # API 的版本
    openapi_url: Optional[str] = "/openapi.json",  # OpenAPI schema 的 URL
    openapi_tags: Optional[List[Dict[str, Any]]] = None,  # OpenAPI 标签元数据
    servers: Optional[List[Dict[str, Union[str, Any]]]] = None,  # 服务器连接信息
    dependencies: Optional[Sequence[Depends]] = None,  # 全局依赖项
    default_response_class: Type[Response] = JSONResponse,  # 默认响应类
    redirect_slashes: bool = True,  # 是否重定向斜杠
    docs_url: Optional[str] = "/docs",  # Swagger UI 的路径
    redoc_url: Optional[str] = "/redoc",  # ReDoc 的路径
    swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect",  # OAuth2 重定向 URL
    swagger_ui_init_oauth: Optional[Dict[str, Any]] = None,  # Swagger UI 的 OAuth2 配置
    middleware: Optional[Sequence[Middleware]] = None,  # 中间件列表
    exception_handlers: Optional[Dict[Union[int, Type[Exception]], Callable[[Request, Any], Coroutine[Any, Any, Response]]]] = None,  # 异常处理器
    on_startup: Optional[Sequence[Callable[[], Any]]] = None,  # 启动时执行的函数
    on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,  # 关闭时执行的函数
    lifespan: Optional[Lifespan[AppType]] = None,  # 生命周期上下文管理器
    terms_of_service: Optional[str] = None,  # 服务条款 URL
    contact: Optional[Dict[str, Union[str, Any]]] = None,  # 联系信息
    license_info: Optional[Dict[str, Union[str, Any]]] = None,  # 许可信息
    openapi_prefix: str = "",  # OpenAPI URL 前缀（已弃用）
    root_path: str = "",  # 代理处理的路径前缀
    root_path_in_servers: bool = True,  # 是否在 OpenAPI 中自动生成 `servers` 字段
    responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,  # OpenAPI 中额外的响应
    callbacks: Optional[List[BaseRoute]] = None,  # OpenAPI 回调
    webhooks: Optional[routing.APIRouter] = None,  # OpenAPI Webhooks
    deprecated: Optional[bool] = None,  # 是否标记所有路径操作为已弃用
    include_in_schema: bool = True,  # 是否在 OpenAPI 中包含所有路径操作
    swagger_ui_parameters: Optional[Dict[str, Any]] = None,  # Swagger UI 配置参数
    generate_unique_id_function: Callable[[routing.APIRoute], str] = generate_unique_id,  # 生成唯一 ID 的函数
    separate_input_output_schemas: bool = True,  # 是否为请求体和响应体生成独立的 OpenAPI 模式
    **extra: Any,  # 额外的关键字参数
) -> None:
    # 初始化应用状态
    self.debug = debug
    self.title = title
    self.summary = summary
    self.description = description
    self.version = version
    self.terms_of_service = terms_of_service
    self.contact = contact
    self.license_info = license_info
    self.openapi_url = openapi_url
    self.openapi_tags = openapi_tags
    self.root_path_in_servers = root_path_in_servers
    self.docs_url = docs_url
    self.redoc_url = redoc_url
    self.swagger_ui_oauth2_redirect_url = swagger_ui_oauth2_redirect_url
    self.swagger_ui_init_oauth = swagger_ui_init_oauth
    self.swagger_ui_parameters = swagger_ui_parameters
    self.servers = servers or []
    self.separate_input_output_schemas = separate_input_output_schemas
    self.extra = extra
    self.openapi_version = "3.1.0"  # OpenAPI 版本
    self.openapi_schema = None  # 缓存的 OpenAPI schema
    self.webhooks = webhooks or routing.APIRouter()  # Webhooks 路由
    self.root_path = root_path or openapi_prefix  # 根路径
    self.state = State()  # 应用状态
    self.dependency_overrides = {}  # 依赖覆盖
    self.router = routing.APIRouter(  # 路由
        routes=routes,
        redirect_slashes=redirect_slashes,
        dependency_overrides_provider=self,
        on_startup=on_startup,
        on_shutdown=on_shutdown,
        lifespan=lifespan,
        default_response_class=default_response_class,
        dependencies=dependencies,
        callbacks=callbacks,
        deprecated=deprecated,
        include_in_schema=include_in_schema,
        responses=responses,
        generate_unique_id_function=generate_unique_id_function,
    )
    self.exception_handlers = exception_handlers or {}  # 异常处理器
    self.user_middleware = middleware or []  # 用户中间件
    self.middleware_stack = None  # 中间件堆栈
    self.setup()  # 初始化设置

app.add_api_route("/items", read_item, methods=["GET"])
app.add_api_websocket_route("/ws", ws_handler)

FastAPI (Starlette)
├─ Router (APIRouter)
│  ├─ HTTP Routes
│  ├─ WebSocket Routes
│  ├─ Sub-Routers
│  └─ Middleware
├─ Exception Handlers
├─ Global Dependencies
├─ Lifespan Context
└─ OpenAPI Generator