feat(data-classes): Add S3ObjectSessionContext

Author: Michael Brewer

aws_lambda_powertools/utilities/data_classes/s3_event.py

@@ -1,7 +1,7 @@
-from typing import Any, Dict, Iterator, Optional
+from typing import Dict, Iterator, Optional
 from urllib.parse import unquote_plus
 
-from aws_lambda_powertools.utilities.data_classes.common import DictWrapper, get_header_value
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
 
 
 class S3Identity(DictWrapper):
@@ -187,208 +187,3 @@ def bucket_name(self) -> str:
     def object_key(self) -> str:
         """Get the object key for the first s3 event record and unquote plus"""
         return unquote_plus(self["Records"][0]["s3"]["object"]["key"])
-
-
-class S3ObjectContext(DictWrapper):
-    """The input and output details for connections to Amazon S3 and S3 Object Lambda."""
-
-    @property
-    def input_s3_url(self) -> str:
-        """A presigned URL that can be used to fetch the original object from Amazon S3.
-        The URL is signed using the original caller’s identity, and their permissions
-        will apply when the URL is used. If there are signed headers in the URL, the
-        Lambda function must include these in the call to Amazon S3, except for the Host."""
-        return self["inputS3Url"]
-
-    @property
-    def output_route(self) -> str:
-        """A routing token that is added to the S3 Object Lambda URL when the Lambda function
-        calls `WriteGetObjectResponse`."""
-        return self["outputRoute"]
-
-    @property
-    def output_token(self) -> str:
-        """An opaque token used by S3 Object Lambda to match the WriteGetObjectResponse call
-        with the original caller."""
-        return self["outputToken"]
-
-
-class S3ObjectConfiguration(DictWrapper):
-    """Configuration information about the S3 Object Lambda access point."""
-
-    @property
-    def access_point_arn(self) -> str:
-        """The Amazon Resource Name (ARN) of the S3 Object Lambda access point that received
-        this request."""
-        return self["accessPointArn"]
-
-    @property
-    def supporting_access_point_arn(self) -> str:
-        """The ARN of the supporting access point that is specified in the S3 Object Lambda
-        access point configuration."""
-        return self["supportingAccessPointArn"]
-
-    @property
-    def payload(self) -> str:
-        """Custom data that is applied to the S3 Object Lambda access point configuration.
-        S3 Object Lambda treats this as an opaque string, so it might need to be decoded
-        before use."""
-        return self["payload"]
-
-
-class S3ObjectUserRequest(DictWrapper):
-    """ Information about the original call to S3 Object Lambda."""
-
-    @property
-    def url(self) -> str:
-        """The decoded URL of the request as received by S3 Object Lambda, excluding any
-        authorization-related query parameters."""
-        return self["url"]
-
-    @property
-    def headers(self) -> Dict[str, str]:
-        """A map of string to strings containing the HTTP headers and their values from the
-        original call, excluding any authorization-related headers. If the same header appears
-        multiple times, their values are combined into a comma-delimited list. The case of the
-        original headers is retained in this map."""
-        return self["headers"]
-
-    def get_header_value(
-        self, name: str, default_value: Optional[str] = None, case_sensitive: Optional[bool] = False
-    ) -> Optional[str]:
-        """Get header value by name
-
-        Parameters
-        ----------
-        name: str
-            Header name
-        default_value: str, optional
-            Default value if no value was found by name
-        case_sensitive: bool
-            Whether to use a case sensitive look up
-        Returns
-        -------
-        str, optional
-            Header value
-        """
-        return get_header_value(self.headers, name, default_value, case_sensitive)
-
-
-class S3ObjectUserIdentity(DictWrapper):
-    """Details about the identity that made the call to S3 Object Lambda.
-
-    Documentation:
-    -------------
-    - https://docs.aws.amazon.com/awscloudtrail/latest/userguide/cloudtrail-event-reference-user-identity.html
-    """
-
-    @property
-    def get_type(self) -> str:
-        """The type of identity.
-
-        The following values are possible:
-
-        - Root – The request was made with your AWS account credentials. If the userIdentity
-          type is Root and you set an alias for your account, the userName field contains your account alias.
-          For more information, see Your AWS Account ID and Its Alias.
-        - IAMUser – The request was made with the credentials of an IAM user.
-        - AssumedRole – The request was made with temporary security credentials that were obtained
-          with a role via a call to the AWS Security Token Service (AWS STS) AssumeRole API. This can include
-          roles for Amazon EC2 and cross-account API access.
-        - FederatedUser – The request was made with temporary security credentials that were obtained via a
-          call to the AWS STS GetFederationToken API. The sessionIssuer element indicates if the API was
-          called with root or IAM user credentials.
-        - AWSAccount – The request was made by another AWS account.
-        -  AWSService – The request was made by an AWS account that belongs to an AWS service.
-          For example, AWS Elastic Beanstalk assumes an IAM role in your account to call other AWS services
-          on your behalf.
-        """
-        return self["type"]
-
-    @property
-    def account_id(self) -> str:
-        """The account that owns the entity that granted permissions for the request.
-        If the request was made with temporary security credentials, this is the account that owns the IAM
-        user or role that was used to obtain credentials."""
-        return self["accountId"]
-
-    @property
-    def access_key_id(self) -> str:
-        """The access key ID that was used to sign the request.
-
-        If the request was made with temporary security credentials, this is the access key ID of
-        the temporary credentials. For security reasons, accessKeyId might not be present, or might
-        be displayed as an empty string."""
-        return self["accessKeyId"]
-
-    @property
-    def user_name(self) -> str:
-        """The friendly name of the identity that made the call."""
-        return self["userName"]
-
-    @property
-    def principal_id(self) -> str:
-        """The unique identifier for the identity that made the call.
-
-        For requests made with temporary security credentials, this value includes
-        the session name that is passed to the AssumeRole, AssumeRoleWithWebIdentity,
-        or GetFederationToken API call."""
-        return self["principalId"]
-
-    @property
-    def arn(self) -> str:
-        """The ARN of the principal that made the call.
-        The last section of the ARN contains the user or role that made the call."""
-        return self["arn"]
-
-    @property
-    def session_context(self) -> Optional[Dict[str, Any]]:
-        """ If the request was made with temporary security credentials,
-        this element provides information about the session that was created for those credentials."""
-        return self.get("sessionContext")
-
-
-class S3ObjectLambdaEvent(DictWrapper):
-    """S3 object event notification
-
-    Documentation:
-    -------------
-    - https://docs.aws.amazon.com/AmazonS3/latest/userguide/olap-writing-lambda.html
-    """
-
-    @property
-    def request_id(self) -> str:
-        """The Amazon S3 request ID for this request. We recommend that you log this value to help with debugging."""
-        return self["xAmzRequestId"]
-
-    @property
-    def object_context(self) -> S3ObjectContext:
-        """The input and output details for connections to Amazon S3 and S3 Object Lambda."""
-        return S3ObjectContext(self["getObjectContext"])
-
-    @property
-    def configuration(self) -> S3ObjectConfiguration:
-        """Configuration information about the S3 Object Lambda access point."""
-        return S3ObjectConfiguration(self["configuration"])
-
-    @property
-    def user_request(self) -> S3ObjectUserRequest:
-        """Information about the original call to S3 Object Lambda."""
-        return S3ObjectUserRequest(self["userRequest"])
-
-    @property
-    def user_identity(self) -> S3ObjectUserIdentity:
-        """Details about the identity that made the call to S3 Object Lambda."""
-        return S3ObjectUserIdentity(self["userIdentity"])
-
-    @property
-    def protocol_version(self) -> str:
-        """The version ID of the context provided.
-
-        The format of this field is `{Major Version}`.`{Minor Version}`.
-        The minor version numbers are always two-digit numbers. Any removal or change to the semantics of a
-        field will necessitate a major version bump and will require active opt-in. Amazon S3 can add new
-        fields at any time, at which point you might experience a minor version bump. Due to the nature of
-        software rollouts, it is possible that you might see multiple minor versions in use at once.
-        """
-        return self["protocolVersion"]