import dataclasses import inspect import sys from collections.abc import ( AsyncGenerator, AsyncIterable, AsyncIterator, Callable, Generator, Iterable, Iterator, Mapping, Sequence, ) from contextlib import AsyncExitStack, contextmanager from copy import copy, deepcopy from dataclasses import dataclass from functools import partial from typing import ( Annotated, Any, ForwardRef, Literal, Union, cast, get_args, get_origin, ) from fastapi import params from fastapi._compat import ( ModelField, RequiredParam, Undefined, copy_field_info, create_body_model, evaluate_forwardref, field_annotation_is_scalar, field_annotation_is_scalar_sequence, field_annotation_is_sequence, get_cached_model_fields, get_missing_field_error, is_bytes_or_nonable_bytes_annotation, is_bytes_sequence_annotation, is_scalar_field, is_uploadfile_or_nonable_uploadfile_annotation, is_uploadfile_sequence_annotation, lenient_issubclass, sequence_types, serialize_sequence_value, value_is_sequence, ) from fastapi.background import BackgroundTasks from fastapi.concurrency import ( asynccontextmanager, contextmanager_in_threadpool, ) from fastapi.dependencies.models import ( Dependant, _get_cache_key, _get_computed_scope, _get_oauth_scopes, _is_async_gen_callable, _is_coroutine_callable, _is_gen_callable, _UsesScopesCache, ) from fastapi.exceptions import DependencyScopeError from fastapi.logger import logger from fastapi.security.oauth2 import SecurityScopes from fastapi.types import DependencyCacheKey, UnionType from fastapi.utils import create_model_field, get_path_param_names from pydantic import ( AliasChoices, AliasPath, BaseModel, BeforeValidator, ConfigDict, GetJsonSchemaHandler, GetPydanticSchema, Json, Tag, ) from pydantic.fields import FieldInfo from pydantic_core import CoreSchema from starlette.background import BackgroundTasks as StarletteBackgroundTasks from starlette.concurrency import run_in_threadpool from starlette.datastructures import ( FormData, Headers, ImmutableMultiDict, QueryParams, UploadFile, ) from starlette.requests import HTTPConnection, Request from starlette.responses import Response from starlette.websockets import WebSocket from typing_inspection.typing_objects import is_typealiastype multipart_not_installed_error = ( 'Form data requires "python-multipart" to be installed. \n' 'You can install "python-multipart" with: \n\n' "pip install python-multipart\n" ) multipart_incorrect_install_error = ( 'Form data requires "python-multipart" to be installed. ' 'It seems you installed "multipart" instead. \n' 'You can remove "multipart" with: \n\n' "pip uninstall multipart\n\n" 'And then install "python-multipart" with: \n\n' "pip install python-multipart\n" ) def ensure_multipart_is_installed() -> None: try: from python_multipart import __version__ # Import an attribute that can be mocked/deleted in testing assert __version__ > "0.0.12" except (ImportError, AssertionError): try: # __version__ is available in both multiparts, and can be mocked from multipart import ( # type: ignore[no-redef,import-untyped] __version__, ) assert __version__ try: # parse_options_header is only available in the right multipart from multipart.multipart import ( # type: ignore[import-untyped] parse_options_header, ) assert parse_options_header except ImportError: logger.error(multipart_incorrect_install_error) raise RuntimeError(multipart_incorrect_install_error) from None except ImportError: logger.error(multipart_not_installed_error) raise RuntimeError(multipart_not_installed_error) from None def get_parameterless_sub_dependant(*, depends: params.Depends, path: str) -> Dependant: assert callable(depends.dependency), ( "A parameter-less dependency must have a callable dependency" ) own_oauth_scopes: list[str] = [] if isinstance(depends, params.Security) and depends.scopes: own_oauth_scopes.extend(depends.scopes) return get_dependant( path=path, call=depends.dependency, scope=depends.scope, own_oauth_scopes=own_oauth_scopes, ) 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_model_type(annotation: Any) -> type[BaseModel] | None: if get_origin(annotation) in (Union, UnionType): args = get_args(annotation) if len(args) == 2 and type(None) in args: annotation = next(arg for arg in args if arg is not type(None)) if lenient_issubclass(annotation, BaseModel): return annotation return None def _get_param_model(field: ModelField) -> type[BaseModel] | None: annotation = field.field_info.annotation if isinstance(field.field_info, (params.Query, params.Header, params.Cookie)): annotation = _get_model_type(annotation) if lenient_issubclass(annotation, BaseModel) and not _is_json_field(field): return annotation return None def _get_flat_fields_from_params(fields: Sequence[ModelField]) -> list[ModelField]: flat_fields: list[ModelField] = [] for field in fields: if (model := _get_param_model(field)) is not None: flat_fields.extend(get_cached_model_fields(model)) else: flat_fields.append(field) return flat_fields def get_flat_params(dependant: Dependant) -> list[ModelField]: path_params: list[ModelField] = [] query_params: list[ModelField] = [] header_params: list[ModelField] = [] cookie_params: list[ModelField] = [] visited: list[DependencyCacheKey] = [] uses_scopes_cache: _UsesScopesCache = {} dependants = [dependant] while dependants: current_dependant = dependants.pop() cache_key = _get_cache_key( dependant=current_dependant, uses_scopes_cache=uses_scopes_cache, ) if cache_key in visited: continue visited.append(cache_key) path_params.extend(current_dependant.path_params) query_params.extend(current_dependant.query_params) header_params.extend(current_dependant.header_params) cookie_params.extend(current_dependant.cookie_params) dependants.extend(reversed(current_dependant.dependencies)) path_params = _get_flat_fields_from_params(path_params) query_params = _get_flat_fields_from_params(query_params) header_params = _get_flat_fields_from_params(header_params) cookie_params = _get_flat_fields_from_params(cookie_params) return path_params + query_params + header_params + cookie_params def _get_signature(call: Callable[..., Any]) -> inspect.Signature: try: signature = inspect.signature(call, eval_str=True) except NameError: # Handle type annotations with if TYPE_CHECKING, not used by FastAPI # e.g. dependency return types if sys.version_info >= (3, 14): from annotationlib import Format signature = inspect.signature(call, annotation_format=Format.FORWARDREF) else: signature = inspect.signature(call) return signature def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature: signature = _get_signature(call) unwrapped = inspect.unwrap(call) globalns = getattr(unwrapped, "__globals__", {}) typed_params = [ inspect.Parameter( name=param.name, kind=param.kind, default=param.default, annotation=get_typed_annotation(param.annotation, globalns), ) for param in signature.parameters.values() ] typed_signature = inspect.Signature(typed_params) return typed_signature def get_typed_annotation(annotation: Any, globalns: dict[str, Any]) -> Any: if isinstance(annotation, str): annotation = ForwardRef(annotation) annotation = evaluate_forwardref(annotation, globalns, globalns) if annotation is type(None): return None return annotation def get_typed_return_annotation(call: Callable[..., Any]) -> Any: signature = _get_signature(call) unwrapped = inspect.unwrap(call) annotation = signature.return_annotation if annotation is inspect.Signature.empty: return None globalns = getattr(unwrapped, "__globals__", {}) return get_typed_annotation(annotation, globalns) _STREAM_ORIGINS = { AsyncIterable, AsyncIterator, AsyncGenerator, Iterable, Iterator, Generator, } def get_stream_item_type(annotation: Any) -> Any | None: origin = get_origin(annotation) if origin is not None and origin in _STREAM_ORIGINS: type_args = get_args(annotation) if type_args: return type_args[0] return Any return None def get_dependant( *, path: str, call: Callable[..., Any], name: str | None = None, own_oauth_scopes: list[str] | None = None, parent_oauth_scopes: list[str] | None = None, use_cache: bool = True, scope: Literal["function", "request"] | None = None, ) -> Dependant: dependant = Dependant( call=call, name=name, path=path, use_cache=use_cache, scope=scope, own_oauth_scopes=own_oauth_scopes, parent_oauth_scopes=parent_oauth_scopes, ) current_scopes = (parent_oauth_scopes or []) + (own_oauth_scopes or []) path_param_names = get_path_param_names(path) endpoint_signature = get_typed_signature(call) signature_params = endpoint_signature.parameters for param_name, param in signature_params.items(): param_details = analyze_param( param_name=param_name, annotation=param.annotation, value=param.default, path_param_names=path_param_names, ) if param_details.depends is not None: assert param_details.depends.dependency if ( ( _is_gen_callable(dependant.call) or _is_async_gen_callable(dependant.call) ) and _get_computed_scope(dependant=dependant) == "request" and param_details.depends.scope == "function" ): assert dependant.call call_name = getattr(dependant.call, "__name__", "") raise DependencyScopeError( f'The dependency "{call_name}" has a scope of ' '"request", it cannot depend on dependencies with scope "function".' ) sub_own_oauth_scopes: list[str] = [] if isinstance(param_details.depends, params.Security): if param_details.depends.scopes: sub_own_oauth_scopes = list(param_details.depends.scopes) sub_dependant = get_dependant( path=path, call=param_details.depends.dependency, name=param_name, own_oauth_scopes=sub_own_oauth_scopes, parent_oauth_scopes=current_scopes, use_cache=param_details.depends.use_cache, scope=param_details.depends.scope, ) dependant.dependencies.append(sub_dependant) continue if add_non_field_param_to_dependency( param_name=param_name, type_annotation=param_details.type_annotation, dependant=dependant, ): assert param_details.field is None, ( f"Cannot specify multiple FastAPI annotations for {param_name!r}" ) continue assert param_details.field is not None if isinstance(param_details.field.field_info, params.Body): dependant.body_params.append(param_details.field) else: add_param_to_fields(field=param_details.field, dependant=dependant) return dependant def add_non_field_param_to_dependency( *, param_name: str, type_annotation: Any, dependant: Dependant ) -> bool | None: if lenient_issubclass(type_annotation, Request): dependant.request_param_name = param_name return True elif lenient_issubclass(type_annotation, WebSocket): dependant.websocket_param_name = param_name return True elif lenient_issubclass(type_annotation, HTTPConnection): dependant.http_connection_param_name = param_name return True elif lenient_issubclass(type_annotation, Response): dependant.response_param_name = param_name return True elif lenient_issubclass(type_annotation, StarletteBackgroundTasks): dependant.background_tasks_param_name = param_name return True elif lenient_issubclass(type_annotation, SecurityScopes): dependant.security_scopes_param_name = param_name return True return None @dataclass class ParamDetails: type_annotation: Any depends: params.Depends | None field: ModelField | None def analyze_param( *, param_name: str, annotation: Any, value: Any, path_param_names: set[str], ) -> ParamDetails: is_path_param = param_name in path_param_names field_info = None depends = None type_annotation: Any = Any use_annotation: Any = Any if is_typealiastype(annotation): # unpack in case PEP 695 type syntax is used annotation = annotation.__value__ if annotation is not inspect.Signature.empty: use_annotation = annotation type_annotation = annotation # Extract Annotated info if get_origin(use_annotation) is Annotated: annotated_args = get_args(annotation) type_annotation = annotated_args[0] fastapi_annotations = [ arg for arg in annotated_args[1:] if isinstance(arg, (FieldInfo, params.Depends)) ] fastapi_specific_annotations = [ arg for arg in fastapi_annotations if isinstance( arg, ( params.Param, params.Body, params.Depends, ), ) ] if fastapi_specific_annotations: fastapi_annotation: FieldInfo | params.Depends | None = ( fastapi_specific_annotations[-1] ) else: fastapi_annotation = None # Set default for Annotated FieldInfo if isinstance(fastapi_annotation, FieldInfo): # Copy `field_info` because we mutate `field_info.default` below. field_info = copy_field_info( field_info=fastapi_annotation, annotation=use_annotation, ) assert ( field_info.default == Undefined or field_info.default == RequiredParam ), ( f"`{field_info.__class__.__name__}` default value cannot be set in" f" `Annotated` for {param_name!r}. Set the default value with `=` instead." ) if value is not inspect.Signature.empty: assert not is_path_param, "Path parameters cannot have default values" field_info.default = value else: field_info.default = RequiredParam # Get Annotated Depends elif isinstance(fastapi_annotation, params.Depends): depends = fastapi_annotation # Get Depends from default value if isinstance(value, params.Depends): assert depends is None, ( "Cannot specify `Depends` in `Annotated` and default value" f" together for {param_name!r}" ) assert field_info is None, ( "Cannot specify a FastAPI annotation in `Annotated` and `Depends` as a" f" default value together for {param_name!r}" ) depends = value # Get FieldInfo from default value elif isinstance(value, FieldInfo): assert field_info is None, ( "Cannot specify FastAPI annotations in `Annotated` and default value" f" together for {param_name!r}" ) field_info = value if isinstance(field_info, FieldInfo): field_info.annotation = type_annotation # Get Depends from type annotation if depends is not None and depends.dependency is None: # Copy `depends` before mutating it depends = copy(depends) depends = dataclasses.replace(depends, dependency=type_annotation) # Handle non-param type annotations like Request # Only apply special handling when there's no explicit Depends - if there's a Depends, # the dependency will be called and its return value used instead of the special injection if depends is None and lenient_issubclass( type_annotation, ( Request, WebSocket, HTTPConnection, Response, StarletteBackgroundTasks, SecurityScopes, ), ): assert field_info is None, ( f"Cannot specify FastAPI annotation for type {type_annotation!r}" ) # Handle default assignations, neither field_info nor depends was not found in Annotated nor default value elif field_info is None and depends is None: default_value = value if value is not inspect.Signature.empty else RequiredParam if is_path_param: # We might check here that `default_value is RequiredParam`, but the fact is that the same # parameter might sometimes be a path parameter and sometimes not. See # `tests/test_infer_param_optionality.py` for an example. field_info = params.Path(annotation=use_annotation) elif is_uploadfile_or_nonable_uploadfile_annotation( type_annotation ) or is_uploadfile_sequence_annotation(type_annotation): field_info = params.File(annotation=use_annotation, default=default_value) elif not field_annotation_is_scalar(annotation=type_annotation): field_info = params.Body(annotation=use_annotation, default=default_value) else: field_info = params.Query(annotation=use_annotation, default=default_value) if get_origin(use_annotation) is Annotated: inferred_default = field_info.default field_info = copy_field_info( field_info=field_info, annotation=use_annotation, ) field_info.default = inferred_default if is_path_param: # A default factory can apply when this endpoint is used without the # path parameter, but path parameters must always be required. field_info.default_factory = None field = None # It's a field_info, not a dependency if field_info is not None: # Handle field_info.in_ if is_path_param: assert isinstance(field_info, params.Path), ( f"Cannot use `{field_info.__class__.__name__}` for path param" f" {param_name!r}" ) elif ( isinstance(field_info, params.Param) and getattr(field_info, "in_", None) is None ): field_info.in_ = params.ParamTypes.query use_annotation_from_field_info = use_annotation if isinstance(field_info, params.Form): ensure_multipart_is_installed() if _get_form_union_models(field_info.annotation): field_info.annotation = _add_form_union_validators( field_info.annotation ) validation_alias = field_info.validation_alias if ( isinstance(field_info, params.Path) and isinstance(validation_alias, AliasChoices) and all(isinstance(choice, str) for choice in validation_alias.choices) ): for choice in cast(list[str], validation_alias.choices): if choice in path_param_names: # Resolve the route name for both validation and OpenAPI without # changing metadata shared with other routes. field_info = copy(field_info) field_info.validation_alias = choice break if not field_info.alias and getattr(field_info, "convert_underscores", None): alias = param_name.replace("_", "-") else: alias = field_info.alias or param_name field_info.alias = alias field = create_model_field( name=param_name, type_=use_annotation_from_field_info, default=field_info.default, alias=alias, field_info=field_info, ) if is_path_param: assert is_scalar_field(field=field), ( "Path params must be of one of the supported types" ) elif isinstance(field_info, params.Query): assert ( is_scalar_field(field) or field_annotation_is_scalar_sequence(field.field_info.annotation) or _get_model_type(field.field_info.annotation) is not None ), f"Query parameter {param_name!r} must be one of the supported types" return ParamDetails(type_annotation=type_annotation, depends=depends, field=field) def add_param_to_fields(*, field: ModelField, dependant: Dependant) -> None: field_info = field.field_info field_info_in = getattr(field_info, "in_", None) if field_info_in == params.ParamTypes.path: dependant.path_params.append(field) elif field_info_in == params.ParamTypes.query: dependant.query_params.append(field) elif field_info_in == params.ParamTypes.header: dependant.header_params.append(field) else: assert field_info_in == params.ParamTypes.cookie, ( f"non-body parameters must be in path, query, header or cookie: {field.name}" ) dependant.cookie_params.append(field) async def _solve_generator( *, dependant: Dependant, stack: AsyncExitStack, sub_values: dict[str, Any] ) -> Any: assert dependant.call if _is_async_gen_callable(dependant.call): cm = asynccontextmanager(dependant.call)(**sub_values) elif _is_gen_callable(dependant.call): cm = contextmanager_in_threadpool(contextmanager(dependant.call)(**sub_values)) return await stack.enter_async_context(cm) @dataclass class SolvedDependency: values: dict[str, Any] errors: list[Any] background_tasks: StarletteBackgroundTasks | None response: Response dependency_cache: dict[DependencyCacheKey, Any] async def solve_dependencies( *, request: Request | WebSocket, dependant: Dependant, body: Any = Undefined, background_tasks: StarletteBackgroundTasks | None = None, response: Response | None = None, dependency_overrides_provider: Any | None = None, dependency_cache: dict[DependencyCacheKey, Any] | None = None, # TODO: remove this parameter later, no longer used, not removing it yet as some # people might be monkey patching this function (although that's not supported) async_exit_stack: AsyncExitStack, embed_body_fields: bool, _uses_scopes_cache: _UsesScopesCache | None = None, ) -> SolvedDependency: request_astack = request.scope.get("fastapi_inner_astack") assert isinstance(request_astack, AsyncExitStack), ( "fastapi_inner_astack not found in request scope" ) function_astack = request.scope.get("fastapi_function_astack") assert isinstance(function_astack, AsyncExitStack), ( "fastapi_function_astack not found in request scope" ) values: dict[str, Any] = {} errors: list[Any] = [] if response is None: response = Response() del response.headers["content-length"] response.status_code = None # type: ignore if dependency_cache is None: dependency_cache = {} if _uses_scopes_cache is None: _uses_scopes_cache = {} for sub_dependant in dependant.dependencies: sub_dependant.call = cast(Callable[..., Any], sub_dependant.call) call = sub_dependant.call use_sub_dependant = sub_dependant if ( dependency_overrides_provider and dependency_overrides_provider.dependency_overrides ): original_call = sub_dependant.call call = getattr( dependency_overrides_provider, "dependency_overrides", {} ).get(original_call, original_call) use_path: str = sub_dependant.path # type: ignore use_sub_dependant = get_dependant( path=use_path, call=call, name=sub_dependant.name, parent_oauth_scopes=_get_oauth_scopes(dependant=sub_dependant), scope=sub_dependant.scope, ) 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, async_exit_stack=async_exit_stack, embed_body_fields=embed_body_fields, _uses_scopes_cache=_uses_scopes_cache, ) background_tasks = solved_result.background_tasks if solved_result.errors: errors.extend(solved_result.errors) continue sub_dependant_cache_key = _get_cache_key( dependant=sub_dependant, uses_scopes_cache=_uses_scopes_cache, ) if sub_dependant.use_cache and sub_dependant_cache_key in dependency_cache: solved = dependency_cache[sub_dependant_cache_key] elif _is_gen_callable(use_sub_dependant.call) or _is_async_gen_callable( use_sub_dependant.call ): use_astack = request_astack if sub_dependant.scope == "function": use_astack = function_astack solved = await _solve_generator( dependant=use_sub_dependant, stack=use_astack, sub_values=solved_result.values, ) elif _is_coroutine_callable(use_sub_dependant.call): solved = await call(**solved_result.values) else: solved = await run_in_threadpool(call, **solved_result.values) if sub_dependant.name is not None: values[sub_dependant.name] = solved if sub_dependant_cache_key not in dependency_cache: dependency_cache[sub_dependant_cache_key] = solved 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, ) 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 if dependant.body_params: ( body_values, body_errors, ) = await request_body_to_args( # body_params checked above body_fields=dependant.body_params, received_body=body, embed_body_fields=embed_body_fields, ) values.update(body_values) errors.extend(body_errors) 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 if dependant.background_tasks_param_name: if background_tasks is None: background_tasks = BackgroundTasks() values[dependant.background_tasks_param_name] = background_tasks if dependant.response_param_name: values[dependant.response_param_name] = response if dependant.security_scopes_param_name: values[dependant.security_scopes_param_name] = SecurityScopes( scopes=_get_oauth_scopes(dependant=dependant) ) return SolvedDependency( values=values, errors=errors, background_tasks=background_tasks, response=response, dependency_cache=dependency_cache, ) def _validate_value_with_model_field( *, field: ModelField, value: Any, values: dict[str, Any], loc: tuple[str, ...] ) -> tuple[Any, list[Any]]: if value is Undefined: if field.field_info.is_required(): return None, [get_missing_field_error(loc=loc)] value = deepcopy(field.default) if not field.field_info.validate_default: return value, [] return field.validate(value, values, loc=loc) def _is_json_field(field: ModelField) -> bool: return any(type(item) is Json for item in field.field_info.metadata) def _is_empty_file_input(value: Any) -> bool: # Browsers send an empty filename and content when no file is selected. return isinstance(value, UploadFile) and value.filename == "" and value.size == 0 def _get_multidict_value( field: ModelField, values: Mapping[str, Any], alias: str ) -> Any: if ( (not _is_json_field(field)) and field_annotation_is_sequence(field.field_info.annotation) and isinstance(values, (ImmutableMultiDict, Headers)) ): value = values.getlist(alias) if isinstance(values, FormData): value = [item for item in value if not _is_empty_file_input(item)] else: value = values.get(alias, None) if ( value is None or ( isinstance(values, FormData) and ( (isinstance(value, str) and value == "") or _is_empty_file_input(value) ) ) or ( field_annotation_is_sequence(field.field_info.annotation) and len(value) == 0 ) ): # Apply defaults during validation, not while extracting request data. return Undefined return value def _request_param_model_to_args( model_field: ModelField, model: type[BaseModel], received_params: Mapping[str, Any], ) -> tuple[Any, list[Any]]: fields_to_extract = get_cached_model_fields(model) default_convert_underscores = getattr( model_field.field_info, "convert_underscores", True ) params_to_process: dict[str, Any] = {} header_aliases: dict[str, str] = {} processed_keys = set() own_names: set[str] = set() for field in fields_to_extract: convert_underscores = isinstance(received_params, Headers) and getattr( field.field_info, "convert_underscores", default_convert_underscores ) aliases = _get_param_aliases(field, convert_underscores=convert_underscores) own_names.update( name.lower() if isinstance(received_params, Headers) else name for _, name in aliases ) alias, request_alias = next( (pair for pair in aliases if pair[1] in received_params), aliases[0] ) # Only reshape the selected alias. Unused choices can be extra fields # or aliases used by other fields with different types. value = _get_multidict_value(field, received_params, alias=request_alias) if value is not Undefined: params_to_process[alias] = value if isinstance(received_params, Headers): processed_keys.add(request_alias.lower()) for validation_alias, header_alias in aliases: if validation_alias != header_alias: # Do not accept an unconverted field name as an extra header. processed_keys.add(validation_alias.lower()) if alias != request_alias: header_aliases[field.name] = request_alias else: processed_keys.add(request_alias) # Let each model apply its extra policy to all remaining inputs. for key in received_params.keys(): if key not in processed_keys: if isinstance(received_params, (ImmutableMultiDict, Headers)): value = received_params.getlist(key) if isinstance(value, list) and (len(value) == 1): params_to_process[key] = value[0] else: params_to_process[key] = value else: params_to_process[key] = received_params.get(key) field_info = model_field.field_info assert isinstance(field_info, params.Param), "Params must be subclasses of Param" loc: tuple[str, ...] = (field_info.in_.value,) value = params_to_process if field_info.annotation is not model and not ( own_names.intersection(received_params) or (model.model_config.get("extra") in ("allow", "forbid") and value) ): value = Undefined v_, errors_ = _validate_value_with_model_field( field=model_field, value=value, values={}, loc=loc ) if header_aliases and model.model_config.get("loc_by_alias", True): # Report the header names without changing model aliases or schema. for error in errors_: error_loc = error["loc"] if len(error_loc) > 1 and error_loc[1] in header_aliases: error["loc"] = ( error_loc[0], header_aliases[error_loc[1]], *error_loc[2:], ) return v_, errors_ def request_params_to_args( fields: Sequence[ModelField], received_params: Mapping[str, Any] | QueryParams | Headers, ) -> tuple[dict[str, Any], list[Any]]: values: dict[str, Any] = {} errors: list[Any] = [] if not fields: return values, errors for field in fields: if (model := _get_param_model(field)) is not None: v_, errors_ = _request_param_model_to_args(field, model, received_params) else: aliases = _get_param_aliases(field) _, alias = next( (pair for pair in aliases if pair[1] in received_params), aliases[0] ) value = _get_multidict_value(field, received_params, alias=alias) field_info = field.field_info assert isinstance(field_info, params.Param), ( "Params must be subclasses of Param" ) loc = (field_info.in_.value, alias) v_, errors_ = _validate_value_with_model_field( field=field, value=value, values=values, loc=loc ) if errors_: errors.extend(errors_) else: values[field.name] = v_ return values, errors def _get_form_union_models(annotation: Any) -> list[type[BaseModel]]: if get_origin(annotation) not in (Union, UnionType): return [] models = [] for arg in get_args(annotation): if arg is type(None): continue if get_origin(arg) is Annotated: arg = get_args(arg)[0] if not lenient_issubclass(arg, BaseModel): return [] models.append(arg) # A nullable single model uses the regular form model extraction. return models if len(models) > 1 else [] def _add_form_union_validators(annotation: Any) -> Any: choices: list[Any] = [] for choice in get_args(annotation): if choice is type(None): choices.append(choice) continue model = choice metadata: Sequence[Any] = () if get_origin(choice) is Annotated: model, *metadata = get_args(choice) validator = BeforeValidator(partial(_normalize_form_model, model=model)) # Keep existing discriminator tags or model names in union errors. tag = next( (item for item in reversed(metadata) if isinstance(item, Tag)), Tag(model.__name__), ) choices.append(Annotated[choice, validator, tag]) return Union[tuple(choices)] # noqa: UP007 def _normalize_form_model(value: Any, *, model: type[BaseModel]) -> Any: if isinstance(value, dict): # Repeated form values must be interpreted separately by each union branch. form_data = FormData( [ (key, item) for key, items in value.items() for item in (items if isinstance(items, list) else [items]) ] ) normalized = { alias: item for alias, item, _ in _extract_form_items( get_cached_model_fields(model), form_data ) } # These cannot come from browser form entries, but a validator can supply them. for key, item in value.items(): if item is None or (isinstance(item, list) and not item): normalized[key] = item return normalized return value def _get_form_models(field: ModelField) -> list[type[BaseModel]]: if ( not isinstance(field.field_info, params.Form) or field.field_info.embed or _is_json_field(field) ): return [] model = _get_model_type(field.field_info.annotation) if model is not None: return [model] return _get_form_union_models(field.field_info.annotation) def _get_form_model(field: ModelField) -> type[BaseModel] | None: models = _get_form_models(field) return models[0] if len(models) == 1 else None def _should_embed_body_fields(fields: list[ModelField]) -> bool: if not fields: return False # More than one dependency could have the same field, it would show up as multiple # fields but it's the same one, so count them by name body_param_names_set = {field.name for field in fields} # A top level field has to be a single field, not multiple if len(body_param_names_set) > 1: return True first_field = fields[0] # Dependencies can use the same parameter name for different form models. if _get_form_models(first_field) and any( not _get_form_models(field) or field.field_info.annotation is not first_field.field_info.annotation for field in fields[1:] ): return True # If it explicitly specifies it is embedded, it has to be embedded if getattr(first_field.field_info, "embed", None): return True # If it's a Form (or File) field, it has to be a BaseModel (or a union of BaseModels) to be top level # otherwise it has to be embedded, so that the key value pair can be extracted if ( isinstance(first_field.field_info, params.Form) and not lenient_issubclass(first_field.field_info.annotation, BaseModel) and not _get_form_models(first_field) ): return True return False def _extract_form_items( body_fields: list[ModelField], received_body: FormData, ) -> Iterator[tuple[str, Any, ModelField | None]]: field_aliases: set[str] = set() for field in body_fields: for alias, _ in _get_param_aliases(field): field_aliases.add(alias) value = _get_multidict_value(field, received_body, alias=alias) if value is not Undefined: yield alias, value, field # Leave later aliases unchanged for other fields or extra validation. break for key in received_body.keys(): if key not in field_aliases: param_values = received_body.getlist(key) if len(param_values) == 1: yield key, param_values[0], None else: yield key, param_values, None async def _read_file_bytes(file: UploadFile) -> bytes: # Automatic conversion must not consume a file shared with other parameters. position = file.file.tell() try: return await file.read() finally: await file.seek(position) async def _extract_form_body( body_fields: list[ModelField], received_body: FormData, ) -> AsyncIterator[tuple[str, Any, ModelField | None]]: for alias, value, field in _extract_form_items(body_fields, received_body): if field is not None and isinstance(field.field_info, params.File): if is_bytes_or_nonable_bytes_annotation( field.field_info.annotation ) and isinstance(value, UploadFile): value = await _read_file_bytes(value) elif is_bytes_sequence_annotation( field.field_info.annotation ) and value_is_sequence(value): assert isinstance(value, sequence_types) results: list[bytes | str] = [] for sub_value in value: if isinstance(sub_value, UploadFile): results.append(await _read_file_bytes(sub_value)) else: results.append(sub_value) value = serialize_sequence_value(field=field, value=results) yield alias, value, field def _is_form_model_supplied( model: type[BaseModel], received_body: FormData, ) -> bool: # Check raw keys before empty fields are removed. own_names = { alias for item in get_cached_model_fields(model) for alias, _ in _get_param_aliases(item) } return bool( own_names.intersection(received_body) or (model.model_config.get("extra") in ("allow", "forbid") and received_body) ) async def _extract_form_model( field: ModelField, models: list[type[BaseModel]], received_body: FormData, ) -> Any: if type(None) in get_args(field.field_info.annotation) and not any( _is_form_model_supplied(model, received_body) for model in models ): return Undefined # Union validators normalize each branch from the original form values. model_fields = get_cached_model_fields(models[0]) if len(models) == 1 else [] return { alias: item async for alias, item, _ in _extract_form_body(model_fields, received_body) } async def request_body_to_args( body_fields: list[ModelField], received_body: Any, embed_body_fields: bool, ) -> tuple[dict[str, Any], list[dict[str, Any]]]: values: dict[str, Any] = {} errors: list[dict[str, Any]] = [] assert body_fields, "request_body_to_args() should be called with fields" single_not_embedded_field = ( len(body_fields) == 1 and not embed_body_fields and not _should_embed_body_fields(body_fields) and not _get_form_models(body_fields[0]) ) first_field = body_fields[0] body_to_process = received_body form_values: dict[str, tuple[str, Any]] = {} if isinstance(received_body, FormData): fields_to_extract = [ field for field in body_fields if not _get_form_models(field) ] body_to_process = {} async for alias, value, field in _extract_form_body( fields_to_extract, received_body, ): body_to_process[alias] = value if field is not None: # Keep each parameter's selected alias and scalar/list shape. form_values[field.name] = (alias, value) if single_not_embedded_field: loc: tuple[str, ...] = ("body",) v_, errors_ = _validate_value_with_model_field( field=first_field, value=body_to_process, values=values, loc=loc, ) return {first_field.name: v_}, errors_ for field in body_fields: aliases = [alias for alias, _ in _get_param_aliases(field)] loc = ("body", aliases[0]) value = Undefined if isinstance(received_body, FormData) and (models := _get_form_models(field)): # Extract each model independently from the original form values. value = await _extract_form_model(field, models, received_body) loc = ("body",) elif isinstance(received_body, FormData): alias, value = form_values.get(field.name, (aliases[0], Undefined)) loc = ("body", alias) elif body_to_process is not Undefined and not isinstance( body_to_process, bytes ): try: for alias in aliases: value = body_to_process.get(alias, Undefined) if value is not Undefined: loc = ("body", alias) break # Embedded fields require an object, not a list, scalar, or null. except AttributeError: errors.append(get_missing_field_error(loc)) continue v_, errors_ = _validate_value_with_model_field( field=field, value=value, values=values, loc=loc ) if errors_: errors.extend(errors_) else: values[field.name] = v_ return values, errors def _get_form_model_schema( model: type[BaseModel], core_schema: CoreSchema, handler: GetJsonSchemaHandler ) -> dict[str, Any]: schema = handler(core_schema) model_schema = handler.resolve_ref_schema(schema) properties = model_schema.get("properties", {}) required = model_schema.get("required", []) aliases = {} for field in get_cached_model_fields(model): validation_alias = field.field_info.validation_alias schema_name = ( validation_alias.choices[0] if isinstance(validation_alias, AliasChoices) else get_validation_alias(field) ) names = [alias for alias, _ in _get_param_aliases(field)] if ( isinstance(schema_name, str) and schema_name in properties and ( names[0] != schema_name or (len(names) > 1 and schema_name in required) ) ): aliases[schema_name] = names if not aliases: return schema # Adjust only this form's aliases, leaving shared model schemas intact. model_schema = deepcopy(model_schema) model_schema["properties"] = { aliases.get(name, [name])[0]: value for name, value in model_schema["properties"].items() } model_schema["required"] = [] for name in required: names = aliases.get(name, [name]) if len(names) == 1: model_schema["required"].append(names[0]) else: model_schema.setdefault("allOf", []).append( {"anyOf": [{"required": [name]} for name in names]} ) return model_schema def _apply_form_body_schema( body_params: list[ModelField], schema: dict[str, Any] ) -> None: parameter_schemas = schema.pop("properties") schema.pop("required", None) properties: dict[str, Any] = {} required: list[str] = [] conditions: list[dict[str, Any]] = [] for index, param in enumerate(body_params): field_schema = parameter_schemas[f"field_{index}"] models = _get_form_models(param) if not models: alias = _get_param_aliases(param)[0][0] if alias in properties: conditions.append({"properties": {alias: field_schema}}) else: properties[alias] = field_schema if param.field_info.is_required() and alias not in required: required.append(alias) continue field_schema.pop("default", None) if type(None) in get_args(param.field_info.annotation): # A flat form group is omitted through absent fields, not a null body. field_schema["anyOf"] = [ choice for choice in field_schema["anyOf"] if choice.get("type") != "null" ] if any( model.model_config.get("extra") in ("allow", "forbid") for model in models ): absent: dict[str, Any] = {"maxProperties": 0} else: names = sorted( { alias for model in models for item in get_cached_model_fields(model) for alias, _ in _get_param_aliases(item) } ) absent = ( {"not": {"anyOf": [{"required": [name]} for name in names]}} if names else {} ) if param.field_info.is_required(): conditions.append({"not": absent}) else: field_schema = {"anyOf": [absent, field_schema]} conditions.append(field_schema) if properties: schema["properties"] = properties if required: schema["required"] = required schema["allOf"] = conditions 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 all body parameters into a single field if necessary. Used to check if it's form data (with `isinstance(body_field, params.Form)`) or JSON and to generate the JSON Schema for a request body. This is **not** used to validate/parse the request body, that's done with each individual body parameter. """ if not body_params: return None first_param = body_params[0] has_nullable_form_model = any( _get_form_models(field) and type(None) in get_args(field.field_info.annotation) for field in body_params ) if not embed_body_fields and not has_nullable_form_model: return first_param model_name = "Body_" + name fields = body_params config: ConfigDict | None = None if any(_get_form_models(field) for field in body_params): fields = [] for index, field in enumerate(body_params): field_info = copy(field.field_info) if (model := _get_form_model(field)) is not None: model_annotation: Any = model model_annotation = Annotated[ model_annotation, GetPydanticSchema( get_pydantic_json_schema=partial(_get_form_model_schema, model) ), ] field_info.annotation = cast( Any, model_annotation if field_info.annotation is model else model_annotation | None, ) field_info.alias = field_info.validation_alias = f"field_{index}" field_info.serialization_alias = f"field_{index}" field_info.title = field_info.title or field.name.replace("_", " ").title() fields.append(ModelField(name=f"field_{index}", field_info=field_info)) config = {"json_schema_extra": partial(_apply_form_body_schema, body_params)} BodyModel = create_body_model(fields=fields, model_name=model_name, config=config) if not embed_body_fields: field_info = copy(first_param.field_info) field_info.annotation = BodyModel # The original form field already carries the validation metadata. field_info.metadata = [] field_info.discriminator = None return ModelField(name=first_param.name, field_info=field_info) 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 body_params): BodyFieldInfo: type[params.Body] = params.File elif any(isinstance(f.field_info, params.Form) for f in body_params): BodyFieldInfo = params.Form else: BodyFieldInfo = params.Body if BodyFieldInfo is not params.File: body_param_media_types = [ f.field_info.media_type for f in body_params if isinstance(f.field_info, BodyFieldInfo) ] if ( BodyFieldInfo is params.Form and "multipart/form-data" in body_param_media_types ): BodyFieldInfo_kwargs["media_type"] = "multipart/form-data" elif len(set(body_param_media_types)) == 1: BodyFieldInfo_kwargs["media_type"] = body_param_media_types[0] final_field = create_model_field( name="body", type_=BodyModel, alias="body", field_info=BodyFieldInfo(**BodyFieldInfo_kwargs), ) return final_field def get_validation_alias(field: ModelField) -> str: va = getattr(field, "validation_alias", None) return va or field.alias def _get_param_aliases( field: ModelField, *, convert_underscores: bool = False ) -> list[tuple[str, str]]: """Return (validation alias, request key) pairs in preference order.""" alias = get_validation_alias(field) request_alias = alias if convert_underscores and alias == field.name: request_alias = alias.replace("_", "-") validation_alias = field.field_info.validation_alias if isinstance(field.field_info, params.Path) or isinstance( validation_alias, AliasPath ): return [(alias, request_alias)] aliases = [(alias, request_alias)] if isinstance(validation_alias, AliasChoices): if any(isinstance(choice, AliasPath) for choice in validation_alias.choices): return aliases # Explicit choices are literal names, including underscores in headers. aliases = [ (choice, choice) for choice in cast(list[str], validation_alias.choices) ] config = field.config or {} if config.get("validate_by_alias") is False or config.get( "validate_by_name", config.get("populate_by_name", False) ): request_name = field.name if convert_underscores: request_name = request_name.replace("_", "-") if config.get("validate_by_alias") is False: return [(field.name, request_name)] if field.name not in [alias for alias, _ in aliases]: aliases.append((field.name, request_name)) return aliases