async def app(request: Request) -> Response:
......
solved_result = await solve_dependencies(
request=request,
dependant=dependant,
body=body,
dependency_overrides_provider=dependency_overrides_provider,
)
values, errors, background_tasks, sub_response, _ = solved_result
if errors:
raise RequestValidationError(errors, body=body)
else:
raw_response = await run_endpoint_function(
dependant=dependant, values=values, is_coroutine=is_coroutine
)
......
這里是endpoint的前一步虎眨,request首先要經(jīng)過solve_dependencies()
來與endpoint的依賴樹進(jìn)行匹配您单,這相當(dāng)于API的門衛(wèi)一般的存在梯投。
匹配的結(jié)果包含在solve過程中的結(jié)果稻薇,錯(cuò)誤,以及其他若干信息葵萎。有了詳細(xì)的信息导犹,我們便能精確的定位到問題的所在。
async def solve_dependencies(
*,
request: Union[Request, WebSocket],
dependant: Dependant,
body: Optional[Union[Dict[str, Any], FormData]] = None,
background_tasks: Optional[BackgroundTasks] = None,
response: Optional[Response] = None,
dependency_overrides_provider: Optional[Any] = None,
dependency_cache: Optional[Dict[Tuple[Callable, Tuple[str]], Any]] = None,
) -> Tuple[
Dict[str, Any],
List[ErrorWrapper],
Optional[BackgroundTasks],
Response,
Dict[Tuple[Callable, Tuple[str]], Any],
]:
"""
:param request: 請(qǐng)求報(bào)文
:param dependant: endpoint對(duì)應(yīng)的依賴樹
:param body: 請(qǐng)求體
:param background_tasks: 后臺(tái)任務(wù)
:param response: 子依賴
:param dependency_overrides_provider: app中設(shè)置的依賴替代項(xiàng)
:param dependency_cache: 已完成的依賴
:return:
"""
values: Dict[str, Any] = {}
errors: List[ErrorWrapper] = []
response = response or Response(
content=None,
status_code=None, # type: ignore
headers=None,
media_type=None,
background=None,
)
dependency_cache = dependency_cache or {}
開始解依賴樹
for sub_dependant in dependant.dependencies:
sub_dependant.call = cast(Callable, sub_dependant.call)
sub_dependant.cache_key = cast(
Tuple[Callable, Tuple[str]], sub_dependant.cache_key
)
# cast的作用是標(biāo)注類型羡忘,方便提示
call = sub_dependant.call
use_sub_dependant = sub_dependant
# 分別拿到依賴內(nèi)容和依賴項(xiàng)
if (
dependency_overrides_provider
and dependency_overrides_provider.dependency_overrides
):
# 依賴重寫時(shí)
original_call = sub_dependant.call
call = getattr(
dependency_overrides_provider, "dependency_overrides", {}
).get(original_call, original_call)
# 找到對(duì)應(yīng)的重寫
use_path: str = sub_dependant.path # type: ignore
use_sub_dependant = get_dependant(
path=use_path,
call=call,
name=sub_dependant.name,
security_scopes=sub_dependant.security_scopes,
)
# 重新生成依賴
use_sub_dependant.security_scopes = sub_dependant.security_scopes
solved_result = await solve_dependencies(
request=request,
dependant=use_sub_dependant,
body=body,
background_tasks=background_tasks,
response=response,
dependency_overrides_provider=dependency_overrides_provider,
dependency_cache=dependency_cache,
)
# 獲得依賴項(xiàng)的子依賴的結(jié)果集
(
sub_values,
sub_errors,
background_tasks,
_, # 子依賴項(xiàng)返回與我們相同的響應(yīng)
sub_dependency_cache,
) = solved_result
# 拿到結(jié)果和錯(cuò)誤
上面這部分負(fù)責(zé)解決子依賴谎痢,拿到最后的結(jié)果集,然后用子依賴的結(jié)果集來解決自身
dependency_cache.update(sub_dependency_cache)
# 將子依賴已解決的內(nèi)容注冊(cè)
if sub_errors:
errors.extend(sub_errors)
continue
if sub_dependant.use_cache and sub_dependant.cache_key in dependency_cache:
# 如果設(shè)置了使用緩存壳坪,且該依賴內(nèi)容已被子依賴解決過
solved = dependency_cache[sub_dependant.cache_key]
# 直接抄答案
# 否則照常執(zhí)行
elif is_gen_callable(call) or is_async_gen_callable(call):
stack = request.scope.get("fastapi_astack")
if stack is None:
raise RuntimeError(
async_contextmanager_dependencies_error
) # pragma: no cover
solved = await solve_generator(
call=call, stack=stack, sub_values=sub_values
)
# 用子依賴得到的結(jié)果集舶得,作為參數(shù),傳入到依賴內(nèi)容中爽蝴。
elif is_coroutine_callable(call):
solved = await call(**sub_values)
else:
solved = await run_in_threadpool(call, **sub_values)
# 對(duì)于同步異步不同的執(zhí)行方式
if sub_dependant.name is not None:
values[sub_dependant.name] = solved
# 收集結(jié)果
if sub_dependant.cache_key not in dependency_cache:
dependency_cache[sub_dependant.cache_key] = solved
# 將結(jié)果添加到結(jié)果集
解決該節(jié)點(diǎn)的解決每一個(gè)依賴項(xiàng)
# 依賴項(xiàng)不再有子依賴時(shí)沐批,會(huì)直接跳過上面的for循環(huán)
path_values, path_errors = request_params_to_args(
dependant.path_params, request.path_params
)
query_values, query_errors = request_params_to_args(
dependant.query_params, request.query_params
)
header_values, header_errors = request_params_to_args(
dependant.header_params, request.headers
)
cookie_values, cookie_errors = request_params_to_args(
dependant.cookie_params, request.cookies
)
# 生成依賴樹時(shí),我們將需要的參數(shù)蝎亚,保存到了path_params九孩,query_params等地方。
# 現(xiàn)在就是從Request中提取它們的好時(shí)機(jī)
# 當(dāng)然這個(gè)過程中也會(huì)產(chǎn)生錯(cuò)誤发框,我們會(huì)收集這些錯(cuò)誤
values.update(path_values)
values.update(query_values)
values.update(header_values)
values.update(cookie_values)
errors += path_errors + query_errors + header_errors + cookie_errors
# 合并現(xiàn)有錯(cuò)誤
解決參數(shù)依賴躺彬,分別遍歷依賴的個(gè)參數(shù)需求列表,與request所能提供的做匹配梅惯。
我們來看一下匹配的函數(shù)
def request_params_to_args(
required_params: Sequence[ModelField],
received_params: Union[Mapping[str, Any], QueryParams, Headers],
) -> Tuple[Dict[str, Any], List[ErrorWrapper]]:
values = {}
errors = []
for field in required_params:
if is_scalar_sequence_field(field) and isinstance(
received_params, (QueryParams, Headers)
):
value = received_params.getlist(field.alias) or field.default
else:
value = received_params.get(field.alias)
# 分別對(duì)標(biāo)準(zhǔn)序列參數(shù)和其他參數(shù)的情況進(jìn)行處理宪拥,拿到value
# 這里未處理默認(rèn)值的情況下
field_info = field.field_info
assert isinstance(
field_info, params.Param
), "Params must be subclasses of Param"
if value is None:
if field.required:
errors.append(
ErrorWrapper(
MissingError(), loc=(field_info.in_.value, field.alias)
)
)
# 必須則引發(fā)錯(cuò)誤
else:
values[field.name] = deepcopy(field.default)
# 否則使用默認(rèn)值
continue
v_, errors_ = field.validate(
value, values, loc=(field_info.in_.value, field.alias)
)
if isinstance(errors_, ErrorWrapper):
errors.append(errors_)
elif isinstance(errors_, list):
errors.extend(errors_)
else:
values[field.name] = v_
return values, errors
回到solve_dependencies
if dependant.body_params:
# 如果有user_info(user: User)這樣的參數(shù),User為Model铣减。
# 其會(huì)保存到body_params中她君,現(xiàn)在是處理它們的時(shí)候
(
body_values,
body_errors,
) = await request_body_to_args( # body_params checked above
required_params=dependant.body_params, received_body=body
)
# body傳入進(jìn)去,進(jìn)行匹配葫哗,得到結(jié)果
values.update(body_values)
errors.extend(body_errors)
# 整合結(jié)果
if dependant.http_connection_param_name:
values[dependant.http_connection_param_name] = request
if dependant.request_param_name and isinstance(request, Request):
values[dependant.request_param_name] = request
elif dependant.websocket_param_name and isinstance(request, WebSocket):
values[dependant.websocket_param_name] = request
# 這三者是解決需要特定參數(shù)的時(shí)候缔刹,主要是指Request或WebSocket這樣的參數(shù)
if dependant.background_tasks_param_name:
if background_tasks is None:
background_tasks = BackgroundTasks()
values[dependant.background_tasks_param_name] = background_tasks
# 后臺(tái)任務(wù)
if dependant.response_param_name:
values[dependant.response_param_name] = response
# 如果需要操作Response報(bào)文球涛,這里會(huì)提供sub response,其內(nèi)容最后會(huì)整合到response中
if dependant.security_scopes_param_name:
values[dependant.security_scopes_param_name] = SecurityScopes(
scopes=dependant.security_scopes
)
# 拿到安全域
return values, errors, background_tasks, response, dependency_cache
solve_dependencies
本身也是遞歸函數(shù)校镐,這和get_dependant
是相輔相成的亿扁。