utils.py 4.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136
  1. import re
  2. import warnings
  3. from typing import (
  4. TYPE_CHECKING,
  5. Any,
  6. Literal,
  7. )
  8. import fastapi
  9. from fastapi._compat import (
  10. ModelField,
  11. PydanticSchemaGenerationError,
  12. Undefined,
  13. annotation_is_pydantic_v1,
  14. )
  15. from fastapi.datastructures import DefaultPlaceholder, DefaultType
  16. from fastapi.exceptions import FastAPIDeprecationWarning, PydanticV1NotSupportedError
  17. from pydantic.fields import FieldInfo
  18. from ._compat import v2
  19. if TYPE_CHECKING: # pragma: nocover
  20. from .routing import APIRoute
  21. def is_body_allowed_for_status_code(status_code: int | str | None) -> bool:
  22. if status_code is None:
  23. return True
  24. # Ref: https://github.com/OAI/OpenAPI-Specification/blob/main/versions/3.1.0.md#patterned-fields-1
  25. if status_code in {
  26. "default",
  27. "1XX",
  28. "2XX",
  29. "3XX",
  30. "4XX",
  31. "5XX",
  32. }:
  33. return True
  34. current_status_code = int(status_code)
  35. return not (current_status_code < 200 or current_status_code in {204, 205, 304})
  36. def get_path_param_names(path: str) -> set[str]:
  37. return set(re.findall("{(.*?)}", path))
  38. _invalid_args_message = (
  39. "Invalid args for response field! Hint: "
  40. "check that {type_} is a valid Pydantic field type. "
  41. "If you are using a return type annotation that is not a valid Pydantic "
  42. "field (e.g. Union[Response, dict, None]) you can disable generating the "
  43. "response model from the type annotation with the path operation decorator "
  44. "parameter response_model=None. Read more: "
  45. "https://fastapi.tiangolo.com/tutorial/response-model/"
  46. )
  47. def create_model_field(
  48. name: str,
  49. type_: Any,
  50. default: Any | None = Undefined,
  51. field_info: FieldInfo | None = None,
  52. alias: str | None = None,
  53. mode: Literal["validation", "serialization"] = "validation",
  54. ) -> ModelField:
  55. if annotation_is_pydantic_v1(type_):
  56. raise PydanticV1NotSupportedError(
  57. "pydantic.v1 models are no longer supported by FastAPI."
  58. f" Please update the response model {type_!r}."
  59. )
  60. field_info = field_info or FieldInfo(annotation=type_, default=default, alias=alias)
  61. try:
  62. return v2.ModelField(mode=mode, name=name, field_info=field_info)
  63. except PydanticSchemaGenerationError:
  64. raise fastapi.exceptions.FastAPIError(
  65. _invalid_args_message.format(type_=type_)
  66. ) from None
  67. def generate_operation_id_for_path(
  68. *, name: str, path: str, method: str
  69. ) -> str: # pragma: nocover
  70. warnings.warn(
  71. message="fastapi.utils.generate_operation_id_for_path() was deprecated, "
  72. "it is not used internally, and will be removed soon",
  73. category=FastAPIDeprecationWarning,
  74. stacklevel=2,
  75. )
  76. operation_id = f"{name}{path}"
  77. operation_id = re.sub(r"\W", "_", operation_id)
  78. operation_id = f"{operation_id}_{method.lower()}"
  79. return operation_id
  80. def generate_unique_id(route: "APIRoute") -> str:
  81. operation_id = f"{route.name}{route.path_format}"
  82. operation_id = re.sub(r"\W", "_", operation_id)
  83. assert route.methods
  84. operation_id = f"{operation_id}_{list(route.methods)[0].lower()}"
  85. return operation_id
  86. def deep_dict_update(main_dict: dict[Any, Any], update_dict: dict[Any, Any]) -> None:
  87. for key, value in update_dict.items():
  88. if (
  89. key in main_dict
  90. and isinstance(main_dict[key], dict)
  91. and isinstance(value, dict)
  92. ):
  93. deep_dict_update(main_dict[key], value)
  94. elif (
  95. key in main_dict
  96. and isinstance(main_dict[key], list)
  97. and isinstance(update_dict[key], list)
  98. ):
  99. main_dict[key] = main_dict[key] + update_dict[key]
  100. else:
  101. main_dict[key] = value
  102. def get_value_or_default(
  103. first_item: DefaultPlaceholder | DefaultType,
  104. *extra_items: DefaultPlaceholder | DefaultType,
  105. ) -> DefaultPlaceholder | DefaultType:
  106. """
  107. Pass items or `DefaultPlaceholder`s by descending priority.
  108. The first one to _not_ be a `DefaultPlaceholder` will be returned.
  109. Otherwise, the first item (a `DefaultPlaceholder`) will be returned.
  110. """
  111. items = (first_item,) + extra_items
  112. for item in items:
  113. if not isinstance(item, DefaultPlaceholder):
  114. return item
  115. return first_item