return flat_dependant
+def _get_flat_body_params(dependant: Dependant) -> list[ModelField]:
+ body_params: list[ModelField] = []
+ dependants = [dependant]
+ while dependants:
+ current_dependant = dependants.pop()
+ body_params.extend(current_dependant.body_params)
+ dependants.extend(reversed(current_dependant.dependencies))
+ return body_params
+
+
def _get_flat_fields_from_params(fields: list[ModelField]) -> list[ModelField]:
if not fields:
return fields
return values, errors
-def get_body_field(
- *, flat_dependant: Dependant, name: str, embed_body_fields: bool
+def _get_body_field(
+ *, body_params: list[ModelField], name: str, embed_body_fields: bool
) -> ModelField | None:
"""
Get a ModelField representing the request body for a path operation, combining
This is **not** used to validate/parse the request body, that's done with each
individual body parameter.
"""
- if not flat_dependant.body_params:
+ if not body_params:
return None
- first_param = flat_dependant.body_params[0]
+ first_param = body_params[0]
if not embed_body_fields:
return first_param
model_name = "Body_" + name
- BodyModel = create_body_model(
- fields=flat_dependant.body_params, model_name=model_name
- )
- required = any(
- True for f in flat_dependant.body_params if f.field_info.is_required()
- )
+ BodyModel = create_body_model(fields=body_params, model_name=model_name)
+ required = any(True for f in body_params if f.field_info.is_required())
BodyFieldInfo_kwargs: dict[str, Any] = {
"annotation": BodyModel,
"alias": "body",
}
if not required:
BodyFieldInfo_kwargs["default"] = None
- if any(isinstance(f.field_info, params.File) for f in flat_dependant.body_params):
+ if any(isinstance(f.field_info, params.File) for f in body_params):
BodyFieldInfo: type[params.Body] = params.File
- elif any(isinstance(f.field_info, params.Form) for f in flat_dependant.body_params):
+ elif any(isinstance(f.field_info, params.Form) for f in body_params):
BodyFieldInfo = params.Form
else:
BodyFieldInfo = params.Body
body_param_media_types = [
f.field_info.media_type
- for f in flat_dependant.body_params
+ for f in body_params
if isinstance(f.field_info, params.Body)
]
if len(set(body_param_media_types)) == 1:
_is_gen_callable,
)
from fastapi.dependencies.utils import (
+ _get_body_field,
+ _get_flat_body_params,
_should_embed_body_fields,
- get_body_field,
get_dependant,
- get_flat_dependant,
get_parameterless_sub_dependant,
get_stream_item_type,
get_typed_return_annotation,
path: str,
call: Callable[..., Any],
dependencies: Sequence[params.Depends],
-) -> tuple[Dependant, Dependant, bool]:
+) -> tuple[Dependant, list[ModelField], bool]:
dependant = get_dependant(path=path, call=call, scope="function")
for depends in dependencies[::-1]:
dependant.dependencies.insert(
0,
get_parameterless_sub_dependant(depends=depends, path=path),
)
- flat_dependant = get_flat_dependant(dependant)
- embed_body_fields = _should_embed_body_fields(flat_dependant.body_params)
- return dependant, flat_dependant, embed_body_fields
+ body_params = _get_flat_body_params(dependant)
+ embed_body_fields = _should_embed_body_fields(body_params)
+ return dependant, body_params, embed_body_fields
class _RouteWithPath(Protocol):
assert callable(endpoint), "An endpoint must be a callable"
(
route.dependant,
- flat_dependant,
+ body_params,
route._embed_body_fields,
) = _build_dependant_with_parameterless_dependencies(
path=route.path_format,
call=route.endpoint,
dependencies=route.dependencies,
)
- route.body_field = get_body_field(
- flat_dependant=flat_dependant,
+ route.body_field = _get_body_field(
+ body_params=body_params,
name=route.unique_id,
embed_body_fields=route._embed_body_fields,
)