From cfd6d1afc986f68cb350a1b564e2649377241a09 Mon Sep 17 00:00:00 2001 From: rafrafek <23004737+rafrafek@users.noreply.github.com> Date: Wed, 20 Nov 2024 16:27:15 +0100 Subject: [PATCH 1/2] Fixed type annotation for Api Gateway Router --- .../event_handler/api_gateway.py | 25 ++++++++++--------- 1 file changed, 13 insertions(+), 12 deletions(-) diff --git a/aws_lambda_powertools/event_handler/api_gateway.py b/aws_lambda_powertools/event_handler/api_gateway.py index fcb22addf6b..39c9537f2d9 100644 --- a/aws_lambda_powertools/event_handler/api_gateway.py +++ b/aws_lambda_powertools/event_handler/api_gateway.py @@ -898,6 +898,7 @@ def build(self, event: ResponseEventT, cors: CORSConfig | None = None) -> dict[s **event.header_serializer().serialize(headers=self.response.headers, cookies=self.response.cookies), } +T_route = TypeVar("T_route", bound=Callable[..., Any]) class BaseRouter(ABC): current_event: BaseProxyEvent @@ -924,7 +925,7 @@ def route( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ): + ) -> Callable[[T_route], T_route]: raise NotImplementedError() def use(self, middlewares: list[Callable[..., Response]]) -> None: @@ -984,7 +985,7 @@ def get( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ): + ) -> Callable[[T_route], T_route]: """Get route decorator with GET `method` Examples @@ -1041,7 +1042,7 @@ def post( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ): + ) -> Callable[[T_route], T_route]: """Post route decorator with POST `method` Examples @@ -1099,7 +1100,7 @@ def put( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ): + ) -> Callable[[T_route], T_route]: """Put route decorator with PUT `method` Examples @@ -1157,7 +1158,7 @@ def delete( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ): + ) -> Callable[[T_route], T_route]: """Delete route decorator with DELETE `method` Examples @@ -1214,7 +1215,7 @@ def patch( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable] | None = None, - ): + ) -> Callable[[T_route], T_route]: """Patch route decorator with PATCH `method` Examples @@ -1274,7 +1275,7 @@ def head( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable] | None = None, - ): + ) -> Callable[[T_route], T_route]: """Head route decorator with HEAD `method` Examples @@ -1950,10 +1951,10 @@ def route( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ): + ) -> Callable[[T_route], T_route]: """Route decorator includes parameter `method`""" - def register_resolver(func: Callable): + def register_resolver(func: T_route) -> T_route: methods = (method,) if isinstance(method, str) else method logger.debug(f"Adding route using rule {rule} and methods: {','.join(m.upper() for m in methods)}") @@ -2492,8 +2493,8 @@ def route( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ): - def register_route(func: Callable): + ) -> Callable[[T_route], T_route]: + def register_route(func: T_route) -> T_route: # All dict keys needs to be hashable. So we'll need to do some conversions: methods = (method,) if isinstance(method, str) else tuple(method) frozen_responses = _FrozenDict(responses) if responses else None @@ -2598,7 +2599,7 @@ def route( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ): + ) -> Callable[[T_route], T_route]: # NOTE: see #1552 for more context. return super().route( rule.rstrip("/"), From 6470d2e01e9d7be4c70e729a61d785c3e125a6a5 Mon Sep 17 00:00:00 2001 From: rafrafek <23004737+rafrafek@users.noreply.github.com> Date: Sun, 24 Nov 2024 00:12:39 +0100 Subject: [PATCH 2/2] replaced T_route with AnyCallableT --- .../event_handler/api_gateway.py | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/aws_lambda_powertools/event_handler/api_gateway.py b/aws_lambda_powertools/event_handler/api_gateway.py index d7b544cef77..001fcceac72 100644 --- a/aws_lambda_powertools/event_handler/api_gateway.py +++ b/aws_lambda_powertools/event_handler/api_gateway.py @@ -83,6 +83,7 @@ TypeModelOrEnum, ) from aws_lambda_powertools.shared.cookies import Cookie + from aws_lambda_powertools.shared.types import AnyCallableT from aws_lambda_powertools.utilities.typing import LambdaContext @@ -898,7 +899,6 @@ def build(self, event: ResponseEventT, cors: CORSConfig | None = None) -> dict[s **event.header_serializer().serialize(headers=self.response.headers, cookies=self.response.cookies), } -T_route = TypeVar("T_route", bound=Callable[..., Any]) class BaseRouter(ABC): current_event: BaseProxyEvent @@ -925,7 +925,7 @@ def route( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ) -> Callable[[T_route], T_route]: + ) -> Callable[[AnyCallableT], AnyCallableT]: raise NotImplementedError() def use(self, middlewares: list[Callable[..., Response]]) -> None: @@ -985,7 +985,7 @@ def get( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ) -> Callable[[T_route], T_route]: + ) -> Callable[[AnyCallableT], AnyCallableT]: """Get route decorator with GET `method` Examples @@ -1042,7 +1042,7 @@ def post( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ) -> Callable[[T_route], T_route]: + ) -> Callable[[AnyCallableT], AnyCallableT]: """Post route decorator with POST `method` Examples @@ -1100,7 +1100,7 @@ def put( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ) -> Callable[[T_route], T_route]: + ) -> Callable[[AnyCallableT], AnyCallableT]: """Put route decorator with PUT `method` Examples @@ -1158,7 +1158,7 @@ def delete( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ) -> Callable[[T_route], T_route]: + ) -> Callable[[AnyCallableT], AnyCallableT]: """Delete route decorator with DELETE `method` Examples @@ -1215,7 +1215,7 @@ def patch( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable] | None = None, - ) -> Callable[[T_route], T_route]: + ) -> Callable[[AnyCallableT], AnyCallableT]: """Patch route decorator with PATCH `method` Examples @@ -1275,7 +1275,7 @@ def head( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable] | None = None, - ) -> Callable[[T_route], T_route]: + ) -> Callable[[AnyCallableT], AnyCallableT]: """Head route decorator with HEAD `method` Examples @@ -1951,10 +1951,10 @@ def route( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ) -> Callable[[T_route], T_route]: + ) -> Callable[[AnyCallableT], AnyCallableT]: """Route decorator includes parameter `method`""" - def register_resolver(func: T_route) -> T_route: + def register_resolver(func: AnyCallableT) -> AnyCallableT: methods = (method,) if isinstance(method, str) else method logger.debug(f"Adding route using rule {rule} and methods: {','.join(m.upper() for m in methods)}") @@ -2493,8 +2493,8 @@ def route( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ) -> Callable[[T_route], T_route]: - def register_route(func: T_route) -> T_route: + ) -> Callable[[AnyCallableT], AnyCallableT]: + def register_route(func: AnyCallableT) -> AnyCallableT: # All dict keys needs to be hashable. So we'll need to do some conversions: methods = (method,) if isinstance(method, str) else tuple(method) frozen_responses = _FrozenDict(responses) if responses else None @@ -2599,7 +2599,7 @@ def route( security: list[dict[str, list[str]]] | None = None, openapi_extensions: dict[str, Any] | None = None, middlewares: list[Callable[..., Any]] | None = None, - ) -> Callable[[T_route], T_route]: + ) -> Callable[[AnyCallableT], AnyCallableT]: # NOTE: see #1552 for more context. return super().route( rule.rstrip("/"),