Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 13 additions & 53 deletions aws_lambda_powertools/event_handler/api_gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -1460,9 +1460,11 @@ def lambda_handler(event, context):
```
"""

_proxy_event_type: Enum = ProxyEventType.APIGatewayProxyEvent

def __init__(
self,
proxy_type: Enum = ProxyEventType.APIGatewayProxyEvent,
proxy_type: Enum | None = None,
cors: CORSConfig | None = None,
debug: bool | None = None,
serializer: Callable[[dict], str] | None = None,
Expand Down Expand Up @@ -1495,7 +1497,7 @@ def __init__(
function to deserialize `str`, `bytes`, `bytearray` containing a JSON document to a Python `dict`,
by default json.loads when integrating with EventSource data class
"""
self._proxy_type = proxy_type
self._proxy_type = proxy_type or self._proxy_event_type
self._dynamic_routes: list[Route] = []
self._static_routes: list[Route] = []
self._route_keys: list[str] = []
Expand Down Expand Up @@ -2935,28 +2937,7 @@ class APIGatewayRestResolver(ApiGatewayResolver):
"""Amazon API Gateway REST and HTTP API v1 payload resolver"""

current_event: APIGatewayProxyEvent

def __init__(
self,
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,
response_validation_error_http_code: HTTPStatus | int | None = None,
json_body_deserializer: Callable[[str], dict] | None = None,
):
"""Amazon API Gateway REST and HTTP API v1 payload resolver"""
super().__init__(
ProxyEventType.APIGatewayProxyEvent,
cors,
debug,
serializer,
strip_prefixes,
enable_validation,
response_validation_error_http_code,
json_body_deserializer=json_body_deserializer,
)
_proxy_event_type = ProxyEventType.APIGatewayProxyEvent

def _get_base_path(self) -> str:
# 3 different scenarios:
Expand Down Expand Up @@ -3025,28 +3006,7 @@ class APIGatewayHttpResolver(ApiGatewayResolver):
"""Amazon API Gateway HTTP API v2 payload resolver"""

current_event: APIGatewayProxyEventV2

def __init__(
self,
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,
response_validation_error_http_code: HTTPStatus | int | None = None,
json_body_deserializer: Callable[[str], dict] | None = None,
):
"""Amazon API Gateway HTTP API v2 payload resolver"""
super().__init__(
ProxyEventType.APIGatewayProxyEventV2,
cors,
debug,
serializer,
strip_prefixes,
enable_validation,
response_validation_error_http_code,
json_body_deserializer=json_body_deserializer,
)
_proxy_event_type = ProxyEventType.APIGatewayProxyEventV2

def _get_base_path(self) -> str:
# 3 different scenarios:
Expand All @@ -3066,6 +3026,7 @@ class ALBResolver(ApiGatewayResolver):
"""Amazon Application Load Balancer (ALB) resolver"""

current_event: ALBEvent
_proxy_event_type = ProxyEventType.ALBEvent

def __init__(
self,
Expand Down Expand Up @@ -3105,13 +3066,12 @@ def __init__(
Enables URL-decoding of query parameters (both keys and values), by default False.
"""
super().__init__(
ProxyEventType.ALBEvent,
cors,
debug,
serializer,
strip_prefixes,
enable_validation,
response_validation_error_http_code,
cors=cors,
debug=debug,
serializer=serializer,
strip_prefixes=strip_prefixes,
enable_validation=enable_validation,
response_validation_error_http_code=response_validation_error_http_code,
json_body_deserializer=json_body_deserializer,
)
self.decode_query_parameters = decode_query_parameters
Expand Down
28 changes: 2 additions & 26 deletions aws_lambda_powertools/event_handler/lambda_function_url.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,13 @@
from __future__ import annotations

from typing import TYPE_CHECKING, Pattern
from typing import TYPE_CHECKING

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
from aws_lambda_powertools.utilities.data_classes import LambdaFunctionUrlEvent


Expand Down Expand Up @@ -52,27 +48,7 @@ def lambda_handler(event, context):
"""

current_event: LambdaFunctionUrlEvent

def __init__(
self,
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,
response_validation_error_http_code: HTTPStatus | int | None = None,
json_body_deserializer: Callable[[str], dict] | None = None,
):
super().__init__(
ProxyEventType.LambdaFunctionUrlEvent,
cors,
debug,
serializer,
strip_prefixes,
enable_validation,
response_validation_error_http_code,
json_body_deserializer=json_body_deserializer,
)
_proxy_event_type = ProxyEventType.LambdaFunctionUrlEvent

def _get_base_path(self) -> str:
stage = self.current_event.request_context.stage
Expand Down
52 changes: 3 additions & 49 deletions aws_lambda_powertools/event_handler/vpc_lattice.py
Original file line number Diff line number Diff line change
@@ -1,17 +1,13 @@
from __future__ import annotations

from typing import TYPE_CHECKING, Pattern
from typing import TYPE_CHECKING

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
from aws_lambda_powertools.utilities.data_classes import VPCLatticeEvent, VPCLatticeEventV2


Expand Down Expand Up @@ -48,28 +44,7 @@ def lambda_handler(event, context):
"""

current_event: VPCLatticeEvent

def __init__(
self,
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,
response_validation_error_http_code: HTTPStatus | int | None = None,
json_body_deserializer: Callable[[str], dict] | None = None,
):
"""Amazon VPC Lattice resolver"""
super().__init__(
ProxyEventType.VPCLatticeEvent,
cors,
debug,
serializer,
strip_prefixes,
enable_validation,
response_validation_error_http_code,
json_body_deserializer=json_body_deserializer,
)
_proxy_event_type = ProxyEventType.VPCLatticeEvent

def _get_base_path(self) -> str:
return ""
Expand Down Expand Up @@ -108,28 +83,7 @@ def lambda_handler(event, context):
"""

current_event: VPCLatticeEventV2

def __init__(
self,
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,
response_validation_error_http_code: HTTPStatus | int | None = None,
json_body_deserializer: Callable[[str], dict] | None = None,
):
"""Amazon VPC Lattice resolver"""
super().__init__(
ProxyEventType.VPCLatticeEventV2,
cors,
debug,
serializer,
strip_prefixes,
enable_validation,
response_validation_error_http_code,
json_body_deserializer=json_body_deserializer,
)
_proxy_event_type = ProxyEventType.VPCLatticeEventV2

def _get_base_path(self) -> str:
return ""
Loading