diff --git a/aws_lambda_powertools/utilities/data_classes/__init__.py b/aws_lambda_powertools/utilities/data_classes/__init__.py
index 58464ebcf99..a47c32ee07f 100644
--- a/aws_lambda_powertools/utilities/data_classes/__init__.py
+++ b/aws_lambda_powertools/utilities/data_classes/__init__.py
@@ -6,6 +6,7 @@
 from .api_gateway_proxy_event import APIGatewayProxyEvent, APIGatewayProxyEventV2
 from .appsync_resolver_event import AppSyncResolverEvent
 from .cloud_watch_logs_event import CloudWatchLogsEvent
+from .code_pipeline_job_event import CodePipelineJobEvent
 from .connect_contact_flow_event import ConnectContactFlowEvent
 from .dynamo_db_stream_event import DynamoDBStreamEvent
 from .event_bridge_event import EventBridgeEvent
@@ -21,6 +22,7 @@
     "AppSyncResolverEvent",
     "ALBEvent",
     "CloudWatchLogsEvent",
+    "CodePipelineJobEvent",
     "ConnectContactFlowEvent",
     "DynamoDBStreamEvent",
     "EventBridgeEvent",
diff --git a/aws_lambda_powertools/utilities/data_classes/code_pipeline_job_event.py b/aws_lambda_powertools/utilities/data_classes/code_pipeline_job_event.py
new file mode 100644
index 00000000000..e13d32fb169
--- /dev/null
+++ b/aws_lambda_powertools/utilities/data_classes/code_pipeline_job_event.py
@@ -0,0 +1,236 @@
+import json
+import tempfile
+import zipfile
+from typing import Any, Dict, List, Optional
+from urllib.parse import unquote_plus
+
+import boto3
+
+from aws_lambda_powertools.utilities.data_classes.common import DictWrapper
+
+
+class CodePipelineConfiguration(DictWrapper):
+    @property
+    def function_name(self) -> str:
+        """Function name"""
+        return self["FunctionName"]
+
+    @property
+    def user_parameters(self) -> str:
+        """User parameters"""
+        return self["UserParameters"]
+
+    @property
+    def decoded_user_parameters(self) -> Dict[str, Any]:
+        """Json Decoded user parameters"""
+        return json.loads(self.user_parameters)
+
+
+class CodePipelineActionConfiguration(DictWrapper):
+    """CodePipeline Action Configuration"""
+
+    @property
+    def configuration(self) -> CodePipelineConfiguration:
+        return CodePipelineConfiguration(self["configuration"])
+
+
+class CodePipelineS3Location(DictWrapper):
+    @property
+    def bucket_name(self) -> str:
+        return self["bucketName"]
+
+    @property
+    def key(self) -> str:
+        """Raw S3 object key"""
+        return self["objectKey"]
+
+    @property
+    def object_key(self) -> str:
+        """Unquote plus of the S3 object key"""
+        return unquote_plus(self["objectKey"])
+
+
+class CodePipelineLocation(DictWrapper):
+    @property
+    def get_type(self) -> str:
+        """Location type eg: S3"""
+        return self["type"]
+
+    @property
+    def s3_location(self) -> CodePipelineS3Location:
+        """S3 location"""
+        return CodePipelineS3Location(self["s3Location"])
+
+
+class CodePipelineArtifact(DictWrapper):
+    @property
+    def name(self) -> str:
+        """Name"""
+        return self["name"]
+
+    @property
+    def revision(self) -> Optional[str]:
+        return self.get("revision")
+
+    @property
+    def location(self) -> CodePipelineLocation:
+        return CodePipelineLocation(self["location"])
+
+
+class CodePipelineArtifactCredentials(DictWrapper):
+    @property
+    def access_key_id(self) -> str:
+        return self["accessKeyId"]
+
+    @property
+    def secret_access_key(self) -> str:
+        return self["secretAccessKey"]
+
+    @property
+    def session_token(self) -> str:
+        return self["sessionToken"]
+
+    @property
+    def expiration_time(self) -> Optional[int]:
+        return self.get("expirationTime")
+
+
+class CodePipelineData(DictWrapper):
+    """CodePipeline Job Data"""
+
+    @property
+    def action_configuration(self) -> CodePipelineActionConfiguration:
+        """CodePipeline action configuration"""
+        return CodePipelineActionConfiguration(self["actionConfiguration"])
+
+    @property
+    def input_artifacts(self) -> List[CodePipelineArtifact]:
+        """Represents a CodePipeline input artifact"""
+        return [CodePipelineArtifact(item) for item in self["inputArtifacts"]]
+
+    @property
+    def output_artifacts(self) -> List[CodePipelineArtifact]:
+        """Represents a CodePipeline output artifact"""
+        return [CodePipelineArtifact(item) for item in self["outputArtifacts"]]
+
+    @property
+    def artifact_credentials(self) -> CodePipelineArtifactCredentials:
+        """Represents a CodePipeline artifact credentials"""
+        return CodePipelineArtifactCredentials(self["artifactCredentials"])
+
+    @property
+    def continuation_token(self) -> Optional[str]:
+        """A continuation token if continuing job"""
+        return self.get("continuationToken")
+
+
+class CodePipelineJobEvent(DictWrapper):
+    """AWS CodePipeline Job Event
+
+    Documentation:
+    -------------
+    - https://docs.aws.amazon.com/codepipeline/latest/userguide/actions-invoke-lambda-function.html
+    - https://docs.aws.amazon.com/lambda/latest/dg/services-codepipeline.html
+    """
+
+    def __init__(self, data: Dict[str, Any]):
+        super().__init__(data)
+        self._job = self["CodePipeline.job"]
+
+    @property
+    def get_id(self) -> str:
+        """Job id"""
+        return self._job["id"]
+
+    @property
+    def account_id(self) -> str:
+        """Account id"""
+        return self._job["accountId"]
+
+    @property
+    def data(self) -> CodePipelineData:
+        """Code pipeline jab data"""
+        return CodePipelineData(self._job["data"])
+
+    @property
+    def user_parameters(self) -> str:
+        """Action configuration user parameters"""
+        return self.data.action_configuration.configuration.user_parameters
+
+    @property
+    def decoded_user_parameters(self) -> Dict[str, Any]:
+        """Json Decoded action configuration user parameters"""
+        return self.data.action_configuration.configuration.decoded_user_parameters
+
+    @property
+    def input_bucket_name(self) -> str:
+        """Get the first input artifact bucket name"""
+        return self.data.input_artifacts[0].location.s3_location.bucket_name
+
+    @property
+    def input_object_key(self) -> str:
+        """Get the first input artifact order key unquote plus"""
+        return self.data.input_artifacts[0].location.s3_location.object_key
+
+    def setup_s3_client(self):
+        """Creates an S3 client
+
+        Uses the credentials passed in the event by CodePipeline. These
+        credentials can be used to access the artifact bucket.
+
+        Returns
+        -------
+        BaseClient
+            An S3 client with the appropriate credentials
+        """
+        return boto3.client(
+            "s3",
+            aws_access_key_id=self.data.artifact_credentials.access_key_id,
+            aws_secret_access_key=self.data.artifact_credentials.secret_access_key,
+            aws_session_token=self.data.artifact_credentials.session_token,
+        )
+
+    def find_input_artifact(self, artifact_name: str) -> Optional[CodePipelineArtifact]:
+        """Find an input artifact by artifact name
+
+        Parameters
+        ----------
+        artifact_name : str
+            The name of the input artifact to look for
+
+        Returns
+        -------
+        CodePipelineArtifact, None
+            Matching CodePipelineArtifact if found
+        """
+        for artifact in self.data.input_artifacts:
+            if artifact.name == artifact_name:
+                return artifact
+        return None
+
+    def get_artifact(self, artifact_name: str, filename: str) -> Optional[str]:
+        """Get a file within an artifact zip on s3
+
+        Parameters
+        ----------
+        artifact_name : str
+            Name of the S3 artifact to download
+        filename : str
+            The file name within the artifact zip to extract as a string
+
+        Returns
+        -------
+        str, None
+            Returns the contents file contents as a string
+        """
+        artifact = self.find_input_artifact(artifact_name)
+        if artifact is None:
+            return None
+
+        with tempfile.NamedTemporaryFile() as tmp_file:
+            s3 = self.setup_s3_client()
+            bucket = artifact.location.s3_location.bucket_name
+            key = artifact.location.s3_location.key
+            s3.download_file(bucket, key, tmp_file.name)
+            with zipfile.ZipFile(tmp_file.name, "r") as zip_file:
+                return zip_file.read(filename).decode("UTF-8")
diff --git a/docs/utilities/data_classes.md b/docs/utilities/data_classes.md
index dc56ed8ec41..0fc33d3a3f7 100644
--- a/docs/utilities/data_classes.md
+++ b/docs/utilities/data_classes.md
@@ -52,6 +52,7 @@ Event Source | Data_class
 [API Gateway Proxy event v2](#api-gateway-proxy-v2) | `APIGatewayProxyEventV2`
 [AppSync Resolver](#appsync-resolver) | `AppSyncResolverEvent`
 [CloudWatch Logs](#cloudwatch-logs) | `CloudWatchLogsEvent`
+[CodePipeline Job Event](#codepipeline-job) | `CodePipelineJobEvent`
 [Cognito User Pool](#cognito-user-pool) | Multiple available under `cognito_user_pool_event`
 [Connect Contact Flow](#connect-contact-flow) | `ConnectContactFlowEvent`
 [DynamoDB streams](#dynamodb-streams) | `DynamoDBStreamEvent`, `DynamoDBRecordEventName`
@@ -222,6 +223,58 @@ decompress and parse json data from the event.
             do_something_with(event.timestamp, event.message)
     ```
 
+### CodePipeline Job
+
+Data classes and utility functions to help create continuous delivery pipelines tasks with AWS Lambda
+
+=== "app.py"
+
+    ```python
+    from aws_lambda_powertools import Logger
+    from aws_lambda_powertools.utilities.data_classes import CodePipelineJobEvent
+
+    logger = Logger()
+
+
+    def lambda_handler(event, context):
+        """The Lambda function handler
+
+        If a continuing job then checks the CloudFormation stack status
+        and updates the job accordingly.
+
+        If a new job then kick of an update or creation of the target
+        CloudFormation stack.
+        """
+        event: CodePipelineJobEvent = CodePipelineJobEvent(event)
+
+        # Extract the Job ID
+        job_id = event.get_id
+
+        # Extract the params
+        params: dict = event.decoded_user_parameters
+        stack = params["stack"]
+        artifact_name = params["artifact"]
+        template_file = params["file"]
+
+        try:
+            if event.data.continuation_token:
+                # If we're continuing then the create/update has already been triggered
+                # we just need to check if it has finished.
+                check_stack_update_status(job_id, stack)
+            else:
+                template = event.get_artifact(artifact_name, template_file)
+                # Kick off a stack update or create
+                start_update_or_create(job_id, stack, template)
+        except Exception as e:
+            # If any other exceptions which we didn't expect are raised
+            # then fail the job and log the exception message.
+            logger.exception("Function failed due to exception.")
+            put_job_failure(job_id, "Function exception: " + str(e))
+
+        logger.debug("Function complete.")
+        return "Complete."
+    ```
+
 ### Cognito User Pool
 
 Cognito User Pools have several [different Lambda trigger sources](https://docs.aws.amazon.com/cognito/latest/developerguide/cognito-user-identity-pools-working-with-aws-lambda-triggers.html#cognito-user-identity-pools-working-with-aws-lambda-trigger-sources), all of which map to a different data class, which
diff --git a/tests/events/codePipelineEvent.json b/tests/events/codePipelineEvent.json
new file mode 100644
index 00000000000..d7abe513468
--- /dev/null
+++ b/tests/events/codePipelineEvent.json
@@ -0,0 +1,34 @@
+{
+    "CodePipeline.job": {
+        "id": "11111111-abcd-1111-abcd-111111abcdef",
+        "accountId": "111111111111",
+        "data": {
+            "actionConfiguration": {
+                "configuration": {
+                    "FunctionName": "MyLambdaFunctionForAWSCodePipeline",
+                    "UserParameters": "some-input-such-as-a-URL"
+                }
+            },
+            "inputArtifacts": [
+                {
+                    "name": "ArtifactName",
+                    "revision": null,
+                    "location": {
+                        "type": "S3",
+                        "s3Location": {
+                            "bucketName": "the name of the bucket configured as the pipeline artifact store in Amazon S3, for example codepipeline-us-east-2-1234567890",
+                            "objectKey": "the name of the application, for example CodePipelineDemoApplication.zip"
+                        }
+                    }
+                }
+            ],
+            "outputArtifacts": [],
+            "artifactCredentials": {
+                "accessKeyId": "AKIAIOSFODNN7EXAMPLE",
+                "secretAccessKey": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
+                "sessionToken": "MIICiTCCAfICCQD6m7oRw0uXOjANBgkqhkiG9w0BAQUFADCBiDELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAldBMRAwDgYDVQQHEwdTZWF0dGxlMQ8wDQYDVQQKEwZBbWF6b24xFDASBgNVBAsTC0lBTSBDb25zb2xlMRIwEAYDVQQDEwlUZXN0Q2lsYWMxHzAdBgkqhkiG9w0BCQEWEG5vb25lQGFtYXpvbi5jb20wHhcNMTEwNDI1MjA0NTIxWhcNMTIwNDI0MjA0NTIxWjCBiDELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAldBMRAwDgYDVQQHEwdTZWF0dGxlMQ8wDQYDVQQKEwZBbWF6b24xFDASBgNVBAsTC0lBTSBDb25zb2xlMRIwEAYDVQQDEwlUZXN0Q2lsYWMxHzAdBgkqhkiG9w0BCQEWEG5vb25lQGFtYXpvbi5jb20wgZ8wDQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBAMaK0dn+a4GmWIWJ21uUSfwfEvySWtC2XADZ4nB+BLYgVIk60CpiwsZ3G93vUEIO3IyNoH/f0wYK8m9TrDHudUZg3qX4waLG5M43q7Wgc/MbQITxOUSQv7c7ugFFDzQGBzZswY6786m86gpEIbb3OhjZnzcvQAaRHhdlQWIMm2nrAgMBAAEwDQYJKoZIhvcNAQEFBQADgYEAtCu4nUhVVxYUntneD9+h8Mg9q6q+auNKyExzyLwaxlAoo7TJHidbtS4J5iNmZgXL0FkbFFBjvSfpJIlJ00zbhNYS5f6GuoEDmFJl0ZxBHjJnyp378OD8uTs7fLvjx79LjSTbNYiytVbZPQUQ5Yaxu2jXnimvw3rrszlaEXAMPLE="
+            },
+            "continuationToken": "A continuation token if continuing job"
+        }
+    }
+}
diff --git a/tests/events/codePipelineEventData.json b/tests/events/codePipelineEventData.json
new file mode 100644
index 00000000000..7552f19ca93
--- /dev/null
+++ b/tests/events/codePipelineEventData.json
@@ -0,0 +1,46 @@
+{
+    "CodePipeline.job": {
+        "id": "c0d76431-b0e7-xmpl-97e3-e8ee786eb6f6",
+        "accountId": "123456789012",
+        "data": {
+            "actionConfiguration": {
+                "configuration": {
+                    "FunctionName": "my-function",
+                    "UserParameters": "{\"KEY\": \"VALUE\"}"
+                }
+            },
+            "inputArtifacts": [
+                {
+                    "name": "my-pipeline-SourceArtifact",
+                    "revision": "e0c7xmpl2308ca3071aa7bab414de234ab52eea",
+                    "location": {
+                        "type": "S3",
+                        "s3Location": {
+                            "bucketName": "us-west-2-123456789012-my-pipeline",
+                            "objectKey": "my-pipeline/test-api-2/TdOSFRV"
+                        }
+                    }
+                }
+            ],
+            "outputArtifacts": [
+                {
+                    "name": "invokeOutput",
+                    "revision": null,
+                    "location": {
+                        "type": "S3",
+                        "s3Location": {
+                            "bucketName": "us-west-2-123456789012-my-pipeline",
+                            "objectKey": "my-pipeline/invokeOutp/D0YHsJn"
+                        }
+                    }
+                }
+            ],
+            "artifactCredentials": {
+                "accessKeyId": "AKIAIOSFODNN7EXAMPLE",
+                "secretAccessKey": "6CGtmAa3lzWtV7a...",
+                "sessionToken": "IQoJb3JpZ2luX2VjEA...",
+                "expirationTime": 1575493418000
+            }
+        }
+    }
+}
diff --git a/tests/functional/test_data_classes.py b/tests/functional/test_data_classes.py
index d346eca480a..8a18df0e0f8 100644
--- a/tests/functional/test_data_classes.py
+++ b/tests/functional/test_data_classes.py
@@ -1,15 +1,19 @@
 import base64
 import datetime
 import json
+import zipfile
 from secrets import compare_digest
 from urllib.parse import quote_plus
 
+from pytest_mock import MockerFixture
+
 from aws_lambda_powertools.utilities.data_classes import (
     ALBEvent,
     APIGatewayProxyEvent,
     APIGatewayProxyEventV2,
     AppSyncResolverEvent,
     CloudWatchLogsEvent,
+    CodePipelineJobEvent,
     EventBridgeEvent,
     KinesisStreamEvent,
     S3Event,
@@ -31,6 +35,7 @@
     AppSyncResolverEventInfo,
     get_identity_object,
 )
+from aws_lambda_powertools.utilities.data_classes.code_pipeline_job_event import CodePipelineData
 from aws_lambda_powertools.utilities.data_classes.cognito_user_pool_event import (
     CreateAuthChallengeTriggerEvent,
     CustomMessageTriggerEvent,
@@ -127,11 +132,11 @@ def test_cognito_pre_signup_trigger_event():
 
     # Verify setters
     event.response.auto_confirm_user = True
-    assert event.response.auto_confirm_user is True
+    assert event.response.auto_confirm_user
     event.response.auto_verify_phone = True
-    assert event.response.auto_verify_phone is True
+    assert event.response.auto_verify_phone
     event.response.auto_verify_email = True
-    assert event.response.auto_verify_email is True
+    assert event.response.auto_verify_email
     assert event["response"]["autoVerifyEmail"] is True
 
 
@@ -167,7 +172,7 @@ def test_cognito_user_migration_trigger_event():
     event.response.message_action = "SUPPRESS"
     assert event.response.message_action == "SUPPRESS"
     event.response.force_alias_creation = True
-    assert event.response.force_alias_creation is True
+    assert event.response.force_alias_creation
     event.response.desired_delivery_mediums = ["EMAIL"]
     assert event.response.desired_delivery_mediums == ["EMAIL"]
 
@@ -284,10 +289,10 @@ def test_cognito_define_auth_challenge_trigger_event():
     assert event.response.challenge_name == event["response"]["challengeName"]
     assert event.response.challenge_name == "CUSTOM_CHALLENGE"
     event.response.fail_authentication = True
-    assert event.response.fail_authentication is True
+    assert event.response.fail_authentication
     assert event.response.fail_authentication == event["response"]["failAuthentication"]
     event.response.issue_tokens = True
-    assert event.response.issue_tokens is True
+    assert event.response.issue_tokens
     assert event.response.issue_tokens == event["response"]["issueTokens"]
 
 
@@ -334,7 +339,7 @@ def test_verify_auth_challenge_response_trigger_event():
     # Verify setters
     event.response.answer_correct = True
     assert event.response.answer_correct == event["response"]["answerCorrect"]
-    assert event.response.answer_correct is True
+    assert event.response.answer_correct
 
 
 def test_connect_contact_flow_event_min():
@@ -1106,3 +1111,95 @@ def test_format_time_negative():
     datetime_str = _formatted_time(now, "%H:%M:%S", -12)
     assert isinstance(datetime_str, str)
     assert datetime_str == "02:22:33-12:00:00"
+
+
+def test_code_pipeline_event():
+    event = CodePipelineJobEvent(load_event("codePipelineEvent.json"))
+
+    job = event["CodePipeline.job"]
+    assert job["id"] == event.get_id
+    assert job["accountId"] == event.account_id
+
+    data = event.data
+    assert isinstance(data, CodePipelineData)
+    assert job["data"]["continuationToken"] == data.continuation_token
+    configuration = data.action_configuration.configuration
+    assert "MyLambdaFunctionForAWSCodePipeline" == configuration.function_name
+    assert event.user_parameters == configuration.user_parameters
+    assert "some-input-such-as-a-URL" == configuration.user_parameters
+
+    input_artifacts = data.input_artifacts
+    assert len(input_artifacts) == 1
+    assert "ArtifactName" == input_artifacts[0].name
+    assert input_artifacts[0].revision is None
+    assert "S3" == input_artifacts[0].location.get_type
+
+    output_artifacts = data.output_artifacts
+    assert len(output_artifacts) == 0
+
+    artifact_credentials = data.artifact_credentials
+    artifact_credentials_dict = event["CodePipeline.job"]["data"]["artifactCredentials"]
+    assert artifact_credentials_dict["accessKeyId"] == artifact_credentials.access_key_id
+    assert artifact_credentials_dict["secretAccessKey"] == artifact_credentials.secret_access_key
+    assert artifact_credentials_dict["sessionToken"] == artifact_credentials.session_token
+
+
+def test_code_pipeline_event_decoded_data():
+    event = CodePipelineJobEvent(load_event("codePipelineEventData.json"))
+
+    assert event.data.continuation_token is None
+    decoded_params = event.data.action_configuration.configuration.decoded_user_parameters
+    assert decoded_params == event.decoded_user_parameters
+    assert "VALUE" == decoded_params["KEY"]
+
+    assert "my-pipeline-SourceArtifact" == event.data.input_artifacts[0].name
+
+    output_artifacts = event.data.output_artifacts
+    assert len(output_artifacts) == 1
+    assert "S3" == output_artifacts[0].location.get_type
+    assert "my-pipeline/invokeOutp/D0YHsJn" == output_artifacts[0].location.s3_location.key
+
+    artifact_credentials = event.data.artifact_credentials
+    artifact_credentials_dict = event["CodePipeline.job"]["data"]["artifactCredentials"]
+    assert isinstance(artifact_credentials.expiration_time, int)
+    assert artifact_credentials_dict["expirationTime"] == artifact_credentials.expiration_time
+
+    assert "us-west-2-123456789012-my-pipeline" == event.input_bucket_name
+    assert "my-pipeline/test-api-2/TdOSFRV" == event.input_object_key
+
+
+def test_code_pipeline_get_artifact_not_found():
+    event = CodePipelineJobEvent(load_event("codePipelineEventData.json"))
+
+    assert event.find_input_artifact("not-found") is None
+    assert event.get_artifact("not-found", "foo") is None
+
+
+def test_code_pipeline_get_artifact(mocker: MockerFixture):
+    filename = "foo.json"
+    file_contents = "Foo"
+
+    class MockClient:
+        @staticmethod
+        def download_file(bucket: str, key: str, tmp_name: str):
+            assert bucket == "us-west-2-123456789012-my-pipeline"
+            assert key == "my-pipeline/test-api-2/TdOSFRV"
+            with zipfile.ZipFile(tmp_name, mode="w", compression=zipfile.ZIP_DEFLATED) as zf:
+                zf.writestr(filename, file_contents)
+
+    s3 = mocker.patch("boto3.client")
+    s3.return_value = MockClient()
+
+    event = CodePipelineJobEvent(load_event("codePipelineEventData.json"))
+
+    artifact_str = event.get_artifact(artifact_name="my-pipeline-SourceArtifact", filename=filename)
+
+    s3.assert_called_once_with(
+        "s3",
+        **{
+            "aws_access_key_id": event.data.artifact_credentials.access_key_id,
+            "aws_secret_access_key": event.data.artifact_credentials.secret_access_key,
+            "aws_session_token": event.data.artifact_credentials.session_token,
+        }
+    )
+    assert artifact_str == file_contents