FastAPI 依赖注入详解:生成依赖树

class APIRoute(routing.Route):
    def __init__(...):

        ......

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

        ......

在添加APIRoute节点时,会对endpoint进行解析,生成 依赖树get_dependant便是解析出endpoint的依赖树的函数。

这部分在之前源码解析中讲过,但是当时的理解并不深刻。这次让我们来认真剖析这部分

def get_dependant(
    *,
    path: str,
    call: Callable,
    name: Optional[str] = None,
    security_scopes: Optional[List[str]] = None,
    use_cache: bool = True,
) -> Dependant:
    """
    * 该函数为递归函数, 不止会被endpoint调用, 也会被其依赖项调用。

    :param path: 路径
    :param call: endpoint/依赖项
    :param name: 被依赖项使用, 为参数名
    :param security_scopes: 被依赖项使用, 为积攒的安全域
    :param use_cache: 缓存
    :return: Dependant对象
    """
    path_param_names = get_path_param_names(path)
    # 捕捉路径参数 e.g. "/user/{id}"

    endpoint_signature = get_typed_signature(call)
    signature_params = endpoint_signature.parameters
    # 解析endpoint/依赖项的参数, 通过inspect

    if is_gen_callable(call) or is_async_gen_callable(call):
        check_dependency_contextmanagers()
    # 确保异步上下文管理器import成功

    dependant = Dependant(call=call, name=name, path=path, use_cache=use_cache)
    # 依赖对象

    # 生成依赖树
    for param_name, param in signature_params.items():

        if isinstance(param.default, params.Depends):
            # 如果该参数是Depends()时 (因为其写在默认值位置)
            sub_dependant = get_param_sub_dependant(
                param=param, path=path, security_scopes=security_scopes
            )
            # 生成一个子依赖项
            dependant.dependencies.append(sub_dependant)
            # 加入到父依赖项的节点中
            continue

        if add_non_field_param_to_dependency(param=param, dependant=dependant):
            continue
        # 找出Request, WebSocket, HTTPConnection, Response, BackgroundTasks, SecurityScopes等参数。
        # 将其参数名, 在dependant中标注出来

        # 既不是Depends依赖项, 也不是特殊参数
        # 就当做普通参数来看待
        param_field = get_param_field(
            param=param, default_field_info=params.Query, param_name=param_name
        )
        # 参数默认当做Query, 获取其ModelField

        if param_name in path_param_names:
            # 如果这个参数名在上文解析路径得到的路径参数集合中
            # e.g. "/user/{id}" -> {id, ...} -> param_name = "id"

            assert is_scalar_field(
                field=param_field
            ), "Path params must be of one of the supported types"
            # 判断是否为标准的field类型

            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,
            )
            # 重新按Path生成参数字段, 获得ModelField
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的path参数列表

        elif is_scalar_field(field=param_field):
            # 如果并非path参数, 即默认query参数, 但属于标准field类型
            # 注: cookie属于这类
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的query参数列表

        elif isinstance(
            param.default, (params.Query, params.Header)
        ) and is_scalar_sequence_field(param_field):
            # 如果不是path, 也不是标准的query, 但属于包含有Query()或Header()
            # 且为标准序列参数时
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的query或header参数列表

        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(...)"
            # 上述条件都不满足, 即不是路径参数、标准查询参数、Query查询参数、Header参数中任何一个
            # 则断言一定是Body参数
            dependant.body_params.append(param_field)
            # 将其整合到Body参数列表
    return 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)
    # 捕捉路径参数 e.g. "/user/{id}"

    endpoint_signature = get_typed_signature(call)
    signature_params = endpoint_signature.parameters
    # 解析endpoint/依赖项的参数, 通过inspect

    if is_gen_callable(call) or is_async_gen_callable(call):
        check_dependency_contextmanagers()
    # 确保异步上下文管理器import成功

    dependant = Dependant(call=call, name=name, path=path, use_cache=use_cache)
    # 依赖对象

get_dependant不止被endpoint使用,其依赖项和子依赖都会使用,其为递归函数。
开头生成一个Dependant节点对象,等待下面加工,最终被返回。其形成的是一个树状结构

    for param_name, param in signature_params.items():

接下来把该节点的参数都抓出来,逐个分析。

        if isinstance(param.default, params.Depends):
            # 如果该参数是Depends()时 (因为其写在默认值位置)
            sub_dependant = get_param_sub_dependant(
                param=param, path=path, security_scopes=security_scopes
            )
            # 生成一个子依赖项
            dependant.dependencies.append(sub_dependant)
            # 加入到父依赖项的节点中
            continue

首先判断是否为Depends()项,如果是,则生成子依赖。下面是生成子依赖的流程。

def get_param_sub_dependant(
    *, param: inspect.Parameter, path: str, security_scopes: Optional[List[str]] = None
) -> Dependant:
    depends: params.Depends = param.default
    # 拿到Depends对象

    if depends.dependency:
        dependency = depends.dependency
    else:
        dependency = param.annotation
    # 拿到函数/类, 没有则默认为注解类。
    # 这代表着user: User = Depends() 是被允许的

    return get_sub_dependant(
        depends=depends,
        dependency=dependency,
        path=path,
        name=param.name,
        security_scopes=security_scopes,
    )

拿出Depends中的依赖内容,如果没有就用注解来充当。即user: User = Depends()这种形式可以被允许。

def get_sub_dependant(
    *,
    depends: params.Depends,
    dependency: Callable,
    path: str,
    name: Optional[str] = None,
    security_scopes: Optional[List[str]] = None,
) -> Dependant:
    """
    :param depends: 依赖项对象
    :param dependency: 具体依赖内容
    :param path: 路径
    :param name: 参数名
    :param security_scopes: 安全域
    :return:
    """
    security_requirement = None
    # 安全性要求, 先置为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):
        # 如果依赖内容是安全认证 e.g. Depends(oauth2_scheme)
        # 注: OAuth2是SecurityBase的子类

        use_scopes: List[str] = []
        if isinstance(dependency, (OAuth2, OpenIdConnect)):
            # 注: OAuth2PasswordBearer, OAuth2AuthorizationCodeBearer
            # 两者为OAuth2子类
            use_scopes = security_scopes
            # 如果其为上述两者实例, 则将积攒的安全域, 传入其中。

        security_requirement = SecurityRequirement(
            security_scheme=dependency, scopes=use_scopes
        )
        # 安全性需求置为, SecurityRequirement(SecurityBase, [])
        # 或者 SecurityRequirement(OAuth2, security_scopes)

    # 上文两个判断组合起来的逻辑是
    # 1. 第一个判断, 将后置依赖中的安全域需求整合起来
    # 2. 当扫描到了前置的OAuth2时, 将这些积攒的安全域需求传入其中

    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)
    # 将SecurityRequirement放进这个依赖项中
    # 注意SecurityRequirement存在条件是本依赖项为SecurityBase相关

    sub_dependant.security_scopes = security_scopes
    # 将现有的安全域需求放进这个依赖项中

    return sub_dependant

接下来是对安全相关的处理。我们可以看到,中间又调用了get_dependant,参数包含了namesecurity_scopes。endpoint的根节点传参不包含这两项。

回到get_dependant
        if add_non_field_param_to_dependency(param=param, dependant=dependant):
            continue
        # 找出Request, WebSocket, HTTPConnection, Response, BackgroundTasks, SecurityScopes等参数。
        # 将其参数名, 在dependant中标注出来

        # 既不是Depends依赖项, 也不是特殊参数
        # 就当做普通参数来看待
        param_field = get_param_field(
            param=param, default_field_info=params.Query, param_name=param_name
        )
        # 参数默认当做Query, 获取其ModelField

如果不是Depends参数,则首先默认当成查询参数query,并生成ModelField字段。

        if param_name in path_param_names:
            # 如果这个参数名在上文解析路径得到的路径参数集合中
            # e.g. "/user/{id}" -> {id, ...} -> param_name = "id"

            assert is_scalar_field(
                field=param_field
            ), "Path params must be of one of the supported types"
            # 判断是否为标准的field类型

            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,
            )
            # 重新按Path生成参数字段, 获得ModelField
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的path参数列表

如果其为路径参数,则重新生成ModelField字段。再整合到dependant的参数列表中

        elif is_scalar_field(field=param_field):
            # 如果并非path参数, 即默认query参数, 但属于标准field类型
            # 注: cookie属于这类
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的query参数列表

不是路径参数,但是标准的查询参数

        elif isinstance(
            param.default, (params.Query, params.Header)
        ) and is_scalar_sequence_field(param_field):
            # 如果不是path, 也不是标准的query, 但属于包含有Query()或Header()
            # 且为标准序列参数时
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的query或header参数列表

Query()和Header()两种情况

        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(...)"
            # 上述条件都不满足, 即不是路径参数、标准查询参数、Query查询参数、Header参数中任何一个
            # 则断言一定是Body参数
            dependant.body_params.append(param_field)
            # 将其整合到Body参数列表
    return dependant

当上述条件都不满足,则可以断言为Body()字段。

就此,一个APIRoute的依赖树便生成了
下章说说如何使用依赖树

©著作权归作者所有,转载或内容合作请联系作者
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 202,009评论 5 474
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 84,808评论 2 378
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 148,891评论 0 335
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,283评论 1 272
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,285评论 5 363
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,409评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,809评论 3 393
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,487评论 0 256
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,680评论 1 295
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,499评论 2 318
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,548评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,268评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,815评论 3 304
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,872评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,102评论 1 258
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,683评论 2 348
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,253评论 2 341