feat(trigger): data class and helper functions for lambda trigger events

aws_lambda_powertools/utilities/trigger/api_gateway_proxy_event.py

@@ -2,205 +2,205 @@
 
 
 class APIGatewayEventIdentity:
-    def __init__(self, event: dict):
-        self._val = event
+    def __init__(self, event: Dict[str, Any]):
+        self._v = event
 
     @property
     def access_key(self) -> Optional[str]:
-        return self._val["requestContext"]["identity"].get("accessKey")
+        return self._v["requestContext"]["identity"].get("accessKey")
 
     @property
     def account_id(self) -> Optional[str]:
         """The AWS account ID associated with the request."""
-        return self._val["requestContext"]["identity"].get("accountId")
+        return self._v["requestContext"]["identity"].get("accountId")
 
     @property
     def api_key(self) -> Optional[str]:
         """For API methods that require an API key, this variable is the API key associated with the method request.
         For methods that don't require an API key, this variable is null. """
-        return self._val["requestContext"]["identity"].get("apiKey")
+        return self._v["requestContext"]["identity"].get("apiKey")
 
     @property
     def api_key_id(self) -> Optional[str]:
         """The API key ID associated with an API request that requires an API key."""
-        return self._val["requestContext"]["identity"].get("apiKeyId")
+        return self._v["requestContext"]["identity"].get("apiKeyId")
 
     @property
     def caller(self) -> Optional[str]:
         """The principal identifier of the caller making the request."""
-        return self._val["requestContext"]["identity"].get("caller")
+        return self._v["requestContext"]["identity"].get("caller")
 
     @property
     def cognito_authentication_provider(self) -> Optional[str]:
         """A comma-separated list of the Amazon Cognito authentication providers used by the caller
         making the request. Available only if the request was signed with Amazon Cognito credentials."""
-        return self._val["requestContext"]["identity"].get("cognitoAuthenticationProvider")
+        return self._v["requestContext"]["identity"].get("cognitoAuthenticationProvider")
 
     @property
     def cognito_authentication_type(self) -> Optional[str]:
         """The Amazon Cognito authentication type of the caller making the request.
         Available only if the request was signed with Amazon Cognito credentials."""
-        return self._val["requestContext"]["identity"].get("cognitoAuthenticationType")
+        return self._v["requestContext"]["identity"].get("cognitoAuthenticationType")
 
     @property
     def cognito_identity_id(self) -> Optional[str]:
         """The Amazon Cognito identity ID of the caller making the request.
         Available only if the request was signed with Amazon Cognito credentials."""
-        return self._val["requestContext"]["identity"].get("cognitoIdentityId")
+        return self._v["requestContext"]["identity"].get("cognitoIdentityId")
 
     @property
     def cognito_identity_pool_id(self) -> Optional[str]:
         """The Amazon Cognito identity pool ID of the caller making the request.
         Available only if the request was signed with Amazon Cognito credentials."""
-        return self._val["requestContext"]["identity"].get("cognitoIdentityPoolId")
+        return self._v["requestContext"]["identity"].get("cognitoIdentityPoolId")
 
     @property
     def principal_org_id(self) -> Optional[str]:
         """The AWS organization ID."""
-        return self._val["requestContext"]["identity"].get("principalOrgId")
+        return self._v["requestContext"]["identity"].get("principalOrgId")
 
     @property
     def source_ip(self) -> str:
         """The source IP address of the TCP connection making the request to API Gateway."""
-        return self._val["requestContext"]["identity"]["sourceIp"]
+        return self._v["requestContext"]["identity"]["sourceIp"]
 
     @property
     def user(self) -> Optional[str]:
         """The principal identifier of the user making the request."""
-        return self._val["requestContext"]["identity"].get("user")
+        return self._v["requestContext"]["identity"].get("user")
 
     @property
     def user_agent(self) -> Optional[str]:
         """The User Agent of the API caller."""
-        return self._val["requestContext"]["identity"].get("userAgent")
+        return self._v["requestContext"]["identity"].get("userAgent")
 
     @property
     def user_arn(self) -> Optional[str]:
         """The Amazon Resource Name (ARN) of the effective user identified after authentication."""
-        return self._val["requestContext"]["identity"].get("userArn")
+        return self._v["requestContext"]["identity"].get("userArn")
 
 
 class APIGatewayEventAuthorizer:
-    def __init__(self, event: Dict):
-        self._val = event
+    def __init__(self, event: Dict[str, Any]):
+        self._v = event
 
     @property
     def claims(self) -> Optional[Dict[str, Any]]:
-        return self._val["requestContext"]["authorizer"].get("claims")
+        return self._v["requestContext"]["authorizer"].get("claims")
 
     @property
     def scopes(self) -> Optional[List[str]]:
-        return self._val["requestContext"]["authorizer"].get("scopes")
+        return self._v["requestContext"]["authorizer"].get("scopes")
 
 
 class APIGatewayEventRequestContext:
     def __init__(self, event: Dict[str, Any]):
-        self._val = event
+        self._v = event
 
     @property
     def account_id(self) -> str:
         """The AWS account ID associated with the request."""
-        return self._val["requestContext"]["accountId"]
+        return self._v["requestContext"]["accountId"]
 
     @property
     def api_id(self) -> str:
         """The identifier API Gateway assigns to your API."""
-        return self._val["requestContext"]["apiId"]
+        return self._v["requestContext"]["apiId"]
 
     @property
     def authorizer(self) -> APIGatewayEventAuthorizer:
-        return APIGatewayEventAuthorizer(self._val)
+        return APIGatewayEventAuthorizer(self._v)
 
     @property
     def connected_at(self) -> Optional[int]:
         """The Epoch-formatted connection time. (WebSocket API)"""
-        return self._val["requestContext"].get("connectedAt")
+        return self._v["requestContext"].get("connectedAt")
 
     @property
     def connection_id(self) -> Optional[str]:
         """A unique ID for the connection that can be used to make a callback to the client. (WebSocket API)"""
-        return self._val["requestContext"].get("connectionId")
+        return self._v["requestContext"].get("connectionId")
 
     @property
     def domain_name(self) -> Optional[str]:
         """A domain name"""
-        return self._val["requestContext"].get("domainName")
+        return self._v["requestContext"].get("domainName")
 
     @property
     def domain_prefix(self) -> Optional[str]:
-        return self._val["requestContext"].get("domainPrefix")
+        return self._v["requestContext"].get("domainPrefix")
 
     @property
     def event_type(self) -> Optional[str]:
         """The event type: `CONNECT`, `MESSAGE`, or `DISCONNECT`. (WebSocket API)"""
-        return self._val["requestContext"].get("eventType")
+        return self._v["requestContext"].get("eventType")
 
     @property
     def extended_request_id(self) -> Optional[str]:
         """An automatically generated ID for the API call, which contains more useful information
         for debugging/troubleshooting."""
-        return self._val["requestContext"].get("extendedRequestId")
+        return self._v["requestContext"].get("extendedRequestId")
 
     @property
     def protocol(self) -> str:
         """The request protocol, for example, HTTP/1.1."""
-        return self._val["requestContext"]["protocol"]
+        return self._v["requestContext"]["protocol"]
 
     @property
     def http_method(self) -> str:
         """The HTTP method used. Valid values include: DELETE, GET, HEAD, OPTIONS, PATCH, POST, and PUT."""
-        return self._val["requestContext"]["httpMethod"]
+        return self._v["requestContext"]["httpMethod"]
 
     @property
     def identity(self) -> APIGatewayEventIdentity:
-        return APIGatewayEventIdentity(self._val)
+        return APIGatewayEventIdentity(self._v)
 
     @property
     def message_direction(self) -> Optional[str]:
         """Message direction (WebSocket API)"""
-        return self._val["requestContext"].get("messageDirection")
+        return self._v["requestContext"].get("messageDirection")
 
     @property
     def message_id(self) -> Optional[str]:
         """A unique server-side ID for a message. Available only when the `eventType` is `MESSAGE`."""
-        return self._val["requestContext"].get("messageId")
+        return self._v["requestContext"].get("messageId")
 
     @property
     def path(self) -> str:
-        return self._val["requestContext"]["path"]
+        return self._v["requestContext"]["path"]
 
     @property
     def stage(self) -> str:
         """The deployment stage of the API request """
-        return self._val["requestContext"]["stage"]
+        return self._v["requestContext"]["stage"]
 
     @property
     def request_id(self) -> str:
         """The ID that API Gateway assigns to the API request."""
-        return self._val["requestContext"]["requestId"]
+        return self._v["requestContext"]["requestId"]
 
     @property
     def request_time(self) -> Optional[str]:
         """The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)"""
-        return self._val["requestContext"].get("requestTime")
+        return self._v["requestContext"].get("requestTime")
 
     @property
     def request_time_epoch(self) -> int:
         """The Epoch-formatted request time."""
-        return self._val["requestContext"]["requestTimeEpoch"]
+        return self._v["requestContext"]["requestTimeEpoch"]
 
     @property
     def resource_id(self) -> str:
-        return self._val["requestContext"]["resourceId"]
+        return self._v["requestContext"]["resourceId"]
 
     @property
     def resource_path(self) -> str:
-        return self._val["requestContext"]["resourcePath"]
+        return self._v["requestContext"]["resourcePath"]
 
     @property
     def route_key(self) -> Optional[str]:
         """The selected route key."""
-        return self._val["requestContext"].get("routeKey")
+        return self._v["requestContext"].get("routeKey")
 
 
 class APIGatewayProxyEvent(dict):
@@ -266,102 +266,102 @@ def is_base64_encoded(self) -> bool:
 
 
 class RequestContextV2Http:
-    def __init__(self, event: dict):
-        self._val = event
+    def __init__(self, event: Dict[str, Any]):
+        self._v = event
 
     @property
     def method(self) -> str:
-        return self._val["requestContext"]["http"]["method"]
+        return self._v["requestContext"]["http"]["method"]
 
     @property
     def path(self) -> str:
-        return self._val["requestContext"]["http"]["path"]
+        return self._v["requestContext"]["http"]["path"]
 
     @property
     def protocol(self) -> str:
         """The request protocol, for example, HTTP/1.1."""
-        return self._val["requestContext"]["http"]["protocol"]
+        return self._v["requestContext"]["http"]["protocol"]
 
     @property
     def source_ip(self) -> str:
         """The source IP address of the TCP connection making the request to API Gateway."""
-        return self._val["requestContext"]["http"]["sourceIp"]
+        return self._v["requestContext"]["http"]["sourceIp"]
 
     @property
     def user_agent(self) -> str:
         """The User Agent of the API caller."""
-        return self._val["requestContext"]["http"]["userAgent"]
+        return self._v["requestContext"]["http"]["userAgent"]
 
 
 class RequestContextV2Authorizer:
-    def __init__(self, event: dict):
-        self._val = event
+    def __init__(self, event: Dict[str, Any]):
+        self._v = event
 
     @property
     def jwt_claim(self) -> Dict[str, Any]:
-        return self._val["jwt"]["claims"]
+        return self._v["jwt"]["claims"]
 
     @property
     def jwt_scopes(self) -> List[str]:
-        return self._val["jwt"]["scopes"]
+        return self._v["jwt"]["scopes"]
 
 
 class RequestContextV2:
-    def __init__(self, event: dict):
-        self._val = event
+    def __init__(self, event: Dict[str, Any]):
+        self._v = event
 
     @property
     def account_id(self) -> str:
         """The AWS account ID associated with the request."""
-        return self._val["requestContext"]["accountId"]
+        return self._v["requestContext"]["accountId"]
 
     @property
     def api_id(self) -> str:
         """The identifier API Gateway assigns to your API."""
-        return self._val["requestContext"]["apiId"]
+        return self._v["requestContext"]["apiId"]
 
     @property
     def authorizer(self) -> Optional[RequestContextV2Authorizer]:
-        authorizer = self._val["requestContext"].get("authorizer")
+        authorizer = self._v["requestContext"].get("authorizer")
         return None if authorizer is None else RequestContextV2Authorizer(authorizer)
 
     @property
     def domain_name(self) -> str:
         """A domain name """
-        return self._val["requestContext"]["domainName"]
+        return self._v["requestContext"]["domainName"]
 
     @property
     def domain_prefix(self) -> str:
-        return self._val["requestContext"]["domainPrefix"]
+        return self._v["requestContext"]["domainPrefix"]
 
     @property
     def http(self) -> RequestContextV2Http:
-        return RequestContextV2Http(self._val)
+        return RequestContextV2Http(self._v)
 
     @property
     def request_id(self) -> str:
         """The ID that API Gateway assigns to the API request."""
-        return self._val["requestContext"]["requestId"]
+        return self._v["requestContext"]["requestId"]
 
     @property
     def route_key(self) -> str:
         """The selected route key."""
-        return self._val["requestContext"]["routeKey"]
+        return self._v["requestContext"]["routeKey"]
 
     @property
     def stage(self) -> str:
         """The deployment stage of the API request """
-        return self._val["requestContext"]["stage"]
+        return self._v["requestContext"]["stage"]
 
     @property
     def time(self) -> str:
         """The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)."""
-        return self._val["requestContext"]["time"]
+        return self._v["requestContext"]["time"]
 
     @property
     def time_epoch(self) -> int:
         """The Epoch-formatted request time."""
-        return self._val["requestContext"]["timeEpoch"]
+        return self._v["requestContext"]["timeEpoch"]
 
 
 class APIGatewayProxyEventV2(dict):