Module aws_lambda_powertools.utilities.data_classes.code_pipeline_job_event

Expand source code 
import tempfile
import zipfile
from typing import Any, Dict, List, Optional
from urllib.parse import unquote_plus

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) -> Optional[str]:
        """User parameters"""
        return self.get("UserParameters", None)

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


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):
    _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) -> Optional[int]:
        return self.get("expirationTime")


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

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


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

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


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) -> Optional[str]:
        """Action configuration user parameters"""
        return self.data.action_configuration.configuration.user_parameters

    @property
    def decoded_user_parameters(self) -> Optional[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) -> 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")

Classes

class CodePipelineActionConfiguration (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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

Instance variables

var configuration : CodePipelineConfiguration
Expand source code 
@property
def configuration(self) -> CodePipelineConfiguration:
    return CodePipelineConfiguration(self["configuration"])

Inherited members

class CodePipelineArtifact (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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) -> Optional[str]:
        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

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

class CodePipelineArtifactCredentials (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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) -> Optional[int]:
        return self.get("expirationTime")

Ancestors

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

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

class CodePipelineConfiguration (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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) -> Optional[str]:
        """User parameters"""
        return self.get("UserParameters", None)

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

Ancestors

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

Instance variables

var decoded_user_parameters : Optional[Dict[str, Any]]

Json Decoded user parameters

Expand source code 
@property
def decoded_user_parameters(self) -> Optional[Dict[str, Any]]:
    """Json Decoded user parameters"""
    if self._json_data is None and self.user_parameters is not None:
        self._json_data = self._json_deserializer(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 : Optional[str]

User parameters

Expand source code 
@property
def user_parameters(self) -> Optional[str]:
    """User parameters"""
    return self.get("UserParameters", None)

Inherited members

class CodePipelineData (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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) -> Optional[str]:
        """A continuation token if continuing job"""
        return self.get("continuationToken")

    @property
    def encryption_key(self) -> Optional[CodePipelineEncryptionKey]:
        """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

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 encryption_key : Optional[CodePipelineEncryptionKey]

Represents a CodePipeline encryption key

Expand source code 
@property
def encryption_key(self) -> Optional[CodePipelineEncryptionKey]:
    """Represents a CodePipeline encryption key"""
    key_data = self.get("encryptionKey")
    return CodePipelineEncryptionKey(key_data) if key_data is not None else None
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

class CodePipelineEncryptionKey (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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

Instance variables

var get_id : str
Expand source code 
@property
def get_id(self) -> str:
    return self["id"]
var get_type : str
Expand source code 
@property
def get_type(self) -> str:
    return self["type"]

Inherited members

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

AWS CodePipeline Job Event

Documentation:

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) -> Optional[str]:
        """Action configuration user parameters"""
        return self.data.action_configuration.configuration.user_parameters

    @property
    def decoded_user_parameters(self) -> Optional[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) -> 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
  • collections.abc.Mapping
  • collections.abc.Collection
  • collections.abc.Sized
  • collections.abc.Iterable
  • collections.abc.Container

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 : Optional[Dict[str, Any]]

Json Decoded action configuration user parameters

Expand source code 
@property
def decoded_user_parameters(self) -> Optional[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 : Optional[str]

Action configuration user parameters

Expand source code 
@property
def user_parameters(self) -> Optional[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
    """
    # 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

Inherited members

class CodePipelineLocation (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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

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

class CodePipelineS3Location (data: Dict[str, Any], json_deserializer: Optional[Callable] = 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

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