美文网首页FastAPI 解读 by Gascognya程序员
FastAPI 源码阅读 (三) Endpoint封装

FastAPI 源码阅读 (三) Endpoint封装

作者: Gascognya | 来源:发表于2020-08-26 22:25 被阅读0次

    Starlette中,请求的流动是基于Scope来实现的,到endpoint的前一步,将Scope封装成Request,在FastAPI中,Route节点Endpoint的过程中,加入了大量逻辑,其中包括依赖判断,安全认证,数据类型判断等。那么这些具体是如何实现的呢?
    FastAPIAPIRoute实例化时,会对endpoint的参数进行解析。

    这涉及到inspect库,他可以解析函数的参数,包括其注释的Typing,以及其默认值,这为 id: str = Depends(get_id) 这样的表现形式提供了先决条件。

    这个过程分为两个阶段四个步骤:

    1. 配置response model
    2. 检查参数,搜集类型和依赖。构建dependant。
    3. 获取到request报文,参照denpendant将参数注入
    4. 将返回值注入到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的子类。
                # 但是因为它是一个子类,所以它会通过验证并原样返回。
                # 作为一个新字段,没有继承会原样传递。总是会创建一个新的模型。
    
                # 这段的意思是指,model的子类,也会通过model的认证,这是不可以的。
                # 所以将model的字段都拷贝到一个新类里,这样它就是完完全全的一个新类,
                # 不会受继承的影响了。
                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
            # response_model是指BaseModel的子类
            # response_field是由response_model生成的ModelField验证类
            # secure_cloned_response_field是由response_field克隆而来的
    
            self.responses = responses or {}
            response_fields = {}
            # responses的作用是,在不同code中使用不同的model。所以需要预置不同的验证
            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 = {}
    

    这里提到了一个类型: ModelField,他是根据model生成的一个专属验证对象
    中间这段是对responses的内容进行拆解,关于responses我们之前提到过。
    OpenAPI中的其他响应 - FastAPI

    关于依赖的配置

            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),
                )
    

    这里使用了get_dependant函数,将endpoint传入进去。用来解析出endpoint的依赖项。随后将dependencies手动传入的依赖并入其中

    get_dependant

    def get_dependant(
        *,
        path: str,
        call: Callable,
        name: Optional[str] = None,
        security_scopes: Optional[List[str]] = None,
        use_cache: bool = True,
    ) -> Dependant:
        # 从函数参数字典中解析出字段。
        path_param_names = get_path_param_names(path)
        endpoint_signature = get_typed_signature(call)
        signature_params = endpoint_signature.parameters
        # 例 OrderedDict([('item_id', <Parameter "item_id: str = Depends(abc)">)])
        if is_gen_callable(call) or is_async_gen_callable(call):
            check_dependency_contextmanagers()
        # 把参数中的Depends()抓出来
        dependant = Dependant(call=call, name=name, path=path, use_cache=use_cache)
        # 储存用,没有方法
    
        # 配置各种参数,例如 =Depends(),=Path(),=Query(),或者是普通的参数。将其加入到dependant
        for param_name, param in signature_params.items():
            # 依赖项
            if isinstance(param.default, params.Depends):
                # get_dependant → for get_param_sub_dependant → get_sub_dependant → get_dependant
                # 1,按节点寻找依赖项,将子节点的dependant加入到自身,向上返回dependant对象
                # 2,获取参数中依赖项的具体依赖函数
                # 3,将依赖函数进行判断,例如是否为安全相关,将其作为节点传入到get_dependant中
    
                # 这是个循环遍历,最终建立dependant树
                # 根节点为endpoint的dependant对象
                # 将他的依赖中的函数作为节点,再传入到这个函数中
                # 然后将子依赖的dependant加入到父dependant的dependencies序列中
                sub_dependant = get_param_sub_dependant(
                    param=param, path=path, security_scopes=security_scopes
                )
                dependant.dependencies.append(sub_dependant)
                continue
    
            # 判断是否为Request,WebSocket等参数。
            if add_non_field_param_to_dependency(param=param, dependant=dependant):
                continue
            # 查询参数验证
            param_field = get_param_field(
                param=param, default_field_info=params.Query, param_name=param_name
            )
            # 获取参数的ModelField
            # 路径参数 例如"/student/{id}"
            if param_name in path_param_names:
                assert is_scalar_field(
                    field=param_field
                ), "Path params must be of one of the supported types"
                #
                if isinstance(param.default, params.Path):
                    ignore_default = False
                else:
                    ignore_default = True
                # path_param = Path(), 对路径参数进行认证的
                # 设定为不忽视默认值(忽视掉没法用了)
    
                # 改为路径参数认证
                param_field = get_param_field(
                    param=param,
                    param_name=param_name,
                    default_field_info=params.Path,
                    force_type=params.ParamTypes.path,
                    ignore_default=ignore_default,
                )
                add_param_to_fields(field=param_field, dependant=dependant)
            elif is_scalar_field(field=param_field):
                add_param_to_fields(field=param_field, dependant=dependant)
            elif isinstance(
                param.default, (params.Query, params.Header)
            ) and is_scalar_sequence_field(param_field):
                add_param_to_fields(field=param_field, dependant=dependant)
                # 当为路径参数,标准参数,标准序列参数时,加入到验证
            else:
                field_info = param_field.field_info
                assert isinstance(
                    field_info, params.Body
                ), f"Param: {param_field.name} can only be a request body, using Body(...)"
                dependant.body_params.append(param_field)
                # 否则只能是body
        return dependant
    
    get_param_sub_dependant
    def get_param_sub_dependant(
        *, param: inspect.Parameter, path: str, security_scopes: Optional[List[str]] = None
    ) -> Dependant:
        # 将参数中的Depends中的具体依赖函数dependency剥离出来。
        depends: params.Depends = param.default
        # default = Depends(fun_a)对象
    
        if depends.dependency:
            dependency = depends.dependency
        else:
            dependency = param.annotation
        return get_sub_dependant(
            depends=depends,
            dependency=dependency,
            path=path,
            name=param.name,
            security_scopes=security_scopes,
        )
    

    get_sub_dependant

    def get_sub_dependant(
        *,
        depends: params.Depends,
        dependency: Callable,
        path: str,
        name: Optional[str] = None,
        security_scopes: Optional[List[str]] = None,
    ) -> Dependant:
        security_requirement = None
        # 安全性要求
        security_scopes = security_scopes or []
        # 安全范围
    
        # 安全相关
        if isinstance(depends, params.Security):
            # Security是Depends的子类
            dependency_scopes = depends.scopes
            security_scopes.extend(dependency_scopes)
        if isinstance(dependency, SecurityBase):
            # OAuth2PasswordBearer就是SecurityBase的子类
            # 例: async def read_items(token: str = Depends(oauth2_scheme)):
            use_scopes: List[str] = []
            if isinstance(dependency, (OAuth2, OpenIdConnect)):
                use_scopes = security_scopes
            security_requirement = SecurityRequirement(
                security_scheme=dependency, scopes=use_scopes
            )
        # 将从依赖项中剥离出来的函数,再作为节点,传入到其中
        # 最终形成依赖树
        sub_dependant = get_dependant(
            path=path,
            call=dependency,
            name=name,
            security_scopes=security_scopes,
            use_cache=depends.use_cache,
        )
        if security_requirement:
            sub_dependant.security_requirements.append(security_requirement)
        sub_dependant.security_scopes = security_scopes
        return sub_dependant
    

    get_dependant → for get_param_sub_dependant → get_sub_dependant → get_dependant

    1. 按节点寻找依赖项,将子节点的dependant加入到自身,向上返回dependant对象
    2. 获取参数中依赖项的具体依赖函数
    3. 将依赖函数进行判断,例如是否为安全相关,将其作为节点传入到get_dependant中

    以上是对APIRoute封装的大致过程

    在这阶段中,APIRoute对endpoint进行解析,从中获取关于参数和model的信息。然后进行配置,将APIRoute自身适应化

    相关文章

      网友评论

        本文标题:FastAPI 源码阅读 (三) Endpoint封装

        本文链接:https://www.haomeiwen.com/subject/gokxsktx.html