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
,参数包含了name
和security_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的依赖树便生成了
下章说说如何使用依赖树