Add `form_max_**` parameters for `post`

This commit is contained in:
Yurii Motov 2025-12-03 22:28:08 +01:00
parent c57ac7bdf3
commit 72454fd83b
2 changed files with 101 additions and 1 deletions

View File

@ -2624,6 +2624,39 @@ class FastAPI(Starlette):
"""
),
] = Default(generate_unique_id),
form_max_fields: Annotated[
int,
Doc(
"""
Maximum number of form fields to accept.
This limits the number of fields in a form submission to prevent
potential denial-of-service attacks.
"""
),
] = 1000,
form_max_files: Annotated[
int,
Doc(
"""
Maximum number of files to accept in a form submission.
This limits the number of files in a form submission to prevent
potential denial-of-service attacks.
"""
),
] = 1000,
form_max_part_size: Annotated[
int,
Doc(
"""
Maximum size (in bytes) for each part in a form submission.
This limits the size of each part in a form submission to prevent
potential denial-of-service attacks.
"""
),
] = 1024 * 1024,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP POST operation.
@ -2669,6 +2702,9 @@ class FastAPI(Starlette):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
form_max_fields=form_max_fields,
form_max_files=form_max_files,
form_max_part_size=form_max_part_size,
)
def delete(

View File

@ -300,6 +300,9 @@ def get_request_handler(
response_model_exclude_none: bool = False,
dependency_overrides_provider: Optional[Any] = None,
embed_body_fields: bool = False,
form_max_fields: int = 1000,
form_max_files: int = 1000,
form_max_part_size: int = 1024 * 1024,
) -> Callable[[Request], Coroutine[Any, Any, Response]]:
assert dependant.call is not None, "dependant.call must be a function"
is_coroutine = dependant.is_coroutine_callable
@ -323,7 +326,11 @@ def get_request_handler(
body: Any = None
if body_field:
if is_body_form:
body = await request.form()
body = await request.form(
max_fields=form_max_fields,
max_files=form_max_files,
max_part_size=form_max_part_size,
)
file_stack.push_async_callback(body.close)
else:
body_bytes = await request.body()
@ -535,6 +542,9 @@ class APIRoute(routing.Route):
generate_unique_id_function: Union[
Callable[["APIRoute"], str], DefaultPlaceholder
] = Default(generate_unique_id),
form_max_fields: int = 1000,
form_max_files: int = 1000,
form_max_part_size: int = 1024 * 1024,
) -> None:
self.path = path
self.endpoint = endpoint
@ -565,6 +575,9 @@ class APIRoute(routing.Route):
self.responses = responses or {}
self.name = get_name(endpoint) if name is None else name
self.path_regex, self.path_format, self.param_convertors = compile_path(path)
self.form_max_fields = form_max_fields
self.form_max_files = form_max_files
self.form_max_part_size = form_max_part_size
if methods is None:
methods = ["GET"]
self.methods: Set[str] = {method.upper() for method in methods}
@ -661,6 +674,9 @@ class APIRoute(routing.Route):
response_model_exclude_none=self.response_model_exclude_none,
dependency_overrides_provider=self.dependency_overrides_provider,
embed_body_fields=self._embed_body_fields,
form_max_fields=self.form_max_fields,
form_max_files=self.form_max_files,
form_max_part_size=self.form_max_part_size,
)
def matches(self, scope: Scope) -> Tuple[Match, Scope]:
@ -989,6 +1005,9 @@ class APIRouter(routing.Router):
generate_unique_id_function: Union[
Callable[[APIRoute], str], DefaultPlaceholder
] = Default(generate_unique_id),
form_max_fields: int = 1000,
form_max_files: int = 1000,
form_max_part_size: int = 1024 * 1024,
) -> None:
route_class = route_class_override or self.route_class
responses = responses or {}
@ -1035,6 +1054,9 @@ class APIRouter(routing.Router):
callbacks=current_callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=current_generate_unique_id,
form_max_fields=form_max_fields,
form_max_files=form_max_files,
form_max_part_size=form_max_part_size,
)
self.routes.append(route)
@ -1067,6 +1089,9 @@ class APIRouter(routing.Router):
generate_unique_id_function: Callable[[APIRoute], str] = Default(
generate_unique_id
),
form_max_fields: int = 1000,
form_max_files: int = 1000,
form_max_part_size: int = 1024 * 1024,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
def decorator(func: DecoratedCallable) -> DecoratedCallable:
self.add_api_route(
@ -1095,6 +1120,9 @@ class APIRouter(routing.Router):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
form_max_fields=form_max_fields,
form_max_files=form_max_files,
form_max_part_size=form_max_part_size,
)
return func
@ -2531,6 +2559,39 @@ class APIRouter(routing.Router):
"""
),
] = Default(generate_unique_id),
form_max_fields: Annotated[
int,
Doc(
"""
Maximum number of form fields to accept.
This limits the number of fields in a form submission to prevent
potential denial-of-service attacks.
"""
),
] = 1000,
form_max_files: Annotated[
int,
Doc(
"""
Maximum number of files to accept in a form submission.
This limits the number of files in a form submission to prevent
potential denial-of-service attacks.
"""
),
] = 1000,
form_max_part_size: Annotated[
int,
Doc(
"""
Maximum size (in bytes) for each part in a form submission.
This limits the size of each part in a form submission to prevent
potential denial-of-service attacks.
"""
),
] = 1024 * 1024,
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP POST operation.
@ -2580,6 +2641,9 @@ class APIRouter(routing.Router):
callbacks=callbacks,
openapi_extra=openapi_extra,
generate_unique_id_function=generate_unique_id_function,
form_max_fields=form_max_fields,
form_max_files=form_max_files,
form_max_part_size=form_max_part_size,
)
def delete(