1 前言

前面我们已经大致了解清楚了我们的starlette内部执行机制，接下来我们开始分析一下FastAPI，既然FastAPI是基于starlette，所以它肯定是在它之上扩展了更多的功能点，比如：

• openapi文档整合
• 依赖注入的实现
• pydantic参数校验整合

所以接下来几篇也会围绕着几点进行分析FastAPI框架一些设计理念和原理。

重点-解析一个请求流程中依赖注入和路由请求流程

2  创建app = FastAPI()都做了哪些事情

我们知道要启动一个app服务，需要一个app对象，而这个app对象管理着一堆的其他实现asgi协议标准的其他asgiapp的对象，那么当我们在初始化一个FastAPI类型的app的时候都做了些什么事情呢？

源码的版本为0.78.0版本！

APP初始化的时候源码分析:

class FastAPI(Starlette):
    def __init__(
        self,
        *, # 表示当前参数按位置
        debug: bool = False, # 是否开启debug
        routes: Optional[List[BaseRoute]] = None, # 还是直接使用了starletteBaseRoute
        title: str = "FastAPI", # 文档上的标题
        description: str = "", # 文档的上的描述
        version: str = "0.1.0", # 文档对应的版本
        openapi_url: Optional[str] = "/openapi.json", # 当前生成具体文档内容的openapi_url地址
        openapi_tags: Optional[List[Dict[str, Any]]] = None, # 文档具体的标签组，接口分组标签
        servers: Optional[List[Dict[str, Union[str, Any]]]] = None, # 开发文档上不同的服务地址配置
        dependencies: Optional[Sequence[Depends]] = None, # 整个APP应用的依赖注入对象列表，要么是序列类型，Depends
        default_response_class: Type[Response] = Default(JSONResponse), # 默认响应体类的，默认使用的JSONResponse
        docs_url: Optional[str] = "/docs", # 文档地址，设置NONE则不可见，自定义的是需要设置不可见
        redoc_url: Optional[str] = "/redoc", # 文档地址，设置NONE则不可见，自定义的是需要设置不可见
        swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect", # swagger_ui_oauth 授权重定向地址
        swagger_ui_init_oauth: Optional[Dict[str, Any]] = None, #  swagger_ui_init_oauth授权，团队信息
        middleware: Optional[Sequence[Middleware]] = None, # APP 所包含所有中间件
        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,
        terms_of_service: Optional[str] = None, # 文档上swagger_ui_init_oauth授权，团队信息
        contact: Optional[Dict[str, Union[str, Any]]] = None, # 文档上团队内容介绍
        license_info: Optional[Dict[str, Union[str, Any]]] = None, # 文档上swagger_ui-版权信息
        openapi_prefix: str = "",   # 当前生成具体文档内容的openapi_url地址 前缀设置
        root_path: str = "", # 代理的是会用到根访问目录，因为文档访问走代理的是指向的地址有问题，需要通过设置这个来进行通过代理访问
        root_path_in_servers: bool = True, # 是否自动包含root_path_in_servers，也就是把root_path放到servers列表里面
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None, # 绑定到当前路由对应返回响应体
        callbacks: Optional[List[BaseRoute]] = None, # 暂时不知道，好像是用于路由回调什么滴？后面再研究
        deprecated: Optional[bool] = None, # 标记是否路由管理类废弃--是针对路由管理的
        include_in_schema: bool = True, # 标记是路由组是否显示在文档中，也就是包含咋文档里面显示 false则 表示隐藏
        swagger_ui_parameters: Optional[Dict[str, Any]] = None,
        generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
            generate_unique_id
        ), # 路由ID 生成器，如果需要自定义的情况下
        **extra: Any,
    ) -> None:
        self._debug: bool = debug
        self.title = title
        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.extra = extra
        self.openapi_version = "3.0.2"
        self.openapi_schema: Optional[Dict[str, Any]] = None
        if self.openapi_url:
            assert self.title, "A title must be provided for OpenAPI, e.g.: 'My API'"
            assert self.version, "A version must be provided for OpenAPI, e.g.: '2.1.0'"
        # TODO: remove when discarding the openapi_prefix parameter
        if openapi_prefix:
            # 已废弃了！这个处理机制
            logger.warning(
                '"openapi_prefix" has been deprecated in favor of "root_path", which '
                "follows more closely the ASGI standard, is simpler, and more "
                "automatic. Check the docs at "
                "https://fastapi.tiangolo.com/advanced/sub-applications/"
            )
        # 如果存在根目录的情况下，那么久直接设置不然默认就是使用openapi_prefix
        self.root_path = root_path or openapi_prefix
        # 创建一个字典对象
        self.state: State = State()
        # 依赖注入关系
        self.dependency_overrides: Dict[Callable[..., Any], Callable[..., Any]] = {}
        #
        self.router: routing.APIRouter = routing.APIRouter(
            routes=routes,
            #把app中设置的依赖替代项注入到给到路由管理对象里面
            dependency_overrides_provider=self,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
            # 默认响应报文体
            default_response_class=default_response_class,
            # 依赖注入项列表
            dependencies=dependencies,
            # 路由回调
            callbacks=callbacks,
            # 标记是否废弃次路由管理
            deprecated=deprecated,
            # 标记是否显示在文档上
            include_in_schema=include_in_schema,
            # 使用的是哪一个响应该报文
            responses=responses,
            # ID 生成器对象
            generate_unique_id_function=generate_unique_id_function,
        )
        # 处理处理，设置定义错误管理乐类型--如果是响应报文错误的类型对应的则是exception_handlers或其他
        self.exception_handlers: Dict[
            Any, Callable[[Request, Any], Union[Response, Awaitable[Response]]]
        ] = ({} if exception_handlers is None else dict(exception_handlers))
        # 设置默认的处理异常处理器。---HTTPException的错误对应的是http_exception_handler
        self.exception_handlers.setdefault(HTTPException, http_exception_handler)
        # 设置参数校验默认的异常函数处理器
        self.exception_handlers.setdefault(
            RequestValidationError, request_validation_exception_handler
        )

        # 开始组合所有中间件信息
        self.user_middleware: List[Middleware] = (
            [] if middleware is None else list(middleware)
        )
        # 计算出栈顶的中间件，
        self.middleware_stack: ASGIApp = self.build_middleware_stack()
        # 主要用于openapi初始化
        self.setup()

在初始化过程中，我们需要关注点有：

• openapi相关文档的初始化，和路由设置都包含在了 self.setup()里面
• app里面默认的也有一个类似starlette的路由管理类self.router，不过他是基于扩展而来的routing.APIRoute
• 它也实现类似starlette中一个state对象
• 在创建路由管理类的时候，它针对路由文档的一些设置，其中有具体的dependency_overrides_provider依赖树的管理者，指向的是Fastapi-创建出来app自己对象本身，且传递了具体的整个APP设置的全局依赖项dependencies
• 其他如中间栈顶构建和starlett的初始化过程一样，只不过它比starlette多加了一个AsyncExitStackMiddleware的中间件，用于协程退出后一些清理工作

1：关于AsyncExitStackMiddleware中间件

源码为：

from typing import Optional

from fastapi.concurrency import AsyncExitStack
from starlette.types import ASGIApp, Receive, Scope, Send


class AsyncExitStackMiddleware:
    def __init__(self, app: ASGIApp, context_name: str = "fastapi_astack") -> None:
        self.app = app
        self.context_name = context_name

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        if AsyncExitStack:
            dependency_exception: Optional[Exception] = None
            async with AsyncExitStack() as stack:
                # 传入堆栈中上下文名称到scope里面，当前context_name: str = "fastapi_astack"
                scope[self.context_name] = stack
                try:
                   # 传递到下一个APP
                    await self.app(scope, receive, send)
                except Exception as e:
                    dependency_exception = e
                    raise e
            # 依赖异常存在的话，则抛出依赖异常错误
            if dependency_exception:
                # This exception was possibly handled by the dependency but it should
                # still bubble up so that the ServerErrorMiddleware can return a 500
                # or the ExceptionMiddleware can catch and handle any other exceptions
                raise dependency_exception
        else:
            await self.app(scope, receive, send)  # pragma: no cover

从源码可以分析到他是通过实现一个异步上下文管理器来管理对应的退出清理逻辑过程而关于AsyncExitStack的解析如下：

以下的文字说明：来自：https://tsov.net/uupee/52504/

class contextlib.AsyncExitStack
一个异步上下文管理器， 相近 ExitStack，它支持同步和同步上下文管理器的组合，以及具有清理逻辑的协同程序.

close()方法没有实现，aclose()必须使用它.

enter_async_context (cm）
与enter_context()类似，但需要一个异步的contextmanager.

push_async_exit（exit）
相近 push()但是要求异步上下文管理和协同功能.

push_async_callback（callback, *args, **kwds）
与callback()相似但是需要一个协程功能.

aclose（）
相近 close()但正确处理等待.

2：关于openapi的文档路由初始化

主要初始化核心调用在如下代码行中：

 # 主要用于openapi初始化
  self.setup()

源码为：

  def setup(self) -> None:
        # 如果存在openapi_url
        if self.openapi_url:

            urls = (server_data.get("url") for server_data in self.servers)
            # 设置文档的server_urls
            server_urls = {url for url in urls if url}
            # 定义一个内部的协程对象用户返回openapi内的所有的信息

            async def openapi(req: Request) -> JSONResponse:
                # 判断root_path的路径是否存在
                root_path = req.scope.get("root_path", "").rstrip("/")
                # 如果根路径不在对应的设置server_urls里面
                if root_path not in server_urls:
                    # 那么且它存在并且root_path_in_servers在里面的话吗，吧root_path插入到serversURL里面
                    if root_path and self.root_path_in_servers:
                        self.servers.insert(0, {"url": root_path})
                        server_urls.add(root_path)
                    # self.openapi()---》返回.openapi_schema
                    print('self.openapi()',self.openapi())
                return JSONResponse(self.openapi())

            # 添加openapi_url的路由类型，这个路由类型对应需要处理的函数也就是路由内部逻辑处理是：openapi类型的asgiapp
            # 并且此时改路由不显示在文档里面
            self.add_route(self.openapi_url, openapi, include_in_schema=False)
        # 如果openapi_url值存在且默认没有修改docs_url设置的地址，
        if self.openapi_url and self.docs_url:
            # 返回默认的swagger_ui_html类型的HTMLResponse文档
            async def swagger_ui_html(req: Request) -> HTMLResponse:
                # 根据req.scope里面判断root_path是否存在，存在就是删除string字符串末尾的指定/
                #
                root_path = req.scope.get("root_path", "").rstrip("/")
                # 存在则串联起来重新的设置openapi_url的地址
                openapi_url = root_path + self.openapi_url
                # 获取默认设置oauth2_redirect_url的地址
                oauth2_redirect_url = self.swagger_ui_oauth2_redirect_url
                # 如果存在则和上面一样串联起来
                if oauth2_redirect_url:
                    oauth2_redirect_url = root_path + oauth2_redirect_url
                # 获取默认的get_swagger_ui_html的内容，全部HTML的内容
                return get_swagger_ui_html(
                    openapi_url=openapi_url,
                    title=self.title + " - Swagger UI",
                    oauth2_redirect_url=oauth2_redirect_url,
                    init_oauth=self.swagger_ui_init_oauth,
                    swagger_ui_parameters=self.swagger_ui_parameters,
                )

            # 添加docs_url的路由，对应路由函数内部的逻辑是swagger_ui_html协程函数
            self.add_route(self.docs_url, swagger_ui_html, include_in_schema=False)

            # 如果存在授权地址路由
            if self.swagger_ui_oauth2_redirect_url:
                # 定义授权路由地址内部处理函数
                async def swagger_ui_redirect(req: Request) -> HTMLResponse:
                    return get_swagger_ui_oauth2_redirect_html()
                # 添加路由swagger_ui_oauth2_redirect_url路由地址，处理对应路由函数swagger_ui_redirect
                self.add_route(
                    self.swagger_ui_oauth2_redirect_url,
                    swagger_ui_redirect,
                    include_in_schema=False,
                )
        # 如果redoc_url存在，则也添加redoc_url路由的函数处理
        if self.openapi_url and self.redoc_url:
            async def redoc_html(req: Request) -> HTMLResponse:
                root_path = req.scope.get("root_path", "").rstrip("/")
                openapi_url = root_path + self.openapi_url
                return get_redoc_html(
                    openapi_url=openapi_url, title=self.title + " - ReDoc"
                )
            #  # 添加路由redoc_urll路由地址，处理对应路由函数redoc_html
            self.add_route(self.redoc_url, redoc_html, include_in_schema=False)

源码分析：

• 上述代码中，主要是使用了【starlette中的self.add_route】方法来直接的添加具体文档路由信息，所以我们这里可以看到，关于所有交互文档API路由信息不会显示到我们的具体交互文档上，因为它绕过了Fastapi自定义路由添加处理逻辑。
• 并且每个路由处理逻辑都是直接返回HTMLResponse响应报文内容体。从这里我们可以看到后续如果我们需要自定义的自己的交互文档的话，就可以从上面这些地方进行下手了！

如何自定义文档地址后续给出一个具体的方案！

3：关于依赖注入

app初始化过程中，就默认一个空的依赖管理字典，dependency_overrides，他键和值都可以是任意的Callable类型的，如下代码：

self.dependency_overrides: Dict[Callable[..., Any], Callable[..., Any]] = {}

4：关于路由类初始化

 self.router: routing.APIRouter = routing.APIRouter(
            routes=routes,
            #把app中设置的依赖替代项注入到给到路由管理对象里面
            dependency_overrides_provider=self,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
            # 默认响应报文体
            default_response_class=default_response_class,
            # 依赖注入项列表
            dependencies=dependencies,
            # 路由回调
            callbacks=callbacks,
            # 标记是否废弃次路由管理
            deprecated=deprecated,
            # 标记是否显示在文档上
            include_in_schema=include_in_schema,
            # 使用的是哪一个响应该报文
            responses=responses,
            # ID 生成器对象
            generate_unique_id_function=generate_unique_id_function,
        )

其中它是一个 routing.APIRouter类型的路由管理类，由于我们默认路由是通过其他方式添加的，所以routes此时还是空列表，但是在初始化的时候，我们构建了一个全局dependency_overrides_provider依赖注入提供者的管理对象，并且它就是Fastapi中的APP对象的本身，换句话说，就是所有路由对象的依赖注入的提供者都是基于Fastapi中的APP对象自己。

5：关键中间件初始化

    def build_middleware_stack(self) -> ASGIApp:
        # Duplicate/override from Starlette to add AsyncExitStackMiddleware
        # inside of ExceptionMiddleware, inside of custom user middlewares
        debug = self.debug
        error_handler = None
        exception_handlers = {}

        # 处理对应的错误
        for key, value in self.exception_handlers.items():
            if key in (500, Exception):
                error_handler = value
            else:
                exception_handlers[key] = value
        # 开始串联所有的中间处理器，其实就是一些列的ASGIapp协议的对象
        middleware = (
            [Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
            + self.user_middleware
            + [
                Middleware(
                    ExceptionMiddleware, handlers=exception_handlers, debug=debug
                ),
                # Add FastAPI-specific AsyncExitStackMiddleware for dependencies with
                # contextvars.
                # This needs to happen after user middlewares because those create a
                # new contextvars context copy by using a new AnyIO task group.
                # The initial part of dependencies with yield is executed in the
                # FastAPI code, inside all the middlewares, but the teardown part
                # (after yield) is executed in the AsyncExitStack in this middleware,
                # if the AsyncExitStack lived outside of the custom middlewares and
                # contextvars were set in a dependency with yield in that internal
                # contextvars context, the values would not be available in the
                # outside context of the AsyncExitStack.
                # By putting the middleware and the AsyncExitStack here, inside all
                # user middlewares, the code before and after yield in dependencies
                # with yield is executed in the same contextvars context, so all values
                # set in contextvars before yield is still available after yield as
                # would be expected.
                # Additionally, by having this AsyncExitStack here, after the
                # ExceptionMiddleware, now dependencies can catch handled exceptions,
                # e.g. HTTPException, to customize the teardown code (e.g. DB session
                # rollback).
                # 为依赖项添加FastAPI特定的AsyncExitStackMiddleware

                # 上下文变量。

                # 这需要在用户中间件之后发生，因为这些中间件创建了

                # 通过使用新的AnyIO任务组创建新的contextvars上下文副本。

                # 具有产量的依赖项的初始部分在

                # FastAPI代码，在所有中间件中，但拆卸部分除外

                # （在yield之后）在该中间件的AsyncExitStack中执行，

                # 如果AsyncExitStack位于自定义中间件之外

                # contextvars设置在一个依赖项中，在该内部

                # contextvars上下文中的值不可用

                # AsyncExitStack的外部上下文。

                # 通过将中间件和AsyncExitStack放在这里

                # 用户中间件，前后的代码产生依赖关系

                # with yield在相同的contextvars上下文中执行，因此所有值

                # 在contextvars中设置屈服前在屈服后仍然可用

                # 这是意料之中的。

                # 此外，在

                # ExceptionMiddleware，现在依赖项可以捕获已处理的异常，

                # 例如HTTPException，自定义拆卸代码（例如DB会话

                # 回滚）。
                Middleware(AsyncExitStackMiddleware),
            ]
        )

        # 路由管理类型
        app = self.router
        # 开始逆序之后串联所有的APP对象
        for cls, options in reversed(middleware):
            app = cls(app=app, **options)
        return app

从上述的代码可以看到，这些和我们的之前starlette框架构建是如出一辙。

app对象各种实例方法：

1：路由添加处理

add_route方式，其中它是直接调用了 self.router.add_route的方法，所有最终还是回调父类的add_route的方法去添加路由节点到路由列表上

    def add_route(
        self,
        path: str,
        route: typing.Callable,
        methods: typing.Optional[typing.List[str]] = None,
        name: typing.Optional[str] = None,
        include_in_schema: bool = True,
    ) -> None:  # pragma: no cover
        self.router.add_route(
            path, route, methods=methods, name=name, include_in_schema=include_in_schema
        )

而route则是对add_route方法装饰器一种封装

    def route(
        self,
        path: str,
        methods: typing.Optional[typing.List[str]] = None,
        name: typing.Optional[str] = None,
        include_in_schema: bool = True,
    ) -> typing.Callable:  # pragma: nocover
        """
        We no longer document this decorator style API, and its usage is discouraged.
        Instead you should use the following approach:

        routes = [
            Route(path, endpoint=..., ...),
            ...
        ]

        app = Starlette(routes=routes)
        """

        def decorator(func: typing.Callable) -> typing.Callable:
            self.add_route(
                path,
                func,
                methods=methods,
                name=name,
                include_in_schema=include_in_schema,
            )
            return func

        return decorator

2：自定义中间件的处理，每次添加后都重构build_middleware_stack

    # 代码方式事件
        self, middleware_class: type, **options: typing.Any
    ) -> None:  # pragma: no cover
        self.user_middleware.insert(0, Middleware(middleware_class, **options))
        self.middleware_stack = self.build_middleware_stack()
        
      
    # 装饰器方式实现中间件添加
     def middleware(self, middleware_type: str) -> typing.Callable:  # pragma: nocover
        assert (
            middleware_type == "http"
        ), 'Currently only middleware("http") is supported.'

        def decorator(func: typing.Callable) -> typing.Callable:
            self.add_middleware(BaseHTTPMiddleware, dispatch=func)
            return func

        return decorator

3 自定义异常处理-需要重构build_middleware_stack

    # 代码方式添加
    def add_exception_handler(
        self,
        exc_class_or_status_code: typing.Union[int, typing.Type[Exception]],
        handler: typing.Callable,
    ) -> None:  # pragma: no cover
        self.exception_handlers[exc_class_or_status_code] = handler
        self.middleware_stack = self.build_middleware_stack()
    
    # 装饰器方式的实现
    def exception_handler(
        self, exc_class_or_status_code: typing.Union[int, typing.Type[Exception]]
    ) -> typing.Callable:  # pragma: nocover
        def decorator(func: typing.Callable) -> typing.Callable:
            self.add_exception_handler(exc_class_or_status_code, func)
            return func

        return decorator

4 给路由添加启动和关闭事件的回调处理

    def add_event_handler(
        self, event_type: str, func: typing.Callable
    ) -> None:  # pragma: no cover
        self.router.add_event_handler(event_type, func)

5）include_router添加路由组对象

 def include_router(
        self,
        router: routing.APIRouter,
        *,
        prefix: str = "",
        tags: Optional[List[Union[str, Enum]]] = None,
        dependencies: Optional[Sequence[Depends]] = None,
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        deprecated: Optional[bool] = None,
        include_in_schema: bool = True,
        default_response_class: Type[Response] = Default(JSONResponse),
        callbacks: Optional[List[BaseRoute]] = None,
        generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
            generate_unique_id
        ),
    ) -> None:
        #app里面的 self.router ：outer: routing.APIRouter,
        # 是路由组对象 还可以再添加路由组-router
        # 路由组里面再添加路由节点
        self.router.include_router(
            router,
            prefix=prefix,
            tags=tags,
            dependencies=dependencies,
            responses=responses,
            deprecated=deprecated,
            include_in_schema=include_in_schema,
            default_response_class=default_response_class,
            callbacks=callbacks,
            generate_unique_id_function=generate_unique_id_function,
        )

6）其他路由装饰器的实现

    def get(
        self,
        path: str,
        *,
        response_model: Optional[Type[Any]] = None,
        status_code: Optional[int] = None,
        tags: Optional[List[Union[str, Enum]]] = None,
        dependencies: Optional[Sequence[Depends]] = None,
        summary: Optional[str] = None,
        description: Optional[str] = None,
        response_description: str = "Successful Response",
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        deprecated: Optional[bool] = None,
        operation_id: Optional[str] = None,
        response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
        response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
        response_model_by_alias: bool = True,
        response_model_exclude_unset: bool = False,
        response_model_exclude_defaults: bool = False,
        response_model_exclude_none: bool = False,
        include_in_schema: bool = True,
        response_class: Type[Response] = Default(JSONResponse),
        name: Optional[str] = None,
        callbacks: Optional[List[BaseRoute]] = None,
        openapi_extra: Optional[Dict[str, Any]] = None,
        generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
            generate_unique_id
        ),
    ) -> Callable[[DecoratedCallable], DecoratedCallable]:
       # 调用是自身路由组里面self.router的put，调用自定义self.api_route-添加路由节点在-创建具体路由几点route_class
        return self.router.get(
            path,
            response_model=response_model,
            status_code=status_code,
            tags=tags,
            dependencies=dependencies,
            summary=summary,
            description=description,
            response_description=response_description,
            responses=responses,
            deprecated=deprecated,
            operation_id=operation_id,
            response_model_include=response_model_include,
            response_model_exclude=response_model_exclude,
            response_model_by_alias=response_model_by_alias,
            response_model_exclude_unset=response_model_exclude_unset,
            response_model_exclude_defaults=response_model_exclude_defaults,
            response_model_exclude_none=response_model_exclude_none,
            include_in_schema=include_in_schema,
            response_class=response_class,
            name=name,
            callbacks=callbacks,
            openapi_extra=openapi_extra,
            generate_unique_id_function=generate_unique_id_function,
        )

    def put(
       ---省略
    def post(
       ---省略

    def delete(
       ---省略

    def options(
      ---省略

    def head(
      ---省略

    def patch(
       ---省略

    def trace(
       ---省略

3  创建app = FastAPI()全局依赖和路由分析

上面我们的默认创建一个空没有任何参数APP对象，我们知道FastAPI可以默认创建全局APP服务依赖，所以，此时我们的可以通过全局依赖做各种服务启动的配置校验之类的事情，如下示例代码：

import pathlib

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi import Depends
from api.user.login import user
from api.user.user_info import userInfo
from custom_swagger.custom_swagger import getInstance
from fastapi.responses import PlainTextResponse

def test_dependencie_1():
    print("执行了-全局依赖1")
    return "全局依赖1"


def test_dependencie_2():
    print("执行了-全局依赖2")
    return "全局依赖1"

app = FastAPI(docs_url=None, redoc_url=None,dependencies=[Depends(test_dependencie_1),Depends(test_dependencie_2)])

@app.get('/test_dependencie')
def test_dependencie():
    return PlainTextResponse('测试全局依赖执行！')

app.mount("/static", StaticFiles(directory=f"{pathlib.Path.cwd()}/static"), name="static")
app.include_router(user)  # 用户模块
app.include_router(userInfo)  # 用户信息模块

getInstance(app)

if __name__ == '__main__':
    import uvicorn

    # 当开启了热启动，就是改动代码就自动重启的时候，需要使用字符串方式导入模块
    uvicorn.run('main:app', reload=True)

代码解析：

• 上述代码中创建了app = FastAPI，且通过dependencies=[Depends(test_dependencie_1),Depends(test_dependencie_2)])包含两个依赖注入的项，

• 此时我们定义一个测试路由的，且访问该路由地址此时输出为：

通过上面接口请求访问，我们可以看到全局依赖注入已经在执行了！那么它是如何注入到我们的路由中的呢？答案就是在初始化的时候，交付给了自定义路由管理类：

自定义的APIRouter分析-基类分析

注意下面两个类：

通过源码查看APIRouter，它是继承于starlette中的Router，而Router在前面我们也分析过了，如下源码：

class Router:
    def __init__(
        self,
        routes: typing.Sequence[BaseRoute] = None,
        redirect_slashes: bool = True,
        default: ASGIApp = None,
        on_startup: typing.Sequence[typing.Callable] = None,
        on_shutdown: typing.Sequence[typing.Callable] = None,
        lifespan: typing.Callable[[typing.Any], typing.AsyncContextManager] = None,
    ) -> None:
        # 要管理的所有路由节点实例
        self.routes = [] if routes is None else list(routes)
        # 是否要开启路由斜杠重定向问题
        self.redirect_slashes = redirect_slashes
        # 处理没有找到路由的时候，处理函数，是否使用默认或自定义的
        self.default = self.not_found if default is None else default
        # 启动事件回调列表
        self.on_startup = [] if on_startup is None else list(on_startup)
        # 关闭事件回调列表
        self.on_shutdown = [] if on_shutdown is None else list(on_shutdown)
        # 启动和关闭另一种处理方式
        if lifespan is None:
            self.lifespan_context: typing.Callable[
                [typing.Any], typing.AsyncContextManager
            ] = _DefaultLifespan(self)

        elif inspect.isasyncgenfunction(lifespan):
            warnings.warn(
                "async generator function lifespans are deprecated, "
                "use an @contextlib.asynccontextmanager function instead",
                DeprecationWarning,
            )
            self.lifespan_context = asynccontextmanager(
                lifespan,  # type: ignore[arg-type]
            )
        elif inspect.isgeneratorfunction(lifespan):
            warnings.warn(
                "generator function lifespans are deprecated, "
                "use an @contextlib.asynccontextmanager function instead",
                DeprecationWarning,
            )
            self.lifespan_context = _wrap_gen_lifespan_context(
                lifespan,  # type: ignore[arg-type]
            )
        else:
            self.lifespan_context = lifespan

    async def not_found(self, scope: Scope, receive: Receive, send: Send) -> None:
       # 路由匹配失败的时候处理函数
        if scope["type"] == "websocket":
            websocket_close = WebSocketClose()
            await websocket_close(scope, receive, send)
            return

        # If we're running inside a starlette application then raise an
        # exception, so that the configurable exception handler can deal with
        # returning the response. For plain ASGI apps, just return the response.
        if "app" in scope:
            raise HTTPException(status_code=404)
        else:
            response = PlainTextResponse("Not Found", status_code=404)
        await response(scope, receive, send)
       
    def url_path_for(self, name: str, **path_params: str) -> URLPath:
        # 路由url路由提取。根据名字
        for route in self.routes:
            try:
                return route.url_path_for(name, **path_params)
            except NoMatchFound:
                pass
        # 找不到就抛出无匹配异常
        raise NoMatchFound()

    async def startup(self) -> None:
        """
        Run any `.on_startup` event handlers.
        """
        for handler in self.on_startup:
            if asyncio.iscoroutinefunction(handler):
                await handler()
            else:
                handler()

    async def shutdown(self) -> None:
        """
        Run any `.on_shutdown` event handlers.
        """
        for handler in self.on_shutdown:
            if asyncio.iscoroutinefunction(handler):
                await handler()
            else:
                handler()

    async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        Handle ASGI lifespan messages, which allows us to manage application
        startup and shutdown events.
        """
        started = False
        app = scope.get("app")
        await receive()
        try:
            async with self.lifespan_context(app):
                await send({"type": "lifespan.startup.complete"})
                started = True
                await receive()
        except BaseException:
            exc_text = traceback.format_exc()
            if started:
                await send({"type": "lifespan.shutdown.failed", "message": exc_text})
            else:
                await send({"type": "lifespan.startup.failed", "message": exc_text})
            raise
        else:
            await send({"type": "lifespan.shutdown.complete"})

    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
        """
        The main entry point to the Router class.
        """
        assert scope["type"] in ("http", "websocket", "lifespan")
        print('原来你是在这里的啊！！！！！！！！！！！！！！！！！！')
        if "router" not in scope:
            scope["router"] = self

        if scope["type"] == "lifespan":
            await self.lifespan(scope, receive, send)
            return

        partial = None
        # 开始遍历所有路由节点进行匹配，
        for route in self.routes:
            # Determine if any route matches the incoming scope,
            # and hand over to the matching route if found.
            print("当前请求的路由信息是1",route)
            match, child_scope = route.matches(scope)
            print("当前请求的路由信息是2", match,child_scope)
            if match == Match.FULL:
                # 开始动态的添加我们的路由上的child_scope信息到scope上下文中
                scope.update(child_scope)
                # 开始调用路由对象的handle方法执行具体的路由内部的逻辑
                print("开始调用路由对象的handle方法执行具体的路由内部的逻辑递下来到了路由！")
                # 匹配成功开始执行路由节点的handle的逻辑处理
                await route.handle(scope, receive, send)
                return
            elif match == Match.PARTIAL and partial is None:
                partial = route
                partial_scope = child_scope
        
        # 判断是否是一个偏函数的类型路由？这里目前还有点模糊
        if partial is not None:
            #  Handle partial matches. These are cases where an endpoint is
            # able to handle the request, but is not a preferred option.
            # We use this in particular to deal with "405 Method Not Allowed".
            scope.update(partial_scope)
            await partial.handle(scope, receive, send)
            return
        # 根据斜杠问题处理重定向问题
        if scope["type"] == "http" and self.redirect_slashes and scope["path"] != "/":
            redirect_scope = dict(scope)
            if scope["path"].endswith("/"):
                redirect_scope["path"] = redirect_scope["path"].rstrip("/")
            else:
                redirect_scope["path"] = redirect_scope["path"] + "/"

            for route in self.routes:
                match, child_scope = route.matches(redirect_scope)
                if match != Match.NONE:
                    redirect_url = URL(scope=redirect_scope)
                    response = RedirectResponse(url=str(redirect_url))
                    # 重定向直接返回response退出整个循环
                    await response(scope, receive, send)
                    return

        # 如果都匹配不到，则调用路由管理自带的default函数
        #也就是：self.default = self.not_found if default is None else default
        await self.default(scope, receive, send)

    def __eq__(self, other: typing.Any) -> bool:
        return isinstance(other, Router) and self.routes == other.routes

    # The following usages are now discouraged in favour of configuration
    #  during Router.__init__(...)
    def mount(self, path: str, app: ASGIApp, name: str = None) -> None:
        # 挂载方式的路由处理
        route = Mount(path, app=app, name=name)
        self.routes.append(route)

    def host(self, host: str, app: ASGIApp, name: str = None) -> None:
       # Hsot方式路由处理
        route = Host(host, app=app, name=name)
        self.routes.append(route)

    def add_route(
        self,
        path: str,
        endpoint: typing.Callable,
        methods: typing.List[str] = None,
        name: str = None,
        include_in_schema: bool = True,
    ) -> None:
        # 实例化路由节点对象
        route = Route(
            #路径
            path,
            # 端点函数
            endpoint=endpoint,
            # 方法
            methods=methods,
            name=name,
            include_in_schema=include_in_schema,
        )
        self.routes.append(route)

    def add_websocket_route(
        self, path: str, endpoint: typing.Callable, name: str = None
    ) -> None:
        route = WebSocketRoute(path, endpoint=endpoint, name=name)
        self.routes.append(route)

    def route(
        self,
        path: str,
        methods: typing.List[str] = None,
        name: str = None,
        include_in_schema: bool = True,
    ) -> typing.Callable:
       # 使用route装饰器类型来装饰添加路由
        def decorator(func: typing.Callable) -> typing.Callable:
            self.add_route(
                path,
                func,
                methods=methods,
                name=name,
                include_in_schema=include_in_schema,
            )
            return func

        return decorator

    def websocket_route(self, path: str, name: str = None) -> typing.Callable:
        def decorator(func: typing.Callable) -> typing.Callable:
            self.add_websocket_route(path, func, name=name)
            return func

        return decorator

    def add_event_handler(self, event_type: str, func: typing.Callable) -> None:
        assert event_type in ("startup", "shutdown")
        # 路由中启动事件和关闭事件的添加
        if event_type == "startup":
            self.on_startup.append(func)
        else:
            self.on_shutdown.append(func)

    def on_event(self, event_type: str) -> typing.Callable:
        # 使用装饰器方式添加事件的处理
        def decorator(func: typing.Callable) -> typing.Callable:
            self.add_event_handler(event_type, func)
            return func

        return decorator

通过内部，他后续会自行调用下一层的self.app对象，所有如果我们要集成他扩展，值需要进行self.app封装重写，就可以实现自定义的self.app的调用。

自定义的APIRouter分析-扩展之后的APIRouter有了哪些增强的功能点呢？

class APIRouter(routing.Router):
    def __init__(
        self,
        *,
        prefix: str = "",
        tags: Optional[List[Union[str, Enum]]] = None,
        dependencies: Optional[Sequence[params.Depends]] = None,
        default_response_class: Type[Response] = Default(JSONResponse),
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        callbacks: Optional[List[BaseRoute]] = None,
        routes: Optional[List[routing.BaseRoute]] = None,
        redirect_slashes: bool = True,
        default: Optional[ASGIApp] = None,
        dependency_overrides_provider: Optional[Any] = None,
        route_class: Type[APIRoute] = APIRoute,
        on_startup: Optional[Sequence[Callable[[], Any]]] = None,
        on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
        deprecated: Optional[bool] = None,
        include_in_schema: bool = True,
        generate_unique_id_function: Callable[[APIRoute], str] = Default(
            generate_unique_id
        ),
    ) -> None:
       # 父类初始化照样的进行调用
        super().__init__(
            routes=routes,
            redirect_slashes=redirect_slashes,
            default=default,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
        )
        # 扩展点，判断是否设置了路由前缀prefix
        if prefix:
            # 断言前缀是否存在"/"开始
            assert prefix.startswith("/"), "A path prefix must start with '/'"
            # 断言前缀是否存在"/"结束
            assert not prefix.endswith(
                "/"
            ), "A path prefix must not end with '/', as the routes will start with '/'"
        # 前缀处理
        self.prefix = prefix
        # 路由管理租标签，路由组标签列表，他可以是字符串类型，也可以是Enum类型
        self.tags: List[Union[str, Enum]] = tags or []
        # 默认的依赖注入的列表--这里是全局依赖注入的列表，dependencies开始列表赋值
        self.dependencies = list(dependencies or []) or []
        # 标记是否废弃
        self.deprecated = deprecated
        # 是否包含在文档
        self.include_in_schema = include_in_schema
        # 默认responses 处理
        self.responses = responses or {}
        # 回调处理--初步判断是用于文档中显示返回信息体？但是还不确定，后续再看
        self.callbacks = callbacks or []
        # 依赖注入的提供者--其实就是fastapi的app对象本身
        self.dependency_overrides_provider = dependency_overrides_provider
        # 定义路由类
        self.route_class = route_class
        # 默认default_response_class
        self.default_response_class = default_response_class
        # 路由函数的生成unique_id_function 
        self.generate_unique_id_function = generate_unique_id_function
    
    #自定义扩展添加理由的方法
    def add_api_route(
        self,
        path: str,
        endpoint: Callable[..., Any],
        *,
        response_model: Optional[Type[Any]] = None,
        status_code: Optional[int] = None,
        tags: Optional[List[Union[str, Enum]]] = None,
        dependencies: Optional[Sequence[params.Depends]] = None,
        summary: Optional[str] = None,
        description: Optional[str] = None,
        response_description: str = "Successful Response",
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        deprecated: Optional[bool] = None,
        methods: Optional[Union[Set[str], List[str]]] = None,
        operation_id: Optional[str] = None,
        response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
        response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
        response_model_by_alias: bool = True,
        response_model_exclude_unset: bool = False,
        response_model_exclude_defaults: bool = False,
        response_model_exclude_none: bool = False,
        include_in_schema: bool = True,
        response_class: Union[Type[Response], DefaultPlaceholder] = Default(
            JSONResponse
        ),
        name: Optional[str] = None,
        route_class_override: Optional[Type[APIRoute]] = None,
        callbacks: Optional[List[BaseRoute]] = None,
        openapi_extra: Optional[Dict[str, Any]] = None,
        generate_unique_id_function: Union[
            Callable[[APIRoute], str], DefaultPlaceholder
        ] = Default(generate_unique_id),
    ) -> None:
        # 是否存在route_class_override覆盖类传入，没有则在判断是否存在定义路由类self.route_class
        # 路由类对象赋值
        route_class = route_class_override or self.route_class
        # 响应体类型，通过设置不同的类型，可以根据返回不同的响应码的时候，匹配不同的responses进行返回
        responses = responses or {}
        # 合并combined_responses
        combined_responses = {**self.responses, **responses}
        # 获取默认response_class类型
        current_response_class = get_value_or_default(
            response_class, self.default_response_class
        )
        # 当前路由组标签列表拷贝--因为是独立路由对象，需要拷贝一层
        # 但是这里需要注意点是：
        #1.copy()方法对于List来说，其第一层，是实现了深拷贝，但对于其内嵌套的List，仍然是浅拷贝。
        #因为嵌套的List保存的是地址，复制过去的时候是把地址复制过去了，嵌套的List在内存中指向的还是同一个。
        # current_tags = self.tags.copy() 所以可以理解是一个深拷贝，就是独立一个互不影响的一个拷贝
        current_tags = self.tags.copy()
        # 如果除了基础设置 current_tags，还有新增自定义的tags，则扩展添加tags到current_tags
        # 合并基础的设定的tags+当前路由自定义tags
        if tags:
            current_tags.extend(tags)
        # 当前依赖也进行一次深拷贝，因为dependencies是序列类型的对象
        current_dependencies = self.dependencies.copy()
        # 在全局之上，又新增新的扩展
        if dependencies:
            current_dependencies.extend(dependencies)
        # 回调也拷贝一层
        current_callbacks = self.callbacks.copy()
        if callbacks:
            current_callbacks.extend(callbacks)
        # 当前函数生成，用于那种路由需要自定义ID的那种
        current_generate_unique_id = get_value_or_default(
            generate_unique_id_function, self.generate_unique_id_function
        )
        
        #  route_class_override=Optional[Type[APIRoute]]指向的是一个APIRoute类型路由
        # 而APIRoute也starlette下的class APIRoute(routing.Route):的子类的扩展能实现
        
        route = route_class(
            # 路由前缀
            self.prefix + path,
            # 路由端点
            endpoint=endpoint,
            # 响应模型
            response_model=response_model,
            # 响应码
            status_code=status_code,
            # 路由自身所属的路由组
            tags=current_tags,
            # 路由自身所属的依赖项
            dependencies=current_dependencies,
            # 路由文档上显示的函数标题没有，则默认是函数的名称
            summary=summary,
            # 路由功能点描述
            description=description,
            # 路由响应体报文中描述
            response_description=response_description,
            # 路由对应响应体对象处理，会自动提到这一个响应体里面指向返回
            responses=combined_responses,
            # 是否标记废弃
            deprecated=deprecated or self.deprecated,
            # 路由方法
            methods=methods,
            # 函数对应operation_id，好像，也就是和上面就是路由唯一标记，？这里暂时不详它的作用
            operation_id=operation_id,
            # 响应体模型相关过滤处理
            response_model_include=response_model_include,
             # 响应体模型相关过滤处理
            response_model_exclude=response_model_exclude,
             # 响应体模型相关过滤处理
            response_model_by_alias=response_model_by_alias,
             # 响应体模型相关过滤处理
            response_model_exclude_unset=response_model_exclude_unset,
             # 响应体模型相关过滤处理
            response_model_exclude_defaults=response_model_exclude_defaults,
             # 响应体模型相关过滤处理
            response_model_exclude_none=response_model_exclude_none,
            # 是否显示在文档上的标记
            include_in_schema=include_in_schema and self.include_in_schema,
            # 具体response_class类的设置，自定义类型的responses
            response_class=current_response_class,
            # 路由名称，
            name=name,
            # 把依赖注入的提供者也就是fstapi的app对象注入
            dependency_overrides_provider=self.dependency_overrides_provider,
            # 回调
            callbacks=current_callbacks,
            # openapi_extra 用于文档上一些扩展字段显示，主要是和文档相关显示模型，和自定义显示一些字段扩展
            openapi_extra=openapi_extra,
            # generate_unique_id_function生成器
            generate_unique_id_function=current_generate_unique_id,
        )
        
        # 添加到starlette中的Router里面的routes中
        self.routes.append(route)
    
    ## api_route 是add_api_route 装饰器写法的实现
    def api_route()
        // 省略部分代码
        return decorator
    
    # websocket_route类型路由实现
    def add_api_websocket_route(
        self, path: str, endpoint: Callable[..., Any], name: Optional[str] = None
    ) -> None:
        route = APIWebSocketRoute(
            self.prefix + path,
            endpoint=endpoint,
            name=name,
            dependency_overrides_provider=self.dependency_overrides_provider,
        )
        self.routes.append(route)
   # websocket是 add_api_websocket_route装饰器写法的实现
    def websocket(
        self, path: str, name: Optional[str] = None
    ) -> Callable[[DecoratedCallable], DecoratedCallable]:
        ····
        return decorator
    
    # 路由组里面还有嵌套路由的方法-------，这个有别app的include_router
    def include_router(
        self,
        router: "APIRouter",
        *,
        prefix: str = "",
        tags: Optional[List[Union[str, Enum]]] = None,
        dependencies: Optional[Sequence[params.Depends]] = None,
        default_response_class: Type[Response] = Default(JSONResponse),
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        callbacks: Optional[List[BaseRoute]] = None,
        deprecated: Optional[bool] = None,
        include_in_schema: bool = True,
        generate_unique_id_function: Callable[[APIRoute], str] = Default(
            generate_unique_id
        ),
    ) -> None:
        if prefix:
            assert prefix.startswith("/"), "A path prefix must start with '/'"
            assert not prefix.endswith(
                "/"
            ), "A path prefix must not end with '/', as the routes will start with '/'"
        else:
            for r in router.routes:
                path = getattr(r, "path")
                name = getattr(r, "name", "unknown")
                if path is not None and not path:
                    raise Exception(
                        f"Prefix and path cannot be both empty (path operation: {name})"
                    )
        if responses is None:
            responses = {}
        # 路由管理类，里面还有一个router.routes
        for route in router.routes:
            # 如果路由类型APIRoute自定义扩展APIRoute
            if isinstance(route, APIRoute):
                # 扩展responses设置
                combined_responses = {**responses, **route.responses}
                use_response_class = get_value_or_default(
                    route.response_class,
                    router.default_response_class,
                    default_response_class,
                    self.default_response_class,
                )
                current_tags = []
                if tags:
                    current_tags.extend(tags)
                if route.tags:
                    current_tags.extend(route.tags)
                current_dependencies: List[params.Depends] = []
                if dependencies:
                    current_dependencies.extend(dependencies)
                if route.dependencies:
                    current_dependencies.extend(route.dependencies)
                current_callbacks = []
                if callbacks:
                    current_callbacks.extend(callbacks)
                if route.callbacks:
                    current_callbacks.extend(route.callbacks)
                current_generate_unique_id = get_value_or_default(
                    route.generate_unique_id_function,
                    router.generate_unique_id_function,
                    generate_unique_id_function,
                    self.generate_unique_id_function,
                )
                # 层层嵌套再实现
                self.add_api_route(
                    prefix + route.path,
                    route.endpoint,
                    response_model=route.response_model,
                    status_code=route.status_code,
                    tags=current_tags,
                    dependencies=current_dependencies,
                    summary=route.summary,
                    description=route.description,
                    response_description=route.response_description,
                    responses=combined_responses,
                    deprecated=route.deprecated or deprecated or self.deprecated,
                    methods=route.methods,
                    operation_id=route.operation_id,
                    response_model_include=route.response_model_include,
                    response_model_exclude=route.response_model_exclude,
                    response_model_by_alias=route.response_model_by_alias,
                    response_model_exclude_unset=route.response_model_exclude_unset,
                    response_model_exclude_defaults=route.response_model_exclude_defaults,
                    response_model_exclude_none=route.response_model_exclude_none,
                    include_in_schema=route.include_in_schema
                    and self.include_in_schema
                    and include_in_schema,
                    response_class=use_response_class,
                    name=route.name,
                    route_class_override=type(route),
                    callbacks=current_callbacks,
                    openapi_extra=route.openapi_extra,
                    generate_unique_id_function=current_generate_unique_id,
                )
            # 如果不是自定义，而是父类的也就是starlette中的Route类型的话
            elif isinstance(route, routing.Route):
                # 无法添加其他初始化信息
                methods = list(route.methods or [])
                self.add_route(
                    prefix + route.path,
                    route.endpoint,
                    methods=methods,
                    include_in_schema=route.include_in_schema,
                    name=route.name,
                )
            elif isinstance(route, APIWebSocketRoute):
                self.add_api_websocket_route(
                    prefix + route.path, route.endpoint, name=route.name
                )
            elif isinstance(route, routing.WebSocketRoute):
                self.add_websocket_route(
                    prefix + route.path, route.endpoint, name=route.name
                )
        # 给路由添加启注册时间
        for handler in router.on_startup:
            self.add_event_handler("startup", handler)
        # 给路由添加关闭事件
        for handler in router.on_shutdown:
            self.add_event_handler("shutdown", handler)
    
    # 路由管理类APIRouter中的get装饰器方法的实现
    def get(
        self,
        path: str,
        *,
        response_model: Optional[Type[Any]] = None,
        status_code: Optional[int] = None,
        tags: Optional[List[Union[str, Enum]]] = None,
        dependencies: Optional[Sequence[params.Depends]] = None,
        summary: Optional[str] = None,
        description: Optional[str] = None,
        response_description: str = "Successful Response",
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        deprecated: Optional[bool] = None,
        operation_id: Optional[str] = None,
        response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
        response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
        response_model_by_alias: bool = True,
        response_model_exclude_unset: bool = False,
        response_model_exclude_defaults: bool = False,
        response_model_exclude_none: bool = False,
        include_in_schema: bool = True,
        response_class: Type[Response] = Default(JSONResponse),
        name: Optional[str] = None,
        callbacks: Optional[List[BaseRoute]] = None,
        openapi_extra: Optional[Dict[str, Any]] = None,
        generate_unique_id_function: Callable[[APIRoute], str] = Default(
            generate_unique_id
        ),
    ) -> Callable[[DecoratedCallable], DecoratedCallable]:
       # 看返回类型是Callable[[DecoratedCallable], DecoratedCallable]
       # 内部调用其实还是api_route的装饰器
        return self.api_route(
            path=path,
            response_model=response_model,
            status_code=status_code,
            tags=tags,
            dependencies=dependencies,
            summary=summary,
            description=description,
            response_description=response_description,
            responses=responses,
            deprecated=deprecated,
            methods=["GET"],
            operation_id=operation_id,
            response_model_include=response_model_include,
            response_model_exclude=response_model_exclude,
            response_model_by_alias=response_model_by_alias,
            response_model_exclude_unset=response_model_exclude_unset,
            response_model_exclude_defaults=response_model_exclude_defaults,
            response_model_exclude_none=response_model_exclude_none,
            include_in_schema=include_in_schema,
            response_class=response_class,
            name=name,
            callbacks=callbacks,
            openapi_extra=openapi_extra,
            generate_unique_id_function=generate_unique_id_function,
        )

    def put()
        `````省略

    def post() -> Callable[[DecoratedCallable], DecoratedCallable]:
        `````省略

    def delete(
       ) -> Callable[[DecoratedCallable], DecoratedCallable]:
        `````省略

    def options()
       `````省略

    def head(
       `````省略

    def patch(
      `````省略
       
    def trace(
     `````省略

自定义路由节点中类分析

因为自定义实现的APIRoute(routing.Route是执行想跟具体业务逻辑的地方，也是分析依赖注入执行的地方，所以这里需要具体分析。

class APIRoute(routing.Route):
    def __init__(
        self,
        path: str,
        endpoint: Callable[..., Any],
        *,
        response_model: Optional[Type[Any]] = None,
        status_code: Optional[int] = None,
        tags: Optional[List[Union[str, Enum]]] = None,
        dependencies: Optional[Sequence[params.Depends]] = None,
        summary: Optional[str] = None,
        description: Optional[str] = None,
        response_description: str = "Successful Response",
        responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
        deprecated: Optional[bool] = None,
        name: Optional[str] = None,
        methods: Optional[Union[Set[str], List[str]]] = None,
        operation_id: Optional[str] = None,
        response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
        response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
        response_model_by_alias: bool = True,
        response_model_exclude_unset: bool = False,
        response_model_exclude_defaults: bool = False,
        response_model_exclude_none: bool = False,
        include_in_schema: bool = True,
        response_class: Union[Type[Response], DefaultPlaceholder] = Default(
            JSONResponse
        ),
        dependency_overrides_provider: Optional[Any] = None,
        callbacks: Optional[List[BaseRoute]] = None,
        openapi_extra: Optional[Dict[str, Any]] = None,
        generate_unique_id_function: Union[
            Callable[["APIRoute"], str], DefaultPlaceholder
        ] = Default(generate_unique_id),
    ) -> None:
        # 路由节点路径
        self.path = path
        # 路由节点的端点，也就是逻辑需要处理业务逻辑的函数或方法或类
        self.endpoint = endpoint
        # 业务处理逻辑完成之后返回响应体的模型类
        self.response_model = response_model
        # 路由节点的end标题描述
        self.summary = summary
        # 路由节点响应体信息描述
        self.response_description = response_description
        # 是否废弃
        self.deprecated = deprecated
        # 路由节点ID信息？暂时不详具体作用
        self.operation_id = operation_id
        # 路由响应模型的中过滤机制处理规则
        self.response_model_include = response_model_include
        self.response_model_exclude = response_model_exclude
        self.response_model_by_alias = response_model_by_alias
        self.response_model_exclude_unset = response_model_exclude_unset
        self.response_model_exclude_defaults = response_model_exclude_defaults
        self.response_model_exclude_none = response_model_exclude_none
         # 是否显示在文档上
        self.include_in_schema = include_in_schema
        # 响应体具体类。模型和类不一样
        self.response_class = response_class
        # 路由节点依赖注入提供者--其实就是app对象
        self.dependency_overrides_provider = dependency_overrides_provider
        # 回调
        self.callbacks = callbacks
        # 文档扩展字段
        self.openapi_extra = openapi_extra
        # 函数ID生成器
        self.generate_unique_id_function = generate_unique_id_function
        # 路由节点所属组标签
        self.tags = tags or []
        # 响应体--字典类型的响应体内容
        self.responses = responses or {}
        # 路由名称
        self.name = get_name(endpoint) if name is None else name
        # 路由路径地址合并。嵌套路由处理机制
        self.path_regex, self.path_format, self.param_convertors = compile_path(path)
        # 方法列表
        if methods is None:
            methods = ["GET"]
        # 支持方法列表的合并
        self.methods: Set[str] = {method.upper() for method in methods}
        #current_generate_unique_id 生成
        if isinstance(generate_unique_id_function, DefaultPlaceholder):
            current_generate_unique_id: Callable[
                ["APIRoute"], str
            ] = generate_unique_id_function.value
        else:
            current_generate_unique_id = generate_unique_id_function
        # unique_id最终值信息
        self.unique_id = self.operation_id or current_generate_unique_id(self)
        # normalize enums e.g. http.HTTPStatus
        # 状态响应码
        if isinstance(status_code, IntEnum):
            status_code = int(status_code)
        self.status_code = status_code
        # 如果设置有响应模型类----这个后续还得继续分析，也就是如何字典创建返回响应体模型，
        # 
        if self.response_model:
            assert (
                status_code not in STATUS_CODES_WITH_NO_BODY
            ), f"Status code {status_code} must not have a response body"
            response_name = "Response_" + self.unique_id
            # 开始创建响应模型字典，跟名称和类型
            self.response_field = create_response_field(
                name=response_name, type_=self.response_model
            )
            # Create a clone of the field, so that a Pydantic submodel is not returned
            # as is just because it's an instance of a subclass of a more limited class
            # e.g. UserInDB (containing hashed_password) could be a subclass of User
            # that doesn't have the hashed_password. But because it's a subclass, it
            # would pass the validation and be returned as is.
            # By being a new field, no inheritance will be passed as is. A new model
            # will be always created.
            self.secure_cloned_response_field: Optional[
                ModelField
            ] = create_cloned_field(self.response_field)
        else:
            self.response_field = None  # type: ignore
            self.secure_cloned_response_field = None
        # 依赖注入项的处理
        if dependencies:
            self.dependencies = list(dependencies)
        else:
            self.dependencies = []
            
        # 直接回获取类或函数中文档描述信息 inspect.cleandoc(self.endpoint.__doc__ or "")
        self.description = description or inspect.cleandoc(self.endpoint.__doc__ or "")
        # if a "form feed" character (page break) is found in the description text,
        # truncate description text to the content preceding the first "form feed"
        # 对描述信息的切割分行显示
        self.description = self.description.split("\f")[0]
        # 响应模型response_fields字段生成处理
        response_fields = {}
        for additional_status_code, response in self.responses.items():
            assert isinstance(response, dict), "An additional response must be a dict"
            model = response.get("model")
            if model:
                assert (
                    additional_status_code not in STATUS_CODES_WITH_NO_BODY
                ), f"Status code {additional_status_code} must not have a response body"
                response_name = f"Response_{additional_status_code}_{self.unique_id}"
                response_field = create_response_field(name=response_name, type_=model)
                response_fields[additional_status_code] = response_field
        if response_fields:
            self.response_fields: Dict[Union[int, str], ModelField] = response_fields
        else:
            self.response_fields = {}
          
        # 断言路由节点的类型必须是一个可调用的调用
        assert callable(endpoint), "An endpoint must be a callable"
        # 开始获取依赖注入的dependant
        # dependant是<class 'fastapi.dependencies.models.Dependant'>类型的对象
        self.dependant = get_dependant(path=self.path_format, call=self.endpoint)
        # 循环依赖输入列表项
        for depends in self.dependencies[::-1]:
            # 在dependant里面的dependencies插入具体依赖注入项信息Depends- self.path_format
            #依赖注入depends：Depends(test_dependencie_2) /docs/oauth2-redirect
            #依赖注入depends：Depends(test_dependencie_1) /docs/oauth2-redirect
            self.dependant.dependencies.insert(
                0,
                get_parameterless_sub_dependant(depends=depends, path=self.path_format),
            )
        # 空，因为没设置
        self.body_field = get_body_field(dependant=self.dependant, name=self.unique_id)
        # request_response 路由节点对应的ASGIAPP对象执行get_route_handler
        self.app = request_response(self.get_route_handler())
  
    # 返回的回一个Request列表类型，或者是Coroutine类型的Response
    def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]:
        return get_request_handler(
            dependant=self.dependant,
            body_field=self.body_field,
            status_code=self.status_code,
            response_class=self.response_class,
            response_field=self.secure_cloned_response_field,
            response_model_include=self.response_model_include,
            response_model_exclude=self.response_model_exclude,
            response_model_by_alias=self.response_model_by_alias,
            response_model_exclude_unset=self.response_model_exclude_unset,
            response_model_exclude_defaults=self.response_model_exclude_defaults,
            response_model_exclude_none=self.response_model_exclude_none,
            dependency_overrides_provider=self.dependency_overrides_provider,
        )

    def matches(self, scope: Scope) -> Tuple[Match, Scope]:
        match, child_scope = super().matches(scope)
        if match != Match.NONE:
            child_scope["route"] = self
        return match, child_scope

当一个路由节点执行内部的逻辑的时候，那么最终则是跑到了get_request_handler里面去执行：

def get_request_handler(
    dependant: Dependant,
    body_field: Optional[ModelField] = None,
    status_code: Optional[int] = None,
    response_class: Union[Type[Response], DefaultPlaceholder] = Default(JSONResponse),
    response_field: Optional[ModelField] = None,
    response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
    response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
    response_model_by_alias: bool = True,
    response_model_exclude_unset: bool = False,
    response_model_exclude_defaults: bool = False,
    response_model_exclude_none: bool = False,
    dependency_overrides_provider: Optional[Any] = None,
) -> Callable[[Request], Coroutine[Any, Any, Response]]:
    # 断言依赖注入的dependant.call 是不是已经存在，否则提示"dependant.call must be a function"
    assert dependant.call is not None, "dependant.call must be a function"
    # 判断依赖注入的dependant.call是不是一个协程函数
    is_coroutine = asyncio.iscoroutinefunction(dependant.call)
    # 开始判断要读取body是否是来自form表单提交is_body_form信息
    is_body_form = body_field and isinstance(body_field.field_info, params.Form)
    # 如果响应类处理他是默认DefaultPlaceholder类型的响应类
    if isinstance(response_class, DefaultPlaceholder):
       # ，from starlette.responses import JSONResponse, Response默认Response
        actual_response_class: Type[Response] = response_class.value
    else:
       # 否则是使用自定义的response_class
        actual_response_class = response_class
    
    #开始定义符合app协程对象构建
    async def app(request: Request) -> Response:
        try:
           # 开始解析Request中的body和form信息，所有每次直接的接口的是否会进行获取
            body: Any = None
            # 先判断是否存在body_field
            if body_field:
                # 如果body内容来自表单，那么就读取表单
                if is_body_form:
                    body = await request.form()
                else:
                    # 如果不是直接读取二进制的body
                    body_bytes = await request.body()
                    # 如果存在body_bytes值
                    if body_bytes:
                        # 根据请求头类型设置转化，判断是否需要转JSON格式的类型
                        json_body: Any = Undefined
                        content_type_value = request.headers.get("content-type")
                        # 没有没有设置content_type_value，默认的则获取json值
                        if not content_type_value:
                            json_body = await request.json()
                        else:
                          # 如果有设置值信息,通过email邮件库来设置一个信息类型
                            # 信息类型窗户
                            message = email.message.Message()
                            # 创建信息类型content-type
                            message["content-type"] = content_type_value
                            # 如果信息类型是application，
                            if message.get_content_maintype() == "application":
                               # 二级类型内容类型
                                subtype = message.get_content_subtype()
                                # subtype 内容格式类型
                                if subtype == "json" or subtype.endswith("+json"):
                                    json_body = await request.json()
                        # 如果json_body不为空Undefined
                        if json_body != Undefined:
                           # 那么body就是直接JSON类型
                            body = json_body
                        else:
                           # 否则就是二进制类型的
                            body = body_bytes
        # body类型转换异常
        except json.JSONDecodeError as e:
            # 抛出操作校验码错误的操作，错误的内容是body的信息
            raise RequestValidationError([ErrorWrapper(e, ("body", e.pos))], body=e.doc)
        except Exception as e:
            # 抛出HTTPException解析异常
            raise HTTPException(
                status_code=400, detail="There was an error parsing the body"
            ) from e
        # 再解析body完成后，那么执行依赖项的处理solve_dependencies的意思就是解决依赖注入
        # 返回一个处理结果
        solved_result = await solve_dependencies(
           #依赖项的处理需要传入request对象
            request=request,
            # 需要传入依赖注入dependant
            dependant=dependant,
            # 需要传入对应的body
            body=body,
            # 需要传入依赖注入的提供则对象
            dependency_overrides_provider=dependency_overrides_provider,
        )
        # 当所有的参数解析和依赖项解析都处理完成后，返回结果了
        values, errors, background_tasks, sub_response, _ = solved_result
        # 如果解析过程有错误，就【抛出错误，errors主要是参数解析过程的错误
        if errors:
            raise RequestValidationError(errors, body=body)
        else:
           # 如果没有错误，那有可能是响应体类型的错误或其他类型自定义类型的错误
           # 开始执行run_endpoint_function
            raw_response = await run_endpoint_function(
                dependant=dependant, values=values, is_coroutine=is_coroutine
            )
            ##run_endpoint_function函数内容，判断是协程还是普通函数在执行dependant.call
            ## async def run_endpoint_function(
            ## *, dependant: Dependant, values: Dict[str, Any], is_coroutine: bool
            ##) -> Any:
            ## # Only called by get_request_handler. Has been split into its own function to
              # facilitate profiling endpoints, since inner functions are harder to profile.
             ## assert dependant.call is not None, "dependant.call must be a function"

             ## if is_coroutine:
             ##     return await dependant.call(**values)
             ## else:
              ##    return await run_in_threadpool(dependant.call, **values)
            
            # 如果返回raw_response是Response类型的
            if isinstance(raw_response, Response):
                # 判断路由函数中是否自己存在对应的后台任务列表-是否先存在后台任务background
                if raw_response.background is None:
                    #如果存在，把后台任务赋值给当前raw_response.background
                    raw_response.background = background_tasks
                # 返回响应体
                return raw_response
            # 如果不是Response类型的，那么就自己解析serialize_response
            response_data = await serialize_response(
                field=response_field,
                response_content=raw_response,
                include=response_model_include,
                exclude=response_model_exclude,
                by_alias=response_model_by_alias,
                exclude_unset=response_model_exclude_unset,
                exclude_defaults=response_model_exclude_defaults,
                exclude_none=response_model_exclude_none,
                is_coroutine=is_coroutine,
            )
            # 提取background任务内容
            response_args: Dict[str, Any] = {"background": background_tasks}
            # If status_code was set, use it, otherwise use the default from the
            # response class, in the case of redirect it's 307
            # 判断状态码是否为空
            if status_code is not None:
                response_args["status_code"] = status_code
            # 开始创建actual_response_class自定义actual_response_class类型
            response = actual_response_class(response_data, **response_args)
            # 添加自定义的响应头信息
            response.headers.raw.extend(sub_response.headers.raw)
            # 设置响应码
            if sub_response.status_code:
                response.status_code = sub_response.status_code
            # 返回响应报文
            return response

    return app

依赖注入的解决

源码分析：----这里只是简单分析，由于后续还要增加子依赖项，所以这里暂时不过去的解析！

# 返回是一个元祖Tuple
#-  Dict
#-  List[ErrorWrapper]
#-  Optional[BackgroundTasks]
#-  Dict

async def solve_dependencies(
    *,
    request: Union[Request, WebSocket],
    dependant: Dependant,
    body: Optional[Union[Dict[str, Any], FormData]] = None,
    background_tasks: Optional[BackgroundTasks] = None,
    response: Optional[Response] = None,
    dependency_overrides_provider: Optional[Any] = None,
    dependency_cache: Optional[Dict[Tuple[Callable[..., Any], Tuple[str]], Any]] = None,
) -> Tuple[
    Dict[str, Any],
    List[ErrorWrapper],
    Optional[BackgroundTasks],
    Response,
    Dict[Tuple[Callable[..., Any], Tuple[str]], Any],
]:
    # 值字典类型
    values: Dict[str, Any] = {}
    # 错误类型装饰器列表
    errors: List[ErrorWrapper] = []
    # 如果没有response
    if response is None:
        # 默认创建一个Response
        response = Response()
        # 删除返回长度内容请求头
        del response.headers["content-length"]
        # 设置响应码是Node
        response.status_code = None  # type: ignore
    # 是都开启了依赖注入的缓存
    dependency_cache = dependency_cache or {}
    #是否存在子依赖的Dependant
    sub_dependant: Dependant
    # 判断在子依赖的Dependant信息
    for sub_dependant in dependant.dependencies:
        # 子依赖sub_dependant信息，强制转换为Callable类型
        sub_dependant.call = cast(Callable[..., Any], sub_dependant.call)
        #cast函数是：“”“将值强制转换为类型。
        #这将返回未更改的值。向类型检查器发送此表示返回值具有指定类型，但运行时，我们故意不检查任何内容（我们希望尽可能快）
        # 强制转换为Tuple类型
        sub_dependant.cache_key = cast(
            Tuple[Callable[..., Any], Tuple[str]], sub_dependant.cache_key
        )
        # sub_dependant自已来对象的call方法赋值
        call = sub_dependant.call
        # 使用自已来项de sub_dependant
        use_sub_dependant = sub_dependant
        #如果先存在依赖注入的提供这且这个提供这有自己的依赖项dependency_overrides
        if (
            dependency_overrides_provider
            and dependency_overrides_provider.dependency_overrides
        ):
            original_call = sub_dependant.call
            call = getattr(
                dependency_overrides_provider, "dependency_overrides", {}
            ).get(original_call, original_call)
            use_path: str = sub_dependant.path  # type: ignore
            use_sub_dependant = get_dependant(
                path=use_path,
                call=call,
                name=sub_dependant.name,
                security_scopes=sub_dependant.security_scopes,
            )
            use_sub_dependant.security_scopes = sub_dependant.security_scopes
        # 开始解析执行依赖
        solved_result = await solve_dependencies(
            request=request,
            dependant=use_sub_dependant,
            body=body,
            background_tasks=background_tasks,
            response=response,
            dependency_overrides_provider=dependency_overrides_provider,
            dependency_cache=dependency_cache,
        )
        (
            sub_values,
            sub_errors,
            background_tasks,
            _,  # the subdependency returns the same response we have
            sub_dependency_cache,
        ) = solved_result
        #
        dependency_cache.update(sub_dependency_cache)
        if sub_errors:
            errors.extend(sub_errors)
            continue
        if sub_dependant.use_cache and sub_dependant.cache_key in dependency_cache:
            solved = dependency_cache[sub_dependant.cache_key]
        elif is_gen_callable(call) or is_async_gen_callable(call):
            stack = request.scope.get("fastapi_astack")
            assert isinstance(stack, AsyncExitStack)
            solved = await solve_generator(
                call=call, stack=stack, sub_values=sub_values
            )
        elif is_coroutine_callable(call):
            solved = await call(**sub_values)
        else:
            solved = await run_in_threadpool(call, **sub_values)
        if sub_dependant.name is not None:
            values[sub_dependant.name] = solved
        if sub_dependant.cache_key not in dependency_cache:
            dependency_cache[sub_dependant.cache_key] = solved
    path_values, path_errors = request_params_to_args(
        dependant.path_params, request.path_params
    )
    query_values, query_errors = request_params_to_args(
        dependant.query_params, request.query_params
    )
    header_values, header_errors = request_params_to_args(
        dependant.header_params, request.headers
    )
    cookie_values, cookie_errors = request_params_to_args(
        dependant.cookie_params, request.cookies
    )
    values.update(path_values)
    values.update(query_values)
    values.update(header_values)
    values.update(cookie_values)
    errors += path_errors + query_errors + header_errors + cookie_errors
    if dependant.body_params:
        (
            body_values,
            body_errors,
        ) = await request_body_to_args(  # body_params checked above
            required_params=dependant.body_params, received_body=body
        )
        values.update(body_values)
        errors.extend(body_errors)
    if dependant.http_connection_param_name:
        values[dependant.http_connection_param_name] = request
    if dependant.request_param_name and isinstance(request, Request):
        values[dependant.request_param_name] = request
    elif dependant.websocket_param_name and isinstance(request, WebSocket):
        values[dependant.websocket_param_name] = request
    if dependant.background_tasks_param_name:
        if background_tasks is None:
            background_tasks = BackgroundTasks()
        values[dependant.background_tasks_param_name] = background_tasks
    if dependant.response_param_name:
        values[dependant.response_param_name] = response
    if dependant.security_scopes_param_name:
        values[dependant.security_scopes_param_name] = SecurityScopes(
            scopes=dependant.security_scopes
        )
    return values, errors, background_tasks, response, dependency_cache

上面的依赖注入代码比较繁琐，后续需要结合debug方式来进行梳理，但是从上面简单的分析流程看，目前已经大致了解fastapi一个请求的过程的大致流程了！

以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【程序员小钟同学】

小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822