在Starlette中,请求的流动是基于Scope来实现的,到endpoint的前一步,将Scope封装成Request,在FastAPI中,Route节点→Endpoint的过程中,加入了大量逻辑,其中包括依赖判断,安全认证,数据类型判断等。那么这些具体是如何实现的呢?
在FastAPI中APIRoute实例化时,会对endpoint的参数进行解析。
这涉及到inspect库,他可以解析函数的参数,包括其注释的Typing,以及其默认值,这为
id: str = Depends(get_id)
这样的表现形式提供了先决条件。
这个过程分为两个阶段四个步骤:
- 配置response model
- 检查参数,搜集类型和依赖。构建dependant。
- 获取到request报文,参照denpendant将参数注入
- 将返回值注入到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
- 按节点寻找依赖项,将子节点的dependant加入到自身,向上返回dependant对象
- 获取参数中依赖项的具体依赖函数
- 将依赖函数进行判断,例如是否为安全相关,将其作为节点传入到get_dependant中
以上是对APIRoute封装的大致过程
在这阶段中,APIRoute对endpoint进行解析,从中获取关于参数和model的信息。然后进行配置,将APIRoute自身适应化
网友评论