APIRouter与APIRoute

路由器与路由节点继承于Starlette的路由
在其基础上添加了一些功能，例如装饰器路由。还有对于endpoint前后封装的扩展(依赖，类型检测，返回模型)

APIRouter

class APIRouter(routing.Router):
    def __init__(
            self,
            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,
            default_response_class: Optional[Type[Response]] = None,
            on_startup: Optional[Sequence[Callable]] = None,
            on_shutdown: Optional[Sequence[Callable]] = None,
    ) -> None:
        super().__init__(
            routes=routes,
            redirect_slashes=redirect_slashes,
            default=default,
            on_startup=on_startup,
            on_shutdown=on_shutdown,
        )
        self.dependency_overrides_provider = dependency_overrides_provider
        self.route_class = route_class
        self.default_response_class = default_response_class

路由装饰器的核心方法

    def add_api_route(
            self,
            path: str,
            endpoint: Callable,
            *,
            response_model: Optional[Type[Any]] = None,
            status_code: int = 200,
            tags: Optional[List[str]] = 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: Optional[Type[Response]] = None,
            name: Optional[str] = None,
            route_class_override: Optional[Type[APIRoute]] = None,
            callbacks: Optional[List[APIRoute]] = None,
    ) -> None:
        """
        应用与路由的方法关系(以下用A代表app，R代表router)
            A.add_api_route → R.add_api_route(此处)
            A.include_router → R.include_router

            以下为传参装饰器，在进入add_xxx_route之前，都需要进行一次闭包

            A.api_route(闭包) → R.add_api_route(此处)

            A.get/put/post/delete/options/head/patch/trace
            ↓
            R.get/put/post/delete/options/head/patch/trace
            ↓
            R.api_route(闭包) → R.add_api_route(此处)

            A.websocket(闭包) → A.add_api_websocket_route → R.add_api_websocket_route

            R.add_api_websocket_route(闭包) → R.add_api_websocket_route


        route_class_override: 使用的Route类，默认是APIRoute

        """
        route_class = route_class_override or self.route_class
        # 新建一个Route节点
        route = route_class(
            path,
            endpoint=endpoint,
            response_model=response_model,
            status_code=status_code,
            tags=tags or [],
            dependencies=dependencies,
            summary=summary,
            description=description,
            response_description=response_description,
            responses=responses or {},
            deprecated=deprecated,
            methods=methods,
            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 or self.default_response_class,
            name=name,
            dependency_overrides_provider=self.dependency_overrides_provider,
            callbacks=callbacks,
        )
        self.routes.append(route)
        # 添加到路由当中

    def api_route(...) -> Callable:
        def decorator(func: Callable) -> Callable:
            self.add_api_route(...)
            return func
        return decorator

    def get(...):
        return self.api_route(...)

    def put(...):
        return self.api_route(...)

    def post(...):
        return self.api_route(...)

    def delete(...):
        return self.api_route(...)

    def options(...):
        return self.api_route(...)

    def head(...):
        return self.api_route(...)

    def patch(...):
        return self.api_route()

    def trace(...):
        return self.api_route()

子路由添加

include_router()是fastapi提供的一种向路由中添加子路由的方式。其不同于Mount直接挂载App，而是将子router中的路由都拆解出来添加到根router

    def include_router(
            self,
            router: "APIRouter",
            *,
            prefix: str = "",
            tags: Optional[List[str]] = None,
            dependencies: Optional[Sequence[params.Depends]] = None,
            responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
            default_response_class: Optional[Type[Response]] = None,
    ) -> None:
        """
        将子路由的路由节点都抓出来，加入到自身的router中。

        这个prefix为子路由根路径，设置这个即可省去前缀。
        e.g. /user/login; /user/register → prefix = /user; /login; /register
        """
        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 = {}
        for route in router.routes:
            if isinstance(route, APIRoute):
                combined_responses = {**responses, **route.responses}
                self.add_api_route(
                    prefix + route.path,
                    route.endpoint,
                    response_model=route.response_model,
                    status_code=route.status_code,
                    tags=(route.tags or []) + (tags or []),
                    dependencies=list(dependencies or [])
                                 + list(route.dependencies or []),
                    summary=route.summary,
                    description=route.description,
                    response_description=route.response_description,
                    responses=combined_responses,
                    deprecated=route.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,
                    response_class=route.response_class or default_response_class,
                    name=route.name,
                    route_class_override=type(route),
                    callbacks=route.callbacks,
                )
            elif isinstance(route, routing.Route):
                self.add_route(
                    prefix + route.path,
                    route.endpoint,
                    methods=list(route.methods or []),
                    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)

APIRoute

路由节点，承载了fastapi很大一部分功能。也是fastapi创新的集中体现。代码比较复杂。

class APIRoute(routing.Route):
    def __init__(
            self,
            path: str,
            endpoint: Callable,
            *,
            response_model: Optional[Type[Any]] = None,
            status_code: int = 200,
            tags: Optional[List[str]] = 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: Optional[Type[Response]] = None,
            dependency_overrides_provider: Optional[Any] = None,
            callbacks: Optional[List["APIRoute"]] = None,
    ) -> None:
        """

        Starlette原生
        :param path: 路径
        :param endpoint: 就是你写的函数
        :param methods: 支持的方法
        :param name: 用于反向查找
        :param include_in_schema: 是否被API文档收录

        FastAPI新增

        :param status_code: 状态码
        :param dependencies: 依赖
        :param deprecated: 弃用
        :param callbacks: 回调

        :param response_class: 所使用的response类，默认为JSONResponse
        :param response_model: 返回JSON的模型类
        :param response_model_include: JSON中只包含哪些字段，格式为集合{字段名，字段名，...}
        :param response_model_exclude: JSON中排除哪些字段，同上，都不限于集合，其他序列会自动转换
        :param response_model_by_alias:
        :param response_model_exclude_unset: 返回JSON时，排除掉response模型中为设置的项
        :param response_model_exclude_defaults: 排除掉默认值(与默认值相同并不会)
        :param response_model_exclude_none: 排除掉为None的项
        :param responses: 用于指定，在不同状态码中返回的不同模型.

        :param tags: API文档中，接口所属的类别
        :param summary: API文档中的接口名字
        :param description: API文档中的接口描述
        :param response_description: API文档中接口返回的Response的描述
        :param operation_id: 修改在API文档的前端路由中的路径名
        """
        # 标准枚举 例如 http.HTTPStatus
        if isinstance(status_code, enum.IntEnum):
            status_code = int(status_code)
        self.path = path
        self.endpoint = endpoint
        self.name = get_name(endpoint) if name is None else name
        # 通过endpoint的函数名(__name__)获取

        self.path_regex, self.path_format, self.param_convertors = compile_path(path)
        # 分解路径字符串

        if methods is None:
            methods = ["GET"]
        # 默认添加GET方法

        self.methods = set([method.upper() for method in methods])

        self.unique_id = generate_operation_id_for_path(
            name=self.name, path=self.path_format, method=list(methods)[0]
        )
        # 例如 'read_item_items__item_id__get'
        self.response_model = response_model
        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"
            # 一些状态码不允许有body

            # 下面和pydantic有关
            response_name = "Response_" + self.unique_id
            self.response_field = create_response_field(
                name=response_name, type_=self.response_model
            )

            # 创建字段的克隆，这样Pydantic子模型就不会返回，因为它是一个更有限的类的子类的实例。
            # UserInDB(包含hashed_password)可能是一个没有hashed_password的User的子类。
            # 但是因为它是一个子类，所以它会通过验证并原样返回。
            # 作为一个新字段，没有继承会原样传递。总是会创建一个新的模型。
            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

        self.status_code = status_code
        self.tags = tags or []
        if dependencies:
            self.dependencies = list(dependencies)
        else:
            self.dependencies = []
        self.summary = summary
        self.description = description or inspect.cleandoc(self.endpoint.__doc__ or "")
        # 如果endpoint的doc中有，就按doc中的
        # 如果在描述文本中发现一个“换行符”字符(换行符)，则将描述文本截断到第一个“换行符”之前的内容。
        self.description = self.description.split("\f")[0]
        self.response_description = response_description
        self.responses = responses or {}
        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")
            # 将其中的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"
                # 一些状态码不允许有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
                # 创建{code: ModelField}字典
        if response_fields:
            # 不为空则挂载到实例
            self.response_fields: Dict[Union[int, str], ModelField] = response_fields
        else:
            self.response_fields = {}
        self.deprecated = deprecated
        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

        assert callable(endpoint), "An endpoint must be a callable"
        # 下面开始把参数中的依赖性Depends()抓出来，endpoint也被保存在这里
        self.dependant = get_dependant(path=self.path_format, call=self.endpoint)
        for depends in self.dependencies[::-1]:
            self.dependant.dependencies.insert(
                0,
                get_parameterless_sub_dependant(depends=depends, path=self.path_format),
            )
        # 将装饰器的依赖列表和参数中的依赖项整合起来。

        # body验证
        self.body_field = get_body_field(dependant=self.dependant, name=self.unique_id)
        self.dependency_overrides_provider = dependency_overrides_provider
        self.callbacks = callbacks
        self.app = request_response(self.get_route_handler())

    def get_route_handler(self) -> Callable:
        return get_request_handler(
            dependant=self.dependant,
            body_field=self.body_field,
            status_code=self.status_code,
            response_class=self.response_class or JSONResponse,
            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,
        )

这段定义中，用到了许多工具函数。
response模型验证
create_response_field()
create_cloned_field()
依赖项
get_dependant()
get_parameterless_sub_dependant()
body验证
get_body_field()
封装endpoint
request_response()
get_request_handler()

下一章我们将对这些函数进行解读