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節(jié)點(diǎn)時猴娩,會對endpoint進(jìn)行解析,生成 依賴樹勺阐,get_dependant便是解析出endpoint的依賴樹的函數(shù)卷中。

這部分在之前源碼解析中講過,但是當(dāng)時的理解并不深刻渊抽。這次讓我們來認(rèn)真剖析這部分

def get_dependant(
    *,
    path: str,
    call: Callable,
    name: Optional[str] = None,
    security_scopes: Optional[List[str]] = None,
    use_cache: bool = True,
) -> Dependant:
    """
    * 該函數(shù)為遞歸函數(shù), 不止會被endpoint調(diào)用, 也會被其依賴項(xiàng)調(diào)用蟆豫。

    :param path: 路徑
    :param call: endpoint/依賴項(xiàng)
    :param name: 被依賴項(xiàng)使用, 為參數(shù)名
    :param security_scopes: 被依賴項(xiàng)使用, 為積攢的安全域
    :param use_cache: 緩存
    :return: Dependant對象
    """
    path_param_names = get_path_param_names(path)
    # 捕捉路徑參數(shù) e.g. "/user/{id}"

    endpoint_signature = get_typed_signature(call)
    signature_params = endpoint_signature.parameters
    # 解析endpoint/依賴項(xiàng)的參數(shù), 通過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):
            # 如果該參數(shù)是Depends()時 (因?yàn)槠鋵懺谀J(rèn)值位置)
            sub_dependant = get_param_sub_dependant(
                param=param, path=path, security_scopes=security_scopes
            )
            # 生成一個子依賴項(xiàng)
            dependant.dependencies.append(sub_dependant)
            # 加入到父依賴項(xiàng)的節(jié)點(diǎn)中
            continue

        if add_non_field_param_to_dependency(param=param, dependant=dependant):
            continue
        # 找出Request, WebSocket, HTTPConnection, Response, BackgroundTasks, SecurityScopes等參數(shù)。
        # 將其參數(shù)名, 在dependant中標(biāo)注出來

        # 既不是Depends依賴項(xiàng), 也不是特殊參數(shù)
        # 就當(dāng)做普通參數(shù)來看待
        param_field = get_param_field(
            param=param, default_field_info=params.Query, param_name=param_name
        )
        # 參數(shù)默認(rèn)當(dāng)做Query, 獲取其ModelField

        if param_name in path_param_names:
            # 如果這個參數(shù)名在上文解析路徑得到的路徑參數(shù)集合中
            # 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"
            # 判斷是否為標(biāo)準(zhǔn)的field類型

            if isinstance(param.default, params.Path):
                ignore_default = False
            else:
                ignore_default = True
            # path_param = Path(), 設(shè)置為不忽略默認(rèn), 確保有效

            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生成參數(shù)字段, 獲得ModelField
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的path參數(shù)列表

        elif is_scalar_field(field=param_field):
            # 如果并非path參數(shù), 即默認(rèn)query參數(shù), 但屬于標(biāo)準(zhǔn)field類型
            # 注: cookie屬于這類
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的query參數(shù)列表

        elif isinstance(
            param.default, (params.Query, params.Header)
        ) and is_scalar_sequence_field(param_field):
            # 如果不是path, 也不是標(biāo)準(zhǔn)的query, 但屬于包含有Query()或Header()
            # 且為標(biāo)準(zhǔn)序列參數(shù)時
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的query或header參數(shù)列表

        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(...)"
            # 上述條件都不滿足, 即不是路徑參數(shù)懒闷、標(biāo)準(zhǔn)查詢參數(shù)十减、Query查詢參數(shù)栈幸、Header參數(shù)中任何一個
            # 則斷言一定是Body參數(shù)
            dependant.body_params.append(param_field)
            # 將其整合到Body參數(shù)列表
    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)
    # 捕捉路徑參數(shù) e.g. "/user/{id}"

    endpoint_signature = get_typed_signature(call)
    signature_params = endpoint_signature.parameters
    # 解析endpoint/依賴項(xiàng)的參數(shù), 通過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使用,其依賴項(xiàng)和子依賴都會使用帮辟,其為遞歸函數(shù)速址。
開頭生成一個Dependant節(jié)點(diǎn)對象,等待下面加工由驹,最終被返回芍锚。其形成的是一個樹狀結(jié)構(gòu)

    for param_name, param in signature_params.items():

接下來把該節(jié)點(diǎn)的參數(shù)都抓出來蔓榄,逐個分析并炮。

        if isinstance(param.default, params.Depends):
            # 如果該參數(shù)是Depends()時 (因?yàn)槠鋵懺谀J(rèn)值位置)
            sub_dependant = get_param_sub_dependant(
                param=param, path=path, security_scopes=security_scopes
            )
            # 生成一個子依賴項(xiàng)
            dependant.dependencies.append(sub_dependant)
            # 加入到父依賴項(xiàng)的節(jié)點(diǎn)中
            continue

首先判斷是否為Depends()項(xiàng),如果是润樱,則生成子依賴。下面是生成子依賴的流程羡棵。

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
    # 拿到函數(shù)/類, 沒有則默認(rèn)為注解類壹若。
    # 這代表著user: User = Depends() 是被允許的

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

拿出Depends中的依賴內(nèi)容,如果沒有就用注解來充當(dāng)皂冰。即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: 依賴項(xiàng)對象
    :param dependency: 具體依賴內(nèi)容
    :param path: 路徑
    :param name: 參數(shù)名
    :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)
        # 將依賴項(xiàng)的安全域整合進(jìn)來

    if isinstance(dependency, SecurityBase):
        # 如果依賴內(nèi)容是安全認(rèn)證 e.g. Depends(oauth2_scheme)
        # 注: OAuth2是SecurityBase的子類

        use_scopes: List[str] = []
        if isinstance(dependency, (OAuth2, OpenIdConnect)):
            # 注: OAuth2PasswordBearer, OAuth2AuthorizationCodeBearer
            # 兩者為OAuth2子類
            use_scopes = security_scopes
            # 如果其為上述兩者實(shí)例, 則將積攢的安全域, 傳入其中。

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

    # 上文兩個判斷組合起來的邏輯是
    # 1. 第一個判斷, 將后置依賴中的安全域需求整合起來
    # 2. 當(dāng)掃描到了前置的OAuth2時, 將這些積攢的安全域需求傳入其中

    sub_dependant = get_dependant(
        path=path,
        call=dependency,
        name=name,
        security_scopes=security_scopes,
        use_cache=depends.use_cache,
    )
    # 以這個依賴項(xiàng)作為根節(jié)點(diǎn), 繼續(xù)生產(chǎn)依賴樹

    if security_requirement:
        sub_dependant.security_requirements.append(security_requirement)
    # 將SecurityRequirement放進(jìn)這個依賴項(xiàng)中
    # 注意SecurityRequirement存在條件是本依賴項(xiàng)為SecurityBase相關(guān)

    sub_dependant.security_scopes = security_scopes
    # 將現(xiàn)有的安全域需求放進(jìn)這個依賴項(xiàng)中

    return sub_dependant

接下來是對安全相關(guān)的處理秃流。我們可以看到赂蕴,中間又調(diào)用了get_dependant,參數(shù)包含了namesecurity_scopes舶胀。endpoint的根節(jié)點(diǎn)傳參不包含這兩項(xiàng)概说。

回到get_dependant
        if add_non_field_param_to_dependency(param=param, dependant=dependant):
            continue
        # 找出Request, WebSocket, HTTPConnection, Response, BackgroundTasks, SecurityScopes等參數(shù)。
        # 將其參數(shù)名, 在dependant中標(biāo)注出來

        # 既不是Depends依賴項(xiàng), 也不是特殊參數(shù)
        # 就當(dāng)做普通參數(shù)來看待
        param_field = get_param_field(
            param=param, default_field_info=params.Query, param_name=param_name
        )
        # 參數(shù)默認(rèn)當(dāng)做Query, 獲取其ModelField

如果不是Depends參數(shù)嚣伐,則首先默認(rèn)當(dāng)成查詢參數(shù)query糖赔,并生成ModelField字段。

        if param_name in path_param_names:
            # 如果這個參數(shù)名在上文解析路徑得到的路徑參數(shù)集合中
            # 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"
            # 判斷是否為標(biāo)準(zhǔn)的field類型

            if isinstance(param.default, params.Path):
                ignore_default = False
            else:
                ignore_default = True
            # path_param = Path(), 設(shè)置為不忽略默認(rèn), 確保有效

            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生成參數(shù)字段, 獲得ModelField
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的path參數(shù)列表

如果其為路徑參數(shù)轩端,則重新生成ModelField字段放典。再整合到dependant的參數(shù)列表中

        elif is_scalar_field(field=param_field):
            # 如果并非path參數(shù), 即默認(rèn)query參數(shù), 但屬于標(biāo)準(zhǔn)field類型
            # 注: cookie屬于這類
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的query參數(shù)列表

不是路徑參數(shù),但是標(biāo)準(zhǔn)的查詢參數(shù)

        elif isinstance(
            param.default, (params.Query, params.Header)
        ) and is_scalar_sequence_field(param_field):
            # 如果不是path, 也不是標(biāo)準(zhǔn)的query, 但屬于包含有Query()或Header()
            # 且為標(biāo)準(zhǔn)序列參數(shù)時
            add_param_to_fields(field=param_field, dependant=dependant)
            # 整合到dependant的query或header參數(shù)列表

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(...)"
            # 上述條件都不滿足, 即不是路徑參數(shù)基茵、標(biāo)準(zhǔn)查詢參數(shù)奋构、Query查詢參數(shù)、Header參數(shù)中任何一個
            # 則斷言一定是Body參數(shù)
            dependant.body_params.append(param_field)
            # 將其整合到Body參數(shù)列表
    return dependant

當(dāng)上述條件都不滿足拱层,則可以斷言為Body()字段弥臼。

就此,一個APIRoute的依賴樹便生成了
下章說說如何使用依賴樹

?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
  • 序言:七十年代末根灯,一起剝皮案震驚了整個濱河市醋火,隨后出現(xiàn)的幾起案子悠汽,更是在濱河造成了極大的恐慌,老刑警劉巖芥驳,帶你破解...
    沈念sama閱讀 222,807評論 6 518
  • 序言:濱河連續(xù)發(fā)生了三起死亡事件柿冲,死亡現(xiàn)場離奇詭異,居然都是意外死亡兆旬,警方通過查閱死者的電腦和手機(jī)假抄,發(fā)現(xiàn)死者居然都...
    沈念sama閱讀 95,284評論 3 399
  • 文/潘曉璐 我一進(jìn)店門,熙熙樓的掌柜王于貴愁眉苦臉地迎上來丽猬,“玉大人宿饱,你說我怎么就攤上這事〗潘睿” “怎么了谬以?”我有些...
    開封第一講書人閱讀 169,589評論 0 363
  • 文/不壞的土叔 我叫張陵,是天一觀的道長由桌。 經(jīng)常有香客問我为黎,道長,這世上最難降的妖魔是什么行您? 我笑而不...
    開封第一講書人閱讀 60,188評論 1 300
  • 正文 為了忘掉前任铭乾,我火速辦了婚禮,結(jié)果婚禮上娃循,老公的妹妹穿的比我還像新娘炕檩。我一直安慰自己,他們只是感情好捌斧,可當(dāng)我...
    茶點(diǎn)故事閱讀 69,185評論 6 398
  • 文/花漫 我一把揭開白布笛质。 她就那樣靜靜地躺著,像睡著了一般捞蚂。 火紅的嫁衣襯著肌膚如雪经瓷。 梳的紋絲不亂的頭發(fā)上,一...
    開封第一講書人閱讀 52,785評論 1 314
  • 那天洞难,我揣著相機(jī)與錄音舆吮,去河邊找鬼。 笑死队贱,一個胖子當(dāng)著我的面吹牛色冀,可吹牛的內(nèi)容都是我干的。 我是一名探鬼主播柱嫌,決...
    沈念sama閱讀 41,220評論 3 423
  • 文/蒼蘭香墨 我猛地睜開眼锋恬,長吁一口氣:“原來是場噩夢啊……” “哼!你這毒婦竟也來了编丘?” 一聲冷哼從身側(cè)響起与学,我...
    開封第一講書人閱讀 40,167評論 0 277
  • 序言:老撾萬榮一對情侶失蹤彤悔,失蹤者是張志新(化名)和其女友劉穎,沒想到半個月后索守,有當(dāng)?shù)厝嗽跇淞掷锇l(fā)現(xiàn)了一具尸體晕窑,經(jīng)...
    沈念sama閱讀 46,698評論 1 320
  • 正文 獨(dú)居荒郊野嶺守林人離奇死亡,尸身上長有42處帶血的膿包…… 初始之章·張勛 以下內(nèi)容為張勛視角 年9月15日...
    茶點(diǎn)故事閱讀 38,767評論 3 343
  • 正文 我和宋清朗相戀三年卵佛,在試婚紗的時候發(fā)現(xiàn)自己被綠了杨赤。 大學(xué)時的朋友給我發(fā)了我未婚夫和他白月光在一起吃飯的照片。...
    茶點(diǎn)故事閱讀 40,912評論 1 353
  • 序言:一個原本活蹦亂跳的男人離奇死亡截汪,死狀恐怖疾牲,靈堂內(nèi)的尸體忽然破棺而出蹬癌,到底是詐尸還是另有隱情桐款,我是刑警寧澤,帶...
    沈念sama閱讀 36,572評論 5 351
  • 正文 年R本政府宣布舀患,位于F島的核電站蚓峦,受9級特大地震影響舌剂,放射性物質(zhì)發(fā)生泄漏。R本人自食惡果不足惜枫匾,卻給世界環(huán)境...
    茶點(diǎn)故事閱讀 42,254評論 3 336
  • 文/蒙蒙 一架诞、第九天 我趴在偏房一處隱蔽的房頂上張望拟淮。 院中可真熱鬧干茉,春花似錦、人聲如沸很泊。這莊子的主人今日做“春日...
    開封第一講書人閱讀 32,746評論 0 25
  • 文/蒼蘭香墨 我抬頭看了看天上的太陽委造。三九已至戳鹅,卻和暖如春,著一層夾襖步出監(jiān)牢的瞬間昏兆,已是汗流浹背枫虏。 一陣腳步聲響...
    開封第一講書人閱讀 33,859評論 1 274
  • 我被黑心中介騙來泰國打工, 沒想到剛下飛機(jī)就差點(diǎn)兒被人妖公主榨干…… 1. 我叫王不留爬虱,地道東北人隶债。 一個月前我還...
    沈念sama閱讀 49,359評論 3 379
  • 正文 我出身青樓,卻偏偏與公主長得像跑筝,于是被迫代替她去往敵國和親死讹。 傳聞我的和親對象是個殘疾皇子,可洞房花燭夜當(dāng)晚...
    茶點(diǎn)故事閱讀 45,922評論 2 361