Merge branch 'v3' into parser_annotations

Author: leandrodamascena

aws_lambda_powertools/event_handler/api_gateway.py

@@ -1,8 +1,12 @@
+from __future__ import annotations
+
 import logging
-from typing import Any, Callable, Optional, Type, TypeVar
+from typing import TYPE_CHECKING, Any, Callable, TypeVar
 
 from aws_lambda_powertools.utilities.data_classes import AppSyncResolverEvent
-from aws_lambda_powertools.utilities.typing import LambdaContext
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.utilities.typing import LambdaContext
 
 logger = logging.getLogger(__name__)
 
@@ -17,7 +21,7 @@ class BaseRouter:
     def __init__(self):
         self._resolvers: dict = {}
 
-    def resolver(self, type_name: str = "*", field_name: Optional[str] = None):
+    def resolver(self, type_name: str = "*", field_name: str | None = None):
         """Registers the resolver for field_name
 
         Parameters
@@ -83,7 +87,7 @@ def resolve(
         self,
         event: dict,
         context: LambdaContext,
-        data_model: Type[AppSyncResolverEvent] = AppSyncResolverEvent,
+        data_model: type[AppSyncResolverEvent] = AppSyncResolverEvent,
     ) -> Any:
         """Resolve field_name
 
@@ -189,12 +193,12 @@ def __call__(
         self,
         event: dict,
         context: LambdaContext,
-        data_model: Type[AppSyncResolverEvent] = AppSyncResolverEvent,
+        data_model: type[AppSyncResolverEvent] = AppSyncResolverEvent,
     ) -> Any:
         """Implicit lambda handler which internally calls `resolve`"""
         return self.resolve(event, context, data_model)
 
-    def include_router(self, router: "Router") -> None:
+    def include_router(self, router: Router) -> None:
         """Adds all resolvers defined in a router
 
         Parameters

aws_lambda_powertools/event_handler/appsync.py

@@ -1,5 +1,6 @@
-from re import Match
-from typing import Any, Callable, Dict, List, Optional
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Callable
 
 from typing_extensions import override
 
@@ -9,8 +10,12 @@
     ProxyEventType,
     ResponseBuilder,
 )
-from aws_lambda_powertools.event_handler.openapi.types import OpenAPIResponse
-from aws_lambda_powertools.utilities.data_classes import BedrockAgentEvent
+
+if TYPE_CHECKING:
+    from re import Match
+
+    from aws_lambda_powertools.event_handler.openapi.types import OpenAPIResponse
+    from aws_lambda_powertools.utilities.data_classes import BedrockAgentEvent
 
 
 class BedrockResponseBuilder(ResponseBuilder):
@@ -21,7 +26,7 @@ class BedrockResponseBuilder(ResponseBuilder):
     """
 
     @override
-    def build(self, event: BedrockAgentEvent, *args) -> Dict[str, Any]:
+    def build(self, event: BedrockAgentEvent, *args) -> dict[str, Any]:
         """Build the full response dict to be returned by the lambda"""
         self._route(event, None)
 
@@ -91,16 +96,16 @@ def get(  # type: ignore[override]
         self,
         rule: str,
         description: str,
-        cors: Optional[bool] = None,
+        cors: bool | None = None,
         compress: bool = False,
-        cache_control: Optional[str] = None,
-        summary: Optional[str] = None,
-        responses: Optional[Dict[int, OpenAPIResponse]] = None,
+        cache_control: str | None = None,
+        summary: str | None = None,
+        responses: dict[int, OpenAPIResponse] | None = None,
         response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
-        tags: Optional[List[str]] = None,
-        operation_id: Optional[str] = None,
+        tags: list[str] | None = None,
+        operation_id: str | None = None,
         include_in_schema: bool = True,
-        middlewares: Optional[List[Callable[..., Any]]] = None,
+        middlewares: list[Callable[..., Any]] | None = None,
     ) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
         security = None
 
@@ -126,16 +131,16 @@ def post(  # type: ignore[override]
         self,
         rule: str,
         description: str,
-        cors: Optional[bool] = None,
+        cors: bool | None = None,
         compress: bool = False,
-        cache_control: Optional[str] = None,
-        summary: Optional[str] = None,
-        responses: Optional[Dict[int, OpenAPIResponse]] = None,
+        cache_control: str | None = None,
+        summary: str | None = None,
+        responses: dict[int, OpenAPIResponse] | None = None,
         response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
-        tags: Optional[List[str]] = None,
-        operation_id: Optional[str] = None,
+        tags: list[str] | None = None,
+        operation_id: str | None = None,
         include_in_schema: bool = True,
-        middlewares: Optional[List[Callable[..., Any]]] = None,
+        middlewares: list[Callable[..., Any]] | None = None,
     ):
         security = None
 
@@ -161,16 +166,16 @@ def put(  # type: ignore[override]
         self,
         rule: str,
         description: str,
-        cors: Optional[bool] = None,
+        cors: bool | None = None,
         compress: bool = False,
-        cache_control: Optional[str] = None,
-        summary: Optional[str] = None,
-        responses: Optional[Dict[int, OpenAPIResponse]] = None,
+        cache_control: str | None = None,
+        summary: str | None = None,
+        responses: dict[int, OpenAPIResponse] | None = None,
         response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
-        tags: Optional[List[str]] = None,
-        operation_id: Optional[str] = None,
+        tags: list[str] | None = None,
+        operation_id: str | None = None,
         include_in_schema: bool = True,
-        middlewares: Optional[List[Callable[..., Any]]] = None,
+        middlewares: list[Callable[..., Any]] | None = None,
     ):
         security = None
 
@@ -196,16 +201,16 @@ def patch(  # type: ignore[override]
         self,
         rule: str,
         description: str,
-        cors: Optional[bool] = None,
+        cors: bool | None = None,
         compress: bool = False,
-        cache_control: Optional[str] = None,
-        summary: Optional[str] = None,
-        responses: Optional[Dict[int, OpenAPIResponse]] = None,
+        cache_control: str | None = None,
+        summary: str | None = None,
+        responses: dict[int, OpenAPIResponse] | None = None,
         response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
-        tags: Optional[List[str]] = None,
-        operation_id: Optional[str] = None,
+        tags: list[str] | None = None,
+        operation_id: str | None = None,
         include_in_schema: bool = True,
-        middlewares: Optional[List[Callable]] = None,
+        middlewares: list[Callable] | None = None,
     ):
         security = None
 
@@ -231,16 +236,16 @@ def delete(  # type: ignore[override]
         self,
         rule: str,
         description: str,
-        cors: Optional[bool] = None,
+        cors: bool | None = None,
         compress: bool = False,
-        cache_control: Optional[str] = None,
-        summary: Optional[str] = None,
-        responses: Optional[Dict[int, OpenAPIResponse]] = None,
+        cache_control: str | None = None,
+        summary: str | None = None,
+        responses: dict[int, OpenAPIResponse] | None = None,
         response_description: str = _DEFAULT_OPENAPI_RESPONSE_DESCRIPTION,
-        tags: Optional[List[str]] = None,
-        operation_id: Optional[str] = None,
+        tags: list[str] | None = None,
+        operation_id: str | None = None,
         include_in_schema: bool = True,
-        middlewares: Optional[List[Callable[..., Any]]] = None,
+        middlewares: list[Callable[..., Any]] | None = None,
     ):
         security = None
 
@@ -261,10 +266,10 @@ def delete(  # type: ignore[override]
         )
 
     @override
-    def _convert_matches_into_route_keys(self, match: Match) -> Dict[str, str]:
+    def _convert_matches_into_route_keys(self, match: Match) -> dict[str, str]:
         # In Bedrock Agents, all the parameters come inside the "parameters" key, not on the apiPath
         # So we have to search for route parameters in the parameters key
-        parameters: Dict[str, str] = {}
+        parameters: dict[str, str] = {}
         if match.groupdict() and self.current_event.parameters:
             parameters = {parameter["name"]: parameter["value"] for parameter in self.current_event.parameters}
         return parameters

aws_lambda_powertools/event_handler/bedrock_agent.py

@@ -1,11 +1,15 @@
-from typing import Callable, Dict, List, Optional, Pattern, Union
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Callable, Pattern
 
-from aws_lambda_powertools.event_handler import CORSConfig
 from aws_lambda_powertools.event_handler.api_gateway import (
     ApiGatewayResolver,
     ProxyEventType,
 )
-from aws_lambda_powertools.utilities.data_classes import LambdaFunctionUrlEvent
+
+if TYPE_CHECKING:
+    from aws_lambda_powertools.event_handler import CORSConfig
+    from aws_lambda_powertools.utilities.data_classes import LambdaFunctionUrlEvent
 
 
 class LambdaFunctionUrlResolver(ApiGatewayResolver):
@@ -48,10 +52,10 @@ def lambda_handler(event, context):
 
     def __init__(
         self,
-        cors: Optional[CORSConfig] = None,
-        debug: Optional[bool] = None,
-        serializer: Optional[Callable[[Dict], str]] = None,
-        strip_prefixes: Optional[List[Union[str, Pattern]]] = None,
+        cors: CORSConfig | None = None,
+        debug: bool | None = None,
+        serializer: Callable[[dict], str] | None = None,
+        strip_prefixes: list[str | Pattern] | None = None,
         enable_validation: bool = False,
     ):
         super().__init__(

aws_lambda_powertools/event_handler/lambda_function_url.py

@@ -1,9 +1,13 @@
+from __future__ import annotations
+
 from abc import ABC, abstractmethod
-from typing import Generic, Protocol
+from typing import TYPE_CHECKING, Generic, Protocol
 
-from aws_lambda_powertools.event_handler.api_gateway import Response
 from aws_lambda_powertools.event_handler.types import EventHandlerInstance
 
+if TYPE_CHECKING:
+    from aws_lambda_powertools.event_handler.api_gateway import Response
+
 
 class NextMiddleware(Protocol):
     def __call__(self, app: EventHandlerInstance) -> Response: