Module aws_lambda_powertools.utilities.data_classes.code_pipeline_job_event

Classes

class CodePipelineActionConfiguration (data: dict[str, Any], json_deserializer: Callable | None = None)

CodePipeline Action Configuration

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class CodePipelineActionConfiguration(DictWrapper):
    """CodePipeline Action Configuration"""

    @property
    def configuration(self) -> CodePipelineConfiguration:
        return CodePipelineConfiguration(self["configuration"])

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop configuration : CodePipelineConfiguration

Expand source code

@property
def configuration(self) -> CodePipelineConfiguration:
    return CodePipelineConfiguration(self["configuration"])

Inherited members

• DictWrapper:
  • get
  • raw_event

class CodePipelineArtifact (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class CodePipelineArtifact(DictWrapper):
    @property
    def name(self) -> str:
        """Name"""
        return self["name"]

    @property
    def revision(self) -> str | None:
        return self.get("revision")

    @property
    def location(self) -> CodePipelineLocation:
        return CodePipelineLocation(self["location"])

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop location : CodePipelineLocation

Expand source code

@property
def location(self) -> CodePipelineLocation:
    return CodePipelineLocation(self["location"])

prop name : str

Name

Expand source code

@property
def name(self) -> str:
    """Name"""
    return self["name"]

prop revision : str | None

Expand source code

@property
def revision(self) -> str | None:
    return self.get("revision")

Inherited members

• DictWrapper:
  • get
  • raw_event

class CodePipelineArtifactCredentials (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class CodePipelineArtifactCredentials(DictWrapper):
    _sensitive_properties = ["secret_access_key", "session_token"]

    @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) -> int | None:
        return self.get("expirationTime")

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop access_key_id : str

Expand source code

@property
def access_key_id(self) -> str:
    return self["accessKeyId"]

prop expiration_time : int | None

Expand source code

@property
def expiration_time(self) -> int | None:
    return self.get("expirationTime")

prop secret_access_key : str

Expand source code

@property
def secret_access_key(self) -> str:
    return self["secretAccessKey"]

prop session_token : str

Expand source code

@property
def session_token(self) -> str:
    return self["sessionToken"]

Inherited members

• DictWrapper:
  • get
  • raw_event

class CodePipelineConfiguration (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class CodePipelineConfiguration(DictWrapper):
    @property
    def function_name(self) -> str:
        """Function name"""
        return self["FunctionName"]

    @property
    def user_parameters(self) -> str | None:
        """User parameters"""
        return self.get("UserParameters", None)

    @cached_property
    def decoded_user_parameters(self) -> dict[str, Any]:
        """Json Decoded user parameters"""
        if self.user_parameters is not None:
            return self._json_deserializer(self.user_parameters)
        return {}

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

var decoded_user_parameters

Json Decoded user parameters

Expand source code

def __get__(self, instance, owner=None):
    if instance is None:
        return self
    if self.attrname is None:
        raise TypeError(
            "Cannot use cached_property instance without calling __set_name__ on it.")
    try:
        cache = instance.__dict__
    except AttributeError:  # not all objects have __dict__ (e.g. class defines slots)
        msg = (
            f"No '__dict__' attribute on {type(instance).__name__!r} "
            f"instance to cache {self.attrname!r} property."
        )
        raise TypeError(msg) from None
    val = cache.get(self.attrname, _NOT_FOUND)
    if val is _NOT_FOUND:
        val = self.func(instance)
        try:
            cache[self.attrname] = val
        except TypeError:
            msg = (
                f"The '__dict__' attribute on {type(instance).__name__!r} instance "
                f"does not support item assignment for caching {self.attrname!r} property."
            )
            raise TypeError(msg) from None
    return val

prop function_name : str

Function name

Expand source code

@property
def function_name(self) -> str:
    """Function name"""
    return self["FunctionName"]

prop user_parameters : str | None

User parameters

Expand source code

@property
def user_parameters(self) -> str | None:
    """User parameters"""
    return self.get("UserParameters", None)

Inherited members

• DictWrapper:
  • get
  • raw_event

class CodePipelineData (data: dict[str, Any], json_deserializer: Callable | None = None)

CodePipeline Job Data

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

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) -> str | None:
        """A continuation token if continuing job"""
        return self.get("continuationToken")

    @property
    def encryption_key(self) -> CodePipelineEncryptionKey | None:
        """Represents a CodePipeline encryption key"""
        key_data = self.get("encryptionKey")
        return CodePipelineEncryptionKey(key_data) if key_data is not None else None

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop action_configuration : CodePipelineActionConfiguration

CodePipeline action configuration

Expand source code

@property
def action_configuration(self) -> CodePipelineActionConfiguration:
    """CodePipeline action configuration"""
    return CodePipelineActionConfiguration(self["actionConfiguration"])

prop artifact_credentials : CodePipelineArtifactCredentials

Represents a CodePipeline artifact credentials

Expand source code

@property
def artifact_credentials(self) -> CodePipelineArtifactCredentials:
    """Represents a CodePipeline artifact credentials"""
    return CodePipelineArtifactCredentials(self["artifactCredentials"])

prop continuation_token : str | None

A continuation token if continuing job

Expand source code

@property
def continuation_token(self) -> str | None:
    """A continuation token if continuing job"""
    return self.get("continuationToken")

prop encryption_key : CodePipelineEncryptionKey | None

Represents a CodePipeline encryption key

Expand source code

@property
def encryption_key(self) -> CodePipelineEncryptionKey | None:
    """Represents a CodePipeline encryption key"""
    key_data = self.get("encryptionKey")
    return CodePipelineEncryptionKey(key_data) if key_data is not None else None

prop input_artifacts : list[CodePipelineArtifact]

Represents a CodePipeline input artifact

Expand source code

@property
def input_artifacts(self) -> list[CodePipelineArtifact]:
    """Represents a CodePipeline input artifact"""
    return [CodePipelineArtifact(item) for item in self["inputArtifacts"]]

prop output_artifacts : list[CodePipelineArtifact]

Represents a CodePipeline output artifact

Expand source code

@property
def output_artifacts(self) -> list[CodePipelineArtifact]:
    """Represents a CodePipeline output artifact"""
    return [CodePipelineArtifact(item) for item in self["outputArtifacts"]]

Inherited members

• DictWrapper:
  • get
  • raw_event

class CodePipelineEncryptionKey (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

class CodePipelineEncryptionKey(DictWrapper):
    @property
    def get_id(self) -> str:
        return self["id"]

    @property
    def get_type(self) -> str:
        return self["type"]

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop get_id : str

Expand source code

@property
def get_id(self) -> str:
    return self["id"]

prop get_type : str

Expand source code

@property
def get_type(self) -> str:
    return self["type"]

Inherited members

• DictWrapper:
  • get
  • raw_event

class CodePipelineJobEvent (data: dict[str, Any])

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

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

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 | None:
        """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
        """
        # IMPORTING boto3 within the FUNCTION and not at the top level to get
        # it only when we explicitly want it for better performance.
        import boto3

        from aws_lambda_powertools.shared import user_agent

        s3 = 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,
        )
        user_agent.register_feature_to_client(client=s3, feature="data_classes")
        return s3

    def find_input_artifact(self, artifact_name: str) -> CodePipelineArtifact | None:
        """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) -> str | None:
        """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")

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop account_id : str

Account id

Expand source code

@property
def account_id(self) -> str:
    """Account id"""
    return self._job["accountId"]

prop data : CodePipelineData

Code pipeline jab data

Expand source code

@property
def data(self) -> CodePipelineData:
    """Code pipeline jab data"""
    return CodePipelineData(self._job["data"])

prop decoded_user_parameters : dict[str, Any]

Json Decoded action configuration user parameters

Expand source code

@property
def decoded_user_parameters(self) -> dict[str, Any]:
    """Json Decoded action configuration user parameters"""
    return self.data.action_configuration.configuration.decoded_user_parameters

prop get_id : str

Job id

Expand source code

@property
def get_id(self) -> str:
    """Job id"""
    return self._job["id"]

prop input_bucket_name : str

Get the first input artifact bucket name

Expand source code

@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

prop input_object_key : str

Get the first input artifact order key unquote plus

Expand source code

@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

prop user_parameters : str | None

Action configuration user parameters

Expand source code

@property
def user_parameters(self) -> str | None:
    """Action configuration user parameters"""
    return self.data.action_configuration.configuration.user_parameters

Methods

def find_input_artifact(self, artifact_name: str) ‑> CodePipelineArtifact | None

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

def get_artifact(self, artifact_name: str, filename: str) ‑> str | None

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

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

Inherited members

• DictWrapper:
  • get
  • raw_event

class CodePipelineLocation (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

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"])

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop get_type : str

Location type eg: S3

Expand source code

@property
def get_type(self) -> str:
    """Location type eg: S3"""
    return self["type"]

prop s3_location : CodePipelineS3Location

S3 location

Expand source code

@property
def s3_location(self) -> CodePipelineS3Location:
    """S3 location"""
    return CodePipelineS3Location(self["s3Location"])

Inherited members

• DictWrapper:
  • get
  • raw_event

class CodePipelineS3Location (data: dict[str, Any], json_deserializer: Callable | None = None)

Provides a single read only access to a wrapper dict

Parameters

data : dict[str, Any]

Lambda Event Source Event payload

json_deserializer : Callable, optional

function to deserialize str, bytes, bytearray containing a JSON document to a Python obj, by default json.loads

Expand source code

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"])

Ancestors

• DictWrapper
• collections.abc.Mapping
• collections.abc.Collection
• collections.abc.Sized
• collections.abc.Iterable
• collections.abc.Container
• typing.Generic

Instance variables

prop bucket_name : str

Expand source code

@property
def bucket_name(self) -> str:
    return self["bucketName"]

prop key : str

Raw S3 object key

Expand source code

@property
def key(self) -> str:
    """Raw S3 object key"""
    return self["objectKey"]

prop object_key : str

Unquote plus of the S3 object key

Expand source code

@property
def object_key(self) -> str:
    """Unquote plus of the S3 object key"""
    return unquote_plus(self["objectKey"])

Inherited members

• DictWrapper:
  • get
  • raw_event