在Starlette中戏蔑，請求的流動是基于Scope來實現(xiàn)的氧秘，到endpoint的前一步寻行，將Scope封裝成Request点晴，在FastAPI中感凤，Route節(jié)點(diǎn)→Endpoint的過程中，加入了大量邏輯粒督，其中包括依賴判斷陪竿，安全認(rèn)證，數(shù)據(jù)類型判斷等屠橄。那么這些具體是如何實現(xiàn)的呢族跛？
在FastAPI中APIRoute實例化時，會對endpoint的參數(shù)進(jìn)行解析锐墙。
這涉及到inspect庫礁哄，他可以解析函數(shù)的參數(shù)，包括其注釋的Typing贮匕，以及其默認(rèn)值姐仅，這為
id: str = Depends(get_id)這樣的表現(xiàn)形式提供了先決條件花枫。
這個過程分為兩個階段四個步驟：
- 配置response model
- 檢查參數(shù)刻盐，搜集類型和依賴掏膏。構(gòu)建dependant。
- 獲取到request報文敦锌，參照denpendant將參數(shù)注入
- 將返回值注入到response model中
首先來看response model相關(guān)
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"
# 一些狀態(tài)碼不允許有body
# 下面和pydantic有關(guān)
response_name = "Response_" + self.unique_id
self.response_field = create_response_field(
name=response_name, type_=self.response_model
)
# 創(chuàng)建字段的克隆馒疹，這樣Pydantic子模型就不會返回，因為它是一個更有限的類的子類的實例乙墙。
# UserInDB(包含hashed_password)可能是一個沒有hashed_password的User的子類颖变。
# 但是因為它是一個子類，所以它會通過驗證并原樣返回听想。
# 作為一個新字段腥刹，沒有繼承會原樣傳遞『郝颍總是會創(chuàng)建一個新的模型衔峰。
# 這段的意思是指，model的子類蛙粘，也會通過model的認(rèn)證垫卤，這是不可以的。
# 所以將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评抚。所以需要預(yù)置不同的驗證
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"
# 一些狀態(tài)碼不允許有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
# 創(chuàng)建{code: ModelField}字典
if response_fields:
# 不為空則掛載到實例
self.response_fields: Dict[Union[int, str], ModelField] = response_fields
else:
self.response_fields = {}
這里提到了一個類型: ModelField，他是根據(jù)model生成的一個專屬驗證對象
中間這段是對responses的內(nèi)容進(jìn)行拆解赵讯，關(guān)于responses我們之前提到過盈咳。
OpenAPI中的其他響應(yīng) - FastAPI
關(guān)于依賴的配置
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函數(shù)，將endpoint傳入進(jìn)去边翼。用來解析出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:
# 從函數(shù)參數(shù)字典中解析出字段。
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()
# 把參數(shù)中的Depends()抓出來
dependant = Dependant(call=call, name=name, path=path, use_cache=use_cache)
# 儲存用组底，沒有方法
# 配置各種參數(shù)丈积，例如 =Depends()，=Path()债鸡，=Query()江滨，或者是普通的參數(shù)。將其加入到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厌均，按節(jié)點(diǎn)尋找依賴項唬滑，將子節(jié)點(diǎn)的dependant加入到自身，向上返回dependant對象
# 2，獲取參數(shù)中依賴項的具體依賴函數(shù)
# 3晶密，將依賴函數(shù)進(jìn)行判斷擒悬，例如是否為安全相關(guān)，將其作為節(jié)點(diǎn)傳入到get_dependant中
# 這是個循環(huán)遍歷稻艰，最終建立dependant樹
# 根節(jié)點(diǎn)為endpoint的dependant對象
# 將他的依賴中的函數(shù)作為節(jié)點(diǎn)懂牧，再傳入到這個函數(shù)中
# 然后將子依賴的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等參數(shù)尊勿。
if add_non_field_param_to_dependency(param=param, dependant=dependant):
continue
# 查詢參數(shù)驗證
param_field = get_param_field(
param=param, default_field_info=params.Query, param_name=param_name
)
# 獲取參數(shù)的ModelField
# 路徑參數(shù) 例如"/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(), 對路徑參數(shù)進(jìn)行認(rèn)證的
# 設(shè)定為不忽視默認(rèn)值(忽視掉沒法用了)
# 改為路徑參數(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,
)
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)
# 當(dāng)為路徑參數(shù)僧凤，標(biāo)準(zhǔn)參數(shù)，標(biāo)準(zhǔn)序列參數(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(...)"
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:
# 將參數(shù)中的Depends中的具體依賴函數(shù)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 []
# 安全范圍
# 安全相關(guān)
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
)
# 將從依賴項中剝離出來的函數(shù)，再作為節(jié)點(diǎn)澎语，傳入到其中
# 最終形成依賴樹
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
- 按節(jié)點(diǎn)尋找依賴項吻氧，將子節(jié)點(diǎn)的dependant加入到自身，向上返回dependant對象
- 獲取參數(shù)中依賴項的具體依賴函數(shù)
- 將依賴函數(shù)進(jìn)行判斷咏连，例如是否為安全相關(guān)盯孙，將其作為節(jié)點(diǎn)傳入到get_dependant中
以上是對APIRoute封裝的大致過程
在這階段中，APIRoute對endpoint進(jìn)行解析祟滴，從中獲取關(guān)于參數(shù)和model的信息振惰。然后進(jìn)行配置，將APIRoute自身適應(yīng)化
