refactor(event_handler): use standard collections for types + refactor code

aws_lambda_powertools/event_handler/api_gateway.py

@@ -12,7 +12,7 @@
 from functools import partial
 from http import HTTPStatus
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Callable, Generic, Literal, Mapping, Match, Pattern, Sequence, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Generic, Literal, Match, Pattern, TypeVar, cast
 
 from typing_extensions import override
 
@@ -59,6 +59,9 @@
 )
 from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent
 
+if TYPE_CHECKING:
+    from collections.abc import Callable, Mapping, Sequence
+
 logger = logging.getLogger(__name__)
 
 _DYNAMIC_ROUTE_PATTERN = r"(<\w+>)"
@@ -68,6 +71,7 @@
 _NAMED_GROUP_BOUNDARY_PATTERN = rf"(?P\1[{_SAFE_URI}{_UNSAFE_URI}\\w]+)"
 _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION = "Successful Response"
 _ROUTE_REGEX = "^{}$"
+_JSON_DUMP_CALL = partial(json.dumps, separators=(",", ":"), cls=Encoder)
 
 ResponseEventT = TypeVar("ResponseEventT", bound=BaseProxyEvent)
 ResponseT = TypeVar("ResponseT")
@@ -830,7 +834,7 @@ class ResponseBuilder(Generic[ResponseEventT]):
     def __init__(
         self,
         response: Response,
-        serializer: Callable[[Any], str] = partial(json.dumps, separators=(",", ":"), cls=Encoder),
+        serializer: Callable[[Any], str] = _JSON_DUMP_CALL,
         route: Route | None = None,
     ):
         self.response = response
@@ -1723,8 +1727,9 @@ def get_openapi_schema(
         security = security or self.openapi_config.security
         openapi_extensions = openapi_extensions or self.openapi_config.openapi_extensions
 
+        from pydantic.json_schema import GenerateJsonSchema
+
         from aws_lambda_powertools.event_handler.openapi.compat import (
-            GenerateJsonSchema,
             get_compat_model_name_map,
             get_definitions,
         )

aws_lambda_powertools/event_handler/appsync.py

@@ -3,13 +3,15 @@
 import asyncio
 import logging
 import warnings
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any
 
 from aws_lambda_powertools.event_handler.graphql_appsync.exceptions import InvalidBatchResponse, ResolverNotFoundError
 from aws_lambda_powertools.event_handler.graphql_appsync.router import Router
 from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from aws_lambda_powertools.utilities.typing import LambdaContext
 
 from aws_lambda_powertools.warnings import PowertoolsUserWarning

aws_lambda_powertools/event_handler/bedrock_agent.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import json
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any
 
 from typing_extensions import override
 
@@ -14,6 +14,7 @@
 from aws_lambda_powertools.event_handler.openapi.constants import DEFAULT_API_VERSION, DEFAULT_OPENAPI_VERSION
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     from http import HTTPStatus
     from re import Match
 

aws_lambda_powertools/event_handler/graphql_appsync/_registry.py

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
 import logging
-from typing import Any, Callable
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
 
 logger = logging.getLogger(__name__)
 

aws_lambda_powertools/event_handler/graphql_appsync/base.py

@@ -1,7 +1,10 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import Callable
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from collections.abc import Callable
 
 
 class BaseRouter(ABC):

aws_lambda_powertools/event_handler/graphql_appsync/router.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Callable
+from typing import TYPE_CHECKING
 
 from aws_lambda_powertools.event_handler.graphql_appsync._registry import ResolverRegistry
 from aws_lambda_powertools.event_handler.graphql_appsync.base import BaseRouter
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from aws_lambda_powertools.utilities.data_classes.appsync_resolver_event import AppSyncResolverEvent
     from aws_lambda_powertools.utilities.typing.lambda_context import LambdaContext
 

aws_lambda_powertools/event_handler/lambda_function_url.py

@@ -1,13 +1,14 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Callable, Pattern
+from typing import TYPE_CHECKING, Pattern
 
 from aws_lambda_powertools.event_handler.api_gateway import (
     ApiGatewayResolver,
     ProxyEventType,
 )
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     from http import HTTPStatus
 
     from aws_lambda_powertools.event_handler import CORSConfig

aws_lambda_powertools/event_handler/middlewares/openapi_validation.py

@@ -365,7 +365,7 @@ def _validate_field(
     """
     Validate a field, and append any errors to the existing_errors list.
     """
-    validated_value, errors = field.validate(value, value, loc=loc)
+    validated_value, errors = field.validate(value=value, loc=loc)
 
     if isinstance(errors, list):
         processed_errors = _regenerate_error_with_loc(errors=errors, loc_prefix=())

aws_lambda_powertools/event_handler/openapi/compat.py

@@ -1,36 +1,30 @@
 # mypy: ignore-errors
-# flake8: noqa
 from __future__ import annotations
 
 from collections import deque
-from copy import copy
+from collections.abc import Mapping, Sequence
 
 # MAINTENANCE: remove when deprecating Pydantic v1. Mypy doesn't handle two different code paths that import different
 # versions of a module, so we need to ignore errors here.
-
 from dataclasses import dataclass, is_dataclass
-from enum import Enum
-from typing import TYPE_CHECKING, Any, Deque, FrozenSet, List, Mapping, Sequence, Set, Tuple, Union
-
-from typing_extensions import Annotated, Literal, get_origin, get_args
-
-from pydantic import BaseModel, create_model
-from pydantic.fields import FieldInfo
+from typing import TYPE_CHECKING, Any, Deque, FrozenSet, List, Set, Tuple, Union
 
-from aws_lambda_powertools.event_handler.openapi.types import COMPONENT_REF_PREFIX, UnionType
-
-from pydantic import TypeAdapter, ValidationError
+from pydantic import BaseModel, TypeAdapter, ValidationError, create_model
 
 # Importing from internal libraries in Pydantic may introduce potential risks, as these internal libraries
 # are not part of the public API and may change without notice in future releases.
 # We use this for forward reference, as it allows us to handle forward references in type annotations.
 from pydantic._internal._typing_extra import eval_type_lenient
-from pydantic.fields import FieldInfo
 from pydantic._internal._utils import lenient_issubclass
-from pydantic.json_schema import GenerateJsonSchema, JsonSchemaValue
 from pydantic_core import PydanticUndefined, PydanticUndefinedType
+from typing_extensions import Annotated, Literal, get_args, get_origin
+
+from aws_lambda_powertools.event_handler.openapi.types import UnionType
 
 if TYPE_CHECKING:
+    from pydantic.fields import FieldInfo
+    from pydantic.json_schema import GenerateJsonSchema, JsonSchemaValue
+
     from aws_lambda_powertools.event_handler.openapi.types import IncEx, ModelNameMap
 
 Undefined = PydanticUndefined
@@ -119,7 +113,10 @@ def serialize(
         )
 
     def validate(
-        self, value: Any, values: dict[str, Any] = {}, *, loc: tuple[int | str, ...] = ()
+        self,
+        value: Any,
+        *,
+        loc: tuple[int | str, ...] = (),
     ) -> tuple[Any, list[dict[str, Any]] | None]:
         try:
             return (self._type_adapter.validate_python(value, from_attributes=True), None)
@@ -184,7 +181,8 @@ def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo:
 
 def get_missing_field_error(loc: tuple[str, ...]) -> dict[str, Any]:
     error = ValidationError.from_exception_data(
-        "Field required", [{"type": "missing", "loc": loc, "input": {}}]
+        "Field required",
+        [{"type": "missing", "loc": loc, "input": {}}],
     ).errors()[0]
     error["input"] = None
     return error
@@ -308,7 +306,7 @@ def value_is_sequence(value: Any) -> bool:
 
 def _annotation_is_complex(annotation: type[Any] | None) -> bool:
     return (
-        lenient_issubclass(annotation, (BaseModel, Mapping))  # TODO: UploadFile
+        lenient_issubclass(annotation, (BaseModel, Mapping))  # Keep it to UploadFile
         or _annotation_is_sequence(annotation)
         or is_dataclass(annotation)
     )

aws_lambda_powertools/event_handler/openapi/dependant.py

@@ -2,7 +2,7 @@
 
 import inspect
 import re
-from typing import TYPE_CHECKING, Any, Callable, ForwardRef, cast
+from typing import TYPE_CHECKING, Any, ForwardRef, cast
 
 from aws_lambda_powertools.event_handler.openapi.compat import (
     ModelField,
@@ -27,6 +27,8 @@
 from aws_lambda_powertools.event_handler.openapi.types import OpenAPIResponse, OpenAPIResponseContentModel
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from pydantic import BaseModel
 
 """

aws_lambda_powertools/event_handler/openapi/encoders.py

@@ -8,7 +8,7 @@
 from pathlib import Path, PurePath
 from re import Pattern
 from types import GeneratorType
-from typing import TYPE_CHECKING, Any, Callable
+from typing import TYPE_CHECKING, Any
 from uuid import UUID
 
 from pydantic import BaseModel
@@ -17,6 +17,8 @@
 from aws_lambda_powertools.event_handler.openapi.compat import _model_dump
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from aws_lambda_powertools.event_handler.openapi.types import IncEx
 
 from aws_lambda_powertools.event_handler.openapi.exceptions import SerializationError

aws_lambda_powertools/event_handler/openapi/exceptions.py

@@ -1,4 +1,5 @@
-from typing import Any, Literal, Sequence
+from collections.abc import Sequence
+from typing import Any, Literal
 
 
 class ValidationException(Exception):

aws_lambda_powertools/event_handler/openapi/params.py

@@ -2,7 +2,7 @@
 
 import inspect
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Callable, Literal
+from typing import TYPE_CHECKING, Any, Literal
 
 from pydantic import BaseConfig
 from pydantic.fields import FieldInfo
@@ -20,6 +20,8 @@
 )
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
+
     from aws_lambda_powertools.event_handler.openapi.models import Example
     from aws_lambda_powertools.event_handler.openapi.types import CacheKey
 

aws_lambda_powertools/event_handler/openapi/types.py

@@ -1,9 +1,10 @@
 from __future__ import annotations
 
 import types
-from typing import TYPE_CHECKING, Any, Callable, Dict, Set, Type, TypedDict, Union
+from typing import TYPE_CHECKING, Any, Dict, Set, Type, TypedDict, Union
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     from enum import Enum
 
     from pydantic import BaseModel

aws_lambda_powertools/event_handler/util.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
-from typing import Any, Dict, List, Mapping
+from typing import TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from collections.abc import Mapping
 
 
 class _FrozenDict(dict):
@@ -18,7 +21,7 @@ def __hash__(self):
         return hash(frozenset(self.keys()))
 
 
-class _FrozenListDict(List[Dict[str, List[str]]]):
+class _FrozenListDict(list[dict[str, list[str]]]):
     """
     Freezes a list of dictionaries containing lists of strings.
 

aws_lambda_powertools/event_handler/vpc_lattice.py

@@ -1,13 +1,14 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Callable, Pattern
+from typing import TYPE_CHECKING, Pattern
 
 from aws_lambda_powertools.event_handler.api_gateway import (
     ApiGatewayResolver,
     ProxyEventType,
 )
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     from http import HTTPStatus
 
     from aws_lambda_powertools.event_handler import CORSConfig

tests/functional/event_handler/_pydantic/conftest.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 
 import fastjsonschema

tests/functional/event_handler/_pydantic/test_api_gateway.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from pydantic import BaseModel
 
 from aws_lambda_powertools.event_handler import content_types

tests/functional/event_handler/_pydantic/test_openapi_config.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 
 from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tests/functional/event_handler/_pydantic/test_openapi_encoders.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import math
 from collections import deque
 from dataclasses import dataclass

tests/functional/event_handler/_pydantic/test_openapi_extensions.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 
 from aws_lambda_powertools.event_handler.api_gateway import APIGatewayRestResolver, Router

tests/functional/event_handler/_pydantic/test_openapi_params.py

@@ -278,7 +278,7 @@ class User(BaseModel):
 
     @app.get("/")
     def handler() -> User:
-        return User(name="Ruben Fonseca")
+        return User(name="Powertools")
 
     schema = app.get_openapi_schema()
     assert len(schema.paths.keys()) == 1

tests/functional/event_handler/_pydantic/test_openapi_security.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 from aws_lambda_powertools.event_handler import APIGatewayRestResolver

tests/functional/event_handler/_pydantic/test_openapi_security_schemes.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.event_handler import APIGatewayRestResolver
 from aws_lambda_powertools.event_handler.openapi.models import (
     APIKey,

tests/functional/event_handler/_pydantic/test_openapi_servers.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from aws_lambda_powertools.event_handler.api_gateway import APIGatewayRestResolver
 from aws_lambda_powertools.event_handler.openapi.models import Server