import re import warnings from dataclasses import is_dataclass from typing import ( TYPE_CHECKING, Any, Dict, MutableMapping, Optional, Set, Type, Union, cast, ) from weakref import WeakKeyDictionary import fastapi from fastapi._compat import ( PYDANTIC_V2, BaseConfig, ModelField, PydanticSchemaGenerationError, Undefined, UndefinedType, Validator, lenient_issubclass, ) from fastapi.datastructures import DefaultPlaceholder, DefaultType from pydantic import BaseModel, create_model from pydantic.fields import FieldInfo from typing_extensions import Literal if TYPE_CHECKING: # pragma: nocover from .routing import APIRoute # Cache for `create_cloned_field` _CLONED_TYPES_CACHE: MutableMapping[ Type[BaseModel], Type[BaseModel] ] = WeakKeyDictionary() def is_body_allowed_for_status_code(status_code: Union[int, str, None]) -> bool: if status_code is None: return True # Ref: https://github.com/OAI/OpenAPI-Specification/blob/main/versions/3.1.0.md#patterned-fields-1 if status_code in { "default", "1XX", "2XX", "3XX", "4XX", "5XX", }: return True current_status_code = int(status_code) return not (current_status_code < 200 or current_status_code in {204, 205, 304}) def get_path_param_names(path: str) -> Set[str]: return set(re.findall("{(.*?)}", path)) def create_response_field( name: str, type_: Type[Any], class_validators: Optional[Dict[str, Validator]] = None, default: Optional[Any] = Undefined, required: Union[bool, UndefinedType] = Undefined, model_config: Type[BaseConfig] = BaseConfig, field_info: Optional[FieldInfo] = None, alias: Optional[str] = None, mode: Literal["validation", "serialization"] = "validation", ) -> ModelField: """ Create a new response field. Raises if type_ is invalid. """ class_validators = class_validators or {} if PYDANTIC_V2: field_info = field_info or FieldInfo( annotation=type_, default=default, alias=alias ) else: field_info = field_info or FieldInfo() kwargs = {"name": name, "field_info": field_info} if PYDANTIC_V2: kwargs.update({"mode": mode}) else: kwargs.update( { "type_": type_, "class_validators": class_validators, "default": default, "required": required, "model_config": model_config, "alias": alias, } ) try: return ModelField(**kwargs) # type: ignore[arg-type] except (RuntimeError, PydanticSchemaGenerationError): raise fastapi.exceptions.FastAPIError( "Invalid args for response field! Hint: " f"check that {type_} is a valid Pydantic field type. " "If you are using a return type annotation that is not a valid Pydantic " "field (e.g. Union[Response, dict, None]) you can disable generating the " "response model from the type annotation with the path operation decorator " "parameter response_model=None. Read more: " "https://fastapi.tiangolo.com/tutorial/response-model/" ) from None def create_cloned_field( field: ModelField, *, cloned_types: Optional[MutableMapping[Type[BaseModel], Type[BaseModel]]] = None, ) -> ModelField: if PYDANTIC_V2: return field # cloned_types caches already cloned types to support recursive models and improve # performance by avoiding unnecessary cloning if cloned_types is None: cloned_types = _CLONED_TYPES_CACHE original_type = field.type_ if is_dataclass(original_type) and hasattr(original_type, "__pydantic_model__"): original_type = original_type.__pydantic_model__ use_type = original_type if lenient_issubclass(original_type, BaseModel): original_type = cast(Type[BaseModel], original_type) use_type = cloned_types.get(original_type) if use_type is None: use_type = create_model(original_type.__name__, __base__=original_type) cloned_types[original_type] = use_type for f in original_type.__fields__.values(): use_type.__fields__[f.name] = create_cloned_field( f, cloned_types=cloned_types ) new_field = create_response_field(name=field.name, type_=use_type) new_field.has_alias = field.has_alias # type: ignore[attr-defined] new_field.alias = field.alias # type: ignore[misc] new_field.class_validators = field.class_validators # type: ignore[attr-defined] new_field.default = field.default # type: ignore[misc] new_field.required = field.required # type: ignore[misc] new_field.model_config = field.model_config # type: ignore[attr-defined] new_field.field_info = field.field_info new_field.allow_none = field.allow_none # type: ignore[attr-defined] new_field.validate_always = field.validate_always # type: ignore[attr-defined] if field.sub_fields: # type: ignore[attr-defined] new_field.sub_fields = [ # type: ignore[attr-defined] create_cloned_field(sub_field, cloned_types=cloned_types) for sub_field in field.sub_fields # type: ignore[attr-defined] ] if field.key_field: # type: ignore[attr-defined] new_field.key_field = create_cloned_field( # type: ignore[attr-defined] field.key_field, # type: ignore[attr-defined] cloned_types=cloned_types, ) new_field.validators = field.validators # type: ignore[attr-defined] new_field.pre_validators = field.pre_validators # type: ignore[attr-defined] new_field.post_validators = field.post_validators # type: ignore[attr-defined] new_field.parse_json = field.parse_json # type: ignore[attr-defined] new_field.shape = field.shape # type: ignore[attr-defined] new_field.populate_validators() # type: ignore[attr-defined] return new_field def generate_operation_id_for_path( *, name: str, path: str, method: str ) -> str: # pragma: nocover warnings.warn( "fastapi.utils.generate_operation_id_for_path() was deprecated, " "it is not used internally, and will be removed soon", DeprecationWarning, stacklevel=2, ) operation_id = f"{name}{path}" operation_id = re.sub(r"\W", "_", operation_id) operation_id = f"{operation_id}_{method.lower()}" return operation_id def generate_unique_id(route: "APIRoute") -> str: operation_id = f"{route.name}{route.path_format}" operation_id = re.sub(r"\W", "_", operation_id) assert route.methods operation_id = f"{operation_id}_{list(route.methods)[0].lower()}" return operation_id def deep_dict_update(main_dict: Dict[Any, Any], update_dict: Dict[Any, Any]) -> None: for key, value in update_dict.items(): if ( key in main_dict and isinstance(main_dict[key], dict) and isinstance(value, dict) ): deep_dict_update(main_dict[key], value) elif ( key in main_dict and isinstance(main_dict[key], list) and isinstance(update_dict[key], list) ): main_dict[key] = main_dict[key] + update_dict[key] else: main_dict[key] = value def get_value_or_default( first_item: Union[DefaultPlaceholder, DefaultType], *extra_items: Union[DefaultPlaceholder, DefaultType], ) -> Union[DefaultPlaceholder, DefaultType]: """ Pass items or `DefaultPlaceholder`s by descending priority. The first one to _not_ be a `DefaultPlaceholder` will be returned. Otherwise, the first item (a `DefaultPlaceholder`) will be returned. """ items = (first_item,) + extra_items for item in items: if not isinstance(item, DefaultPlaceholder): return item return first_item