FastAPI 解读 by Gascognya程序员

FastAPI 源码阅读 (三) Endpoint封装

2020-08-26  本文已影响0人  Gascognya

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自身适应化

上一篇 下一篇

猜你喜欢

热点阅读