refactor(openapi): add from __future__ import annotations (#4990)

* refactor(openapi): add from __future__ import annotations

and update code according to ruff rules TCH, UP006, UP007, UP037 and
FA100.

* Fix type alias with Python 3.8

See https://bugs.python.org/issue45117

* Fix pydantic not working with Python 3.8

    TypeError: You have a type annotation 'str | None' which makes use
    of newer typing features than are supported in your version of Python.
    To handle this error, you should either remove the use of new syntax
    or install the `eval_type_backport` package.

* Removing pydantic v1 reference

---------

Co-authored-by: Leandro Damascena <[email protected]>

Author: ericbn

aws_lambda_powertools/event_handler/openapi/compat.py

@@ -1,5 +1,7 @@
 # mypy: ignore-errors
 # flake8: noqa
+from __future__ import annotations
+
 from collections import deque
 from copy import copy
 
@@ -8,7 +10,7 @@
 
 from dataclasses import dataclass, is_dataclass
 from enum import Enum
-from typing import Any, Dict, List, Set, Tuple, Type, Union, FrozenSet, Deque, Sequence, Mapping
+from typing import Any, Deque, FrozenSet, List, Mapping, Sequence, Set, Tuple, Union
 
 from typing_extensions import Annotated, Literal, get_origin, get_args
 
@@ -56,7 +58,7 @@
 
 sequence_types = tuple(sequence_annotation_to_type.keys())
 
-RequestErrorModel: Type[BaseModel] = create_model("Request")
+RequestErrorModel: type[BaseModel] = create_model("Request")
 
 
 class ErrorWrapper(Exception):
@@ -101,8 +103,8 @@ def serialize(
         value: Any,
         *,
         mode: Literal["json", "python"] = "json",
-        include: Union[IncEx, None] = None,
-        exclude: Union[IncEx, None] = None,
+        include: IncEx | None = None,
+        exclude: IncEx | None = None,
         by_alias: bool = True,
         exclude_unset: bool = False,
         exclude_defaults: bool = False,
@@ -120,8 +122,8 @@ def serialize(
         )
 
     def validate(
-        self, value: Any, values: Dict[str, Any] = {}, *, loc: Tuple[Union[int, str], ...] = ()
-    ) -> Tuple[Any, Union[List[Dict[str, Any]], None]]:
+        self, value: Any, values: dict[str, Any] = {}, *, loc: tuple[int | str, ...] = ()
+    ) -> tuple[Any, list[dict[str, Any]] | None]:
         try:
             return (self._type_adapter.validate_python(value, from_attributes=True), None)
         except ValidationError as exc:
@@ -136,11 +138,11 @@ def get_schema_from_model_field(
     *,
     field: ModelField,
     model_name_map: ModelNameMap,
-    field_mapping: Dict[
-        Tuple[ModelField, Literal["validation", "serialization"]],
+    field_mapping: dict[
+        tuple[ModelField, Literal["validation", "serialization"]],
         JsonSchemaValue,
     ],
-) -> Dict[str, Any]:
+) -> dict[str, Any]:
     json_schema = field_mapping[(field, field.mode)]
     if "$ref" not in json_schema:
         # MAINTENANCE: remove when deprecating Pydantic v1
@@ -151,39 +153,39 @@ def get_schema_from_model_field(
 
 def get_definitions(
     *,
-    fields: List[ModelField],
+    fields: list[ModelField],
     schema_generator: GenerateJsonSchema,
     model_name_map: ModelNameMap,
-) -> Tuple[
-    Dict[
-        Tuple[ModelField, Literal["validation", "serialization"]],
-        Dict[str, Any],
+) -> tuple[
+    dict[
+        tuple[ModelField, Literal["validation", "serialization"]],
+        dict[str, Any],
     ],
-    Dict[str, Dict[str, Any]],
+    dict[str, dict[str, Any]],
 ]:
     inputs = [(field, field.mode, field._type_adapter.core_schema) for field in fields]
     field_mapping, definitions = schema_generator.generate_definitions(inputs=inputs)
 
     return field_mapping, definitions
 
 
-def get_compat_model_name_map(fields: List[ModelField]) -> ModelNameMap:
+def get_compat_model_name_map(fields: list[ModelField]) -> ModelNameMap:
     return {}
 
 
 def get_annotation_from_field_info(annotation: Any, field_info: FieldInfo, field_name: str) -> Any:
     return annotation
 
 
-def model_rebuild(model: Type[BaseModel]) -> None:
+def model_rebuild(model: type[BaseModel]) -> None:
     model.model_rebuild()
 
 
 def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo:
     return type(field_info).from_annotation(annotation)
 
 
-def get_missing_field_error(loc: Tuple[str, ...]) -> Dict[str, Any]:
+def get_missing_field_error(loc: tuple[str, ...]) -> dict[str, Any]:
     error = ValidationError.from_exception_data(
         "Field required", [{"type": "missing", "loc": loc, "input": {}}]
     ).errors()[0]
@@ -220,13 +222,13 @@ def serialize_sequence_value(*, field: ModelField, value: Any) -> Sequence[Any]:
     return sequence_annotation_to_type[origin_type](value)  # type: ignore[no-any-return]
 
 
-def _normalize_errors(errors: Sequence[Any]) -> List[Dict[str, Any]]:
+def _normalize_errors(errors: Sequence[Any]) -> list[dict[str, Any]]:
     return errors  # type: ignore[return-value]
 
 
-def create_body_model(*, fields: Sequence[ModelField], model_name: str) -> Type[BaseModel]:
+def create_body_model(*, fields: Sequence[ModelField], model_name: str) -> type[BaseModel]:
     field_params = {f.name: (f.field_info.annotation, f.field_info) for f in fields}
-    model: Type[BaseModel] = create_model(model_name, **field_params)
+    model: type[BaseModel] = create_model(model_name, **field_params)
     return model
 
 
@@ -241,7 +243,7 @@ def model_json(model: BaseModel, **kwargs: Any) -> Any:
 # Common code for both versions
 
 
-def field_annotation_is_complex(annotation: Union[Type[Any], None]) -> bool:
+def field_annotation_is_complex(annotation: type[Any] | None) -> bool:
     origin = get_origin(annotation)
     if origin is Union or origin is UnionType:
         return any(field_annotation_is_complex(arg) for arg in get_args(annotation))
@@ -258,11 +260,11 @@ def field_annotation_is_scalar(annotation: Any) -> bool:
     return annotation is Ellipsis or not field_annotation_is_complex(annotation)
 
 
-def field_annotation_is_sequence(annotation: Union[Type[Any], None]) -> bool:
+def field_annotation_is_sequence(annotation: type[Any] | None) -> bool:
     return _annotation_is_sequence(annotation) or _annotation_is_sequence(get_origin(annotation))
 
 
-def field_annotation_is_scalar_sequence(annotation: Union[Type[Any], None]) -> bool:
+def field_annotation_is_scalar_sequence(annotation: type[Any] | None) -> bool:
     origin = get_origin(annotation)
     if origin is Union or origin is UnionType:
         at_least_one_scalar_sequence = False
@@ -307,24 +309,22 @@ def value_is_sequence(value: Any) -> bool:
     return isinstance(value, sequence_types) and not isinstance(value, (str, bytes))  # type: ignore[arg-type]
 
 
-def _annotation_is_complex(annotation: Union[Type[Any], None]) -> bool:
+def _annotation_is_complex(annotation: type[Any] | None) -> bool:
     return (
         lenient_issubclass(annotation, (BaseModel, Mapping))  # TODO: UploadFile
         or _annotation_is_sequence(annotation)
         or is_dataclass(annotation)
     )
 
 
-def _annotation_is_sequence(annotation: Union[Type[Any], None]) -> bool:
+def _annotation_is_sequence(annotation: type[Any] | None) -> bool:
     if lenient_issubclass(annotation, (str, bytes)):
         return False
     return lenient_issubclass(annotation, sequence_types)
 
 
-def _regenerate_error_with_loc(
-    *, errors: Sequence[Any], loc_prefix: Tuple[Union[str, int], ...]
-) -> List[Dict[str, Any]]:
-    updated_loc_errors: List[Any] = [
+def _regenerate_error_with_loc(*, errors: Sequence[Any], loc_prefix: tuple[str | int, ...]) -> list[dict[str, Any]]:
+    updated_loc_errors: list[Any] = [
         {**err, "loc": loc_prefix + err.get("loc", ())} for err in _normalize_errors(errors)
     ]
 

aws_lambda_powertools/event_handler/openapi/dependant.py

@@ -1,8 +1,8 @@
+from __future__ import annotations
+
 import inspect
 import re
-from typing import Any, Callable, Dict, ForwardRef, List, Optional, Set, Tuple, Type, cast
-
-from pydantic import BaseModel
+from typing import TYPE_CHECKING, Any, Callable, ForwardRef, cast
 
 from aws_lambda_powertools.event_handler.openapi.compat import (
     ModelField,
@@ -26,6 +26,9 @@
 )
 from aws_lambda_powertools.event_handler.openapi.types import OpenAPIResponse, OpenAPIResponseContentModel
 
+if TYPE_CHECKING:
+    from pydantic import BaseModel
+
 """
 This turns the opaque function signature into typed, validated models.
 
@@ -76,7 +79,7 @@ def add_param_to_fields(
         raise AssertionError(f"Unsupported param type: {field_info.in_}")
 
 
-def get_typed_annotation(annotation: Any, globalns: Dict[str, Any]) -> Any:
+def get_typed_annotation(annotation: Any, globalns: dict[str, Any]) -> Any:
     """
     Evaluates a type annotation, which can be a string or a ForwardRef.
     """
@@ -128,7 +131,7 @@ def get_typed_signature(call: Callable[..., Any]) -> inspect.Signature:
         return inspect.Signature(typed_params)
 
 
-def get_path_param_names(path: str) -> Set[str]:
+def get_path_param_names(path: str) -> set[str]:
     """
     Returns the path parameter names from a path template. Those are the strings between { and }.
 
@@ -139,7 +142,7 @@ def get_path_param_names(path: str) -> Set[str]:
 
     Returns
     -------
-    Set[str]
+    set[str]
         The path parameter names
 
     """
@@ -150,8 +153,8 @@ def get_dependant(
     *,
     path: str,
     call: Callable[..., Any],
-    name: Optional[str] = None,
-    responses: Optional[Dict[int, OpenAPIResponse]] = None,
+    name: str | None = None,
+    responses: dict[int, OpenAPIResponse] | None = None,
 ) -> Dependant:
     """
     Returns a dependant model for a handler function. A dependant model is a model that contains
@@ -165,7 +168,7 @@ def get_dependant(
         The handler function
     name: str, optional
         The name of the handler function
-    responses: List[Dict[int, OpenAPIResponse]], optional
+    responses: list[dict[int, OpenAPIResponse]], optional
         The list of extra responses for the handler function
 
     Returns
@@ -210,7 +213,7 @@ def get_dependant(
     return dependant
 
 
-def _add_extra_responses(dependant: Dependant, responses: Optional[Dict[int, OpenAPIResponse]]):
+def _add_extra_responses(dependant: Dependant, responses: dict[int, OpenAPIResponse] | None):
     # Also add the optional extra responses to the dependant model.
     if not responses:
         return
@@ -278,7 +281,7 @@ def is_body_param(*, param_field: ModelField, is_path_param: bool) -> bool:
         return True
 
 
-def get_flat_params(dependant: Dependant) -> List[ModelField]:
+def get_flat_params(dependant: Dependant) -> list[ModelField]:
     """
     Get a list of all the parameters from a Dependant object.
 
@@ -289,7 +292,7 @@ def get_flat_params(dependant: Dependant) -> List[ModelField]:
 
     Returns
     -------
-    List[ModelField]
+    list[ModelField]
         A list of ModelField objects containing the flat parameters from the Dependant object.
 
     """
@@ -302,7 +305,7 @@ def get_flat_params(dependant: Dependant) -> List[ModelField]:
     )
 
 
-def get_body_field(*, dependant: Dependant, name: str) -> Optional[ModelField]:
+def get_body_field(*, dependant: Dependant, name: str) -> ModelField | None:
     """
     Get the Body field for a given Dependant object.
     """
@@ -348,24 +351,24 @@ def get_body_field(*, dependant: Dependant, name: str) -> Optional[ModelField]:
 
 def get_body_field_info(
     *,
-    body_model: Type[BaseModel],
+    body_model: type[BaseModel],
     flat_dependant: Dependant,
     required: bool,
-) -> Tuple[Type[Body], Dict[str, Any]]:
+) -> tuple[type[Body], dict[str, Any]]:
     """
     Get the Body field info and kwargs for a given body model.
     """
 
-    body_field_info_kwargs: Dict[str, Any] = {"annotation": body_model, "alias": "body"}
+    body_field_info_kwargs: dict[str, Any] = {"annotation": body_model, "alias": "body"}
 
     if not required:
         body_field_info_kwargs["default"] = None
 
     if any(isinstance(f.field_info, _File) for f in flat_dependant.body_params):
-        # MAINTENANCE: body_field_info: Type[Body] = _File
+        # MAINTENANCE: body_field_info: type[Body] = _File
         raise NotImplementedError("_File fields are not supported in request bodies")
     elif any(isinstance(f.field_info, _Form) for f in flat_dependant.body_params):
-        # MAINTENANCE: body_field_info: Type[Body] = _Form
+        # MAINTENANCE: body_field_info: type[Body] = _Form
         raise NotImplementedError("_Form fields are not supported in request bodies")
     else:
         body_field_info = Body