Module aws_lambda_powertools.utilities.data_classes.code_pipeline_job_event

Classes

class CodePipelineActionConfiguration (data: Dict[str, Any])

CodePipeline Action Configuration

Expand source code

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

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

Ancestors

• DictWrapper

Instance variables

var configuration : CodePipelineConfiguration

Expand source code

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

Inherited members

• DictWrapper:
  • raw_event

class CodePipelineArtifact (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

Expand source code

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

Ancestors

• DictWrapper

Instance variables

var location : CodePipelineLocation

Expand source code

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

var name : str

Name

Expand source code

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

var revision : Optional[str]

Expand source code

@property
def revision(self) -> Optional[str]:
    return self.get("revision")

Inherited members

• DictWrapper:
  • raw_event

class CodePipelineArtifactCredentials (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

Expand source code

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

Ancestors

• DictWrapper

Instance variables

var access_key_id : str

Expand source code

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

var expiration_time : Optional[int]

Expand source code

@property
def expiration_time(self) -> Optional[int]:
    return self.get("expirationTime")

var secret_access_key : str

Expand source code

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

var session_token : str

Expand source code

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

Inherited members

• DictWrapper:
  • raw_event

class CodePipelineConfiguration (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

Expand source code

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"""
        if self._json_data is None:
            self._json_data = json.loads(self.user_parameters)
        return self._json_data

Ancestors

• DictWrapper

Instance variables

var decoded_user_parameters : Dict[str, Any]

Json Decoded user parameters

Expand source code

@property
def decoded_user_parameters(self) -> Dict[str, Any]:
    """Json Decoded user parameters"""
    if self._json_data is None:
        self._json_data = json.loads(self.user_parameters)
    return self._json_data

var function_name : str

Function name

Expand source code

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

var user_parameters : str

User parameters

Expand source code

@property
def user_parameters(self) -> str:
    """User parameters"""
    return self["UserParameters"]

Inherited members

• DictWrapper:
  • raw_event

class CodePipelineData (data: Dict[str, Any])

CodePipeline Job Data

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

Ancestors

• DictWrapper

Instance variables

var action_configuration : CodePipelineActionConfiguration

CodePipeline action configuration

Expand source code

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

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

var continuation_token : Optional[str]

A continuation token if continuing job

Expand source code

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

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

var 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:
  • 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

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

Ancestors

• DictWrapper

Instance variables

var account_id : str

Account id

Expand source code

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

var data : CodePipelineData

Code pipeline jab data

Expand source code

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

var 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

var get_id : str

Job id

Expand source code

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

var 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

var 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

var user_parameters : str

Action configuration user parameters

Expand source code

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

Methods

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

Expand source code

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

Expand source code

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

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

Expand source code

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,
    )

Inherited members

• DictWrapper:
  • raw_event

class CodePipelineLocation (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

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

Instance variables

var get_type : str

Location type eg: S3

Expand source code

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

var s3_location : CodePipelineS3Location

S3 location

Expand source code

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

Inherited members

• DictWrapper:
  • raw_event

class CodePipelineS3Location (data: Dict[str, Any])

Provides a single read only access to a wrapper dict

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

Instance variables

var bucket_name : str

Expand source code

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

var key : str

Raw S3 object key

Expand source code

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

var 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:
  • raw_event