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ù)包含了name
和security_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的依賴樹便生成了
下章說說如何使用依賴樹