feat(data-classes): authorizer for http api and rest api (#620)

Author: Michael Brewer

aws_lambda_powertools/utilities/data_classes/api_gateway_authorizer_event.py

@@ -1,82 +1,11 @@
 from typing import Any, Dict, List, Optional
 
-from aws_lambda_powertools.utilities.data_classes.common import BaseProxyEvent, DictWrapper
-
-
-class APIGatewayEventIdentity(DictWrapper):
-    @property
-    def access_key(self) -> Optional[str]:
-        return self["requestContext"]["identity"].get("accessKey")
-
-    @property
-    def account_id(self) -> Optional[str]:
-        """The AWS account ID associated with the request."""
-        return self["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["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["requestContext"]["identity"].get("apiKeyId")
-
-    @property
-    def caller(self) -> Optional[str]:
-        """The principal identifier of the caller making the request."""
-        return self["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["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["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["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["requestContext"]["identity"].get("cognitoIdentityPoolId")
-
-    @property
-    def principal_org_id(self) -> Optional[str]:
-        """The AWS organization ID."""
-        return self["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["requestContext"]["identity"]["sourceIp"]
-
-    @property
-    def user(self) -> Optional[str]:
-        """The principal identifier of the user making the request."""
-        return self["requestContext"]["identity"].get("user")
-
-    @property
-    def user_agent(self) -> Optional[str]:
-        """The User Agent of the API caller."""
-        return self["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["requestContext"]["identity"].get("userArn")
+from aws_lambda_powertools.utilities.data_classes.common import (
+    BaseProxyEvent,
+    BaseRequestContext,
+    BaseRequestContextV2,
+    DictWrapper,
+)
 
 
 class APIGatewayEventAuthorizer(DictWrapper):
@@ -89,21 +18,7 @@ def scopes(self) -> Optional[List[str]]:
         return self["requestContext"]["authorizer"].get("scopes")
 
 
-class APIGatewayEventRequestContext(DictWrapper):
-    @property
-    def account_id(self) -> str:
-        """The AWS account ID associated with the request."""
-        return self["requestContext"]["accountId"]
-
-    @property
-    def api_id(self) -> str:
-        """The identifier API Gateway assigns to your API."""
-        return self["requestContext"]["apiId"]
-
-    @property
-    def authorizer(self) -> APIGatewayEventAuthorizer:
-        return APIGatewayEventAuthorizer(self._data)
-
+class APIGatewayEventRequestContext(BaseRequestContext):
     @property
     def connected_at(self) -> Optional[int]:
         """The Epoch-formatted connection time. (WebSocket API)"""
@@ -114,40 +29,11 @@ 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["requestContext"].get("connectionId")
 
-    @property
-    def domain_name(self) -> Optional[str]:
-        """A domain name"""
-        return self["requestContext"].get("domainName")
-
-    @property
-    def domain_prefix(self) -> Optional[str]:
-        return self["requestContext"].get("domainPrefix")
-
     @property
     def event_type(self) -> Optional[str]:
         """The event type: `CONNECT`, `MESSAGE`, or `DISCONNECT`. (WebSocket API)"""
         return self["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["requestContext"].get("extendedRequestId")
-
-    @property
-    def protocol(self) -> str:
-        """The request protocol, for example, HTTP/1.1."""
-        return self["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["requestContext"]["httpMethod"]
-
-    @property
-    def identity(self) -> APIGatewayEventIdentity:
-        return APIGatewayEventIdentity(self._data)
-
     @property
     def message_direction(self) -> Optional[str]:
         """Message direction (WebSocket API)"""
@@ -159,46 +45,18 @@ def message_id(self) -> Optional[str]:
         return self["requestContext"].get("messageId")
 
     @property
-    def path(self) -> str:
-        return self["requestContext"]["path"]
-
-    @property
-    def stage(self) -> str:
-        """The deployment stage of the API request"""
-        return self["requestContext"]["stage"]
-
-    @property
-    def request_id(self) -> str:
-        """The ID that API Gateway assigns to the API request."""
-        return self["requestContext"]["requestId"]
-
-    @property
-    def request_time(self) -> Optional[str]:
-        """The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)"""
-        return self["requestContext"].get("requestTime")
-
-    @property
-    def request_time_epoch(self) -> int:
-        """The Epoch-formatted request time."""
-        return self["requestContext"]["requestTimeEpoch"]
-
-    @property
-    def resource_id(self) -> str:
-        return self["requestContext"]["resourceId"]
-
-    @property
-    def resource_path(self) -> str:
-        return self["requestContext"]["resourcePath"]
+    def operation_name(self) -> Optional[str]:
+        """The name of the operation being performed"""
+        return self["requestContext"].get("operationName")
 
     @property
     def route_key(self) -> Optional[str]:
         """The selected route key."""
         return self["requestContext"].get("routeKey")
 
     @property
-    def operation_name(self) -> Optional[str]:
-        """The name of the operation being performed"""
-        return self["requestContext"].get("operationName")
+    def authorizer(self) -> APIGatewayEventAuthorizer:
+        return APIGatewayEventAuthorizer(self._data)
 
 
 class APIGatewayProxyEvent(BaseProxyEvent):
@@ -238,31 +96,6 @@ def stage_variables(self) -> Optional[Dict[str, str]]:
         return self.get("stageVariables")
 
 
-class RequestContextV2Http(DictWrapper):
-    @property
-    def method(self) -> str:
-        return self["requestContext"]["http"]["method"]
-
-    @property
-    def path(self) -> str:
-        return self["requestContext"]["http"]["path"]
-
-    @property
-    def protocol(self) -> str:
-        """The request protocol, for example, HTTP/1.1."""
-        return self["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["requestContext"]["http"]["sourceIp"]
-
-    @property
-    def user_agent(self) -> str:
-        """The User Agent of the API caller."""
-        return self["requestContext"]["http"]["userAgent"]
-
-
 class RequestContextV2AuthorizerIam(DictWrapper):
     @property
     def access_key(self) -> Optional[str]:
@@ -334,60 +167,12 @@ def iam(self) -> Optional[RequestContextV2AuthorizerIam]:
         return None if iam is None else RequestContextV2AuthorizerIam(iam)
 
 
-class RequestContextV2(DictWrapper):
-    @property
-    def account_id(self) -> str:
-        """The AWS account ID associated with the request."""
-        return self["requestContext"]["accountId"]
-
-    @property
-    def api_id(self) -> str:
-        """The identifier API Gateway assigns to your API."""
-        return self["requestContext"]["apiId"]
-
+class RequestContextV2(BaseRequestContextV2):
     @property
     def authorizer(self) -> Optional[RequestContextV2Authorizer]:
         authorizer = self["requestContext"].get("authorizer")
         return None if authorizer is None else RequestContextV2Authorizer(authorizer)
 
-    @property
-    def domain_name(self) -> str:
-        """A domain name"""
-        return self["requestContext"]["domainName"]
-
-    @property
-    def domain_prefix(self) -> str:
-        return self["requestContext"]["domainPrefix"]
-
-    @property
-    def http(self) -> RequestContextV2Http:
-        return RequestContextV2Http(self._data)
-
-    @property
-    def request_id(self) -> str:
-        """The ID that API Gateway assigns to the API request."""
-        return self["requestContext"]["requestId"]
-
-    @property
-    def route_key(self) -> str:
-        """The selected route key."""
-        return self["requestContext"]["routeKey"]
-
-    @property
-    def stage(self) -> str:
-        """The deployment stage of the API request"""
-        return self["requestContext"]["stage"]
-
-    @property
-    def time(self) -> str:
-        """The CLF-formatted request time (dd/MMM/yyyy:HH:mm:ss +-hhmm)."""
-        return self["requestContext"]["time"]
-
-    @property
-    def time_epoch(self) -> int:
-        """The Epoch-formatted request time."""
-        return self["requestContext"]["timeEpoch"]
-
 
 class APIGatewayProxyEventV2(BaseProxyEvent):
     """AWS Lambda proxy V2 event