CodePipelineBuilder

Bases: AbstractAWSResourceBuilder['CodePipelineBuilder', CodePipelineConfig]

Builder for creating AWS CodePipeline workflows with organizational standards.

Creates production-ready CI/CD pipelines with support for different usage patterns (BUILD_RELEASE, BUILD_DEPLOY, DEPLOY), git integration, artifact management, EventBridge automation, and comprehensive IAM role configuration.

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

class CodePipelineBuilder(AbstractAWSResourceBuilder["CodePipelineBuilder", CodePipelineConfig]):
    """Builder for creating AWS CodePipeline workflows with organizational standards.

    Creates production-ready CI/CD pipelines with support for different usage patterns
    (BUILD_RELEASE, BUILD_DEPLOY, DEPLOY), git integration, artifact management,
    EventBridge automation, and comprehensive IAM role configuration.
    """

    _resource_type : AWSResourceType = AWSResourceType.CODEPIPELINE
    _requires_usage: bool = True

    def reset(self) -> None:
        """Reset the builder to its initial state.

        Side Effects:
            Resets internal builder state via parent class
        """
        super().reset()
        self._orchestrator = PipelineOrchestrator(self)
        self._action_factory = ActionFactory(self)

    def build(self, scope: Construct) -> codepipeline.IPipeline:
        """Build and return the configured CodePipeline.

        Creates pipeline with base configuration, builds stages according to usage pattern,
        and applies organizational tags.

        Args:
            scope: CDK construct scope where the pipeline will be created

        Returns:
            Configured CodePipeline instance

        Raises:
            ValidationError: If pipeline configuration validation fails

        Side Effects:
            Creates CodePipeline with all configured stages and actions
            Applies organizational tags to the pipeline
        """
        super().build()

        pipeline = self._create_base_pipeline(scope)
        self._build_pipeline_stages(scope, pipeline)
        self._tag_resource(pipeline)

        return pipeline

    def _build_pipeline_stages(self, scope: Construct, pipeline: codepipeline.Pipeline) -> None:
        """Build pipeline stages based on usage configuration.

        Args:
            scope: CDK construct scope
            pipeline: Pipeline to configure with stages

        Side Effects:
            Delegates to orchestrator to create appropriate stage configuration
        """
        match self._config.usage:
            case Usage.BUILD_RELEASE:
                self._orchestrator.create_build_release_pipeline(scope, pipeline)
            case Usage.BUILD_DEPLOY:
                self._orchestrator.create_build_deploy_pipeline(scope, pipeline)
            case Usage.DEPLOY:
                self._orchestrator.create_deploy_pipeline(scope, pipeline)
            case Usage.BUILD_ASSESS:
                self._orchestrator.create_build_assess_pipeline(scope, pipeline)

    def _create_base_pipeline(self, scope: Construct) -> codepipeline.Pipeline:
        """Create base CodePipeline with core configuration.

        Args:
            scope: CDK construct scope

        Returns:
            Basic pipeline with role and configuration, without stages

        Side Effects:
            Creates IAM role for pipeline execution
        """
        pipeline_id, pipeline_name = self._generate_pipeline_id_and_name()
        pipeline_role = self._create_pipeline_role(scope)

        return codepipeline.Pipeline(
            scope, pipeline_id,
            pipeline_name=pipeline_name,
            pipeline_type=codepipeline.PipelineType.V2,
            role=pipeline_role,
            restart_execution_on_update=self._config.restart_execution_on_update,
            cross_account_keys=True,
            artifact_bucket=self._config.artifact_bucket
        )

    def _create_source_stage(self, pipeline: codepipeline.Pipeline) -> codepipeline.Artifact:
        """Create source stage for git repository integration.

        Args:
            pipeline: Pipeline to add source stage to

        Returns:
            Source output artifact for downstream stages

        Side Effects:
            Adds source stage to pipeline with CodeCommit or configured action
        """
        source_output = codepipeline.Artifact(
            self._get_name_for_resource("Source-Output", max_length=AWSResourceNameLength.ARTIFACT.value)
        )

        stage_config = getattr(self._config.stages, StageType.SOURCE.value, None)

        if stage_config is None:
            # Default CodeCommit source action
            action = self._action_factory.create_codecommit_source_action(source_output)
        else:
            action = self._action_factory.create_configured_source_action(stage_config, source_output)

        pipeline.add_stage(
            stage_name=self._get_stage_name(StageType.SOURCE),
            actions=[action]
        )

        return source_output

    def _create_control_stage(self, pipeline: codepipeline.Pipeline, 
                             input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
        """Create control stage for pipeline validation and gating.

        Args:
            pipeline: Pipeline to add control stage to
            input_artifact: Input artifact from previous stage

        Returns:
            Dictionary of output artifacts from control stage actions

        Side Effects:
            Adds control stage with CodeBuild actions to pipeline
        """
        return self._create_codebuild_stage(pipeline, StageType.CONTROL, input_artifact)

    def _create_synth_stage(self, pipeline: codepipeline.Pipeline, 
                           input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
        """Create synth stage for CDK synthesis and template generation.

        Args:
            pipeline: Pipeline to add synth stage to
            input_artifact: Input artifact from source stage

        Returns:
            Dictionary of output artifacts from synth stage actions

        Side Effects:
            Adds synth stage with CodeBuild actions to pipeline
        """
        return self._create_codebuild_stage(pipeline, StageType.SYNTH, input_artifact)

    def _create_build_stage(self, pipeline: codepipeline.Pipeline, 
                           input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
        """Create build stage for compilation and testing.

        Args:
            pipeline: Pipeline to add build stage to
            input_artifact: Input artifact from source stage

        Returns:
            Dictionary of output artifacts from build actions

        Side Effects:
            Adds build stage with CodeBuild actions to pipeline
        """
        return self._create_codebuild_stage(pipeline, StageType.BUILD, input_artifact)

    def _create_publish_stage(self, pipeline: codepipeline.Pipeline, 
                             input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
        """Create publish stage for package/image publishing.

        Args:
            pipeline: Pipeline to add publish stage to
            input_artifact: Input artifact from previous stage

        Returns:
            Dictionary of output artifacts from publish actions

        Side Effects:
            Adds publish stage with CodeBuild actions to pipeline
        """
        return self._create_codebuild_stage(pipeline, StageType.PUBLISH, input_artifact)

    def _create_self_mutate_stage(self, pipeline: codepipeline.Pipeline, 
                                 input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
        """Create self-mutate stage for pipeline updates.

        Args:
            pipeline: Pipeline to add self-mutate stage to
            input_artifact: Input artifact from source stage

        Returns:
            Dictionary of output artifacts (empty for self-mutate)

        Side Effects:
            Adds self-mutate stage with CodeBuild actions to pipeline
        """
        return self._create_codebuild_stage(pipeline, StageType.SELF_MUTATE, input_artifact, include_outputs=False)

    def _create_manual_source_stage(self, pipeline: codepipeline.Pipeline, 
                                   input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
        """Create manual source stage for artifact-based deployments.

        Args:
            pipeline: Pipeline to add manual source stage to
            input_artifact: Input artifact from source stage

        Returns:
            Dictionary of output artifacts from manual source actions

        Side Effects:
            Adds manual source stage with CodeBuild actions to pipeline
        """
        return self._create_codebuild_stage(pipeline, StageType.MANUAL_SOURCE, input_artifact)

    def _create_approval_stage(self, pipeline: codepipeline.Pipeline) -> None:
        """Create manual approval stage for pipeline gating.

        Args:
            pipeline: Pipeline to add approval stage to

        Side Effects:
            Adds manual approval stage to pipeline
        """
        action = self._action_factory.create_manual_approval_action()
        pipeline.add_stage(
            stage_name=self._get_stage_name(StageType.MANUAL_APPROVAL),
            actions=[action]
        )

    def _create_codebuild_stage(self, pipeline: codepipeline.Pipeline, stage_type: StageType,
                               input_artifact: codepipeline.Artifact, 
                               available_outputs: Dict[str, codepipeline.Artifact] = None,
                               include_outputs: bool = True) -> Dict[str, codepipeline.Artifact]:
        """Create generic CodeBuild stage with multiple actions.

        Args:
            pipeline: Pipeline to add stage to
            stage_type: Type of stage being created
            input_artifact: Primary input artifact
            available_outputs: Available outputs from previous stages
            include_outputs: Whether to create output artifacts

        Returns:
            Dictionary of output artifacts from stage actions

        Side Effects:
            Adds stage with configured CodeBuild actions to pipeline
        """
        outputs: Dict[str, codepipeline.Artifact] = {}
        actions: List[Any] = []

        stage_config: StagesConfig = getattr(self._config.stages, stage_type.value, None)

        for item_name, item_cfg in stage_config.items():
            item_cfg = cast(StageItemConfig, item_cfg)

            # Determine input artifact
            stage_input = self._get_stage_input_artifact(item_cfg, input_artifact, available_outputs)

            # Get extra inputs
            extra_inputs = self._get_extra_input_artifacts(item_cfg, available_outputs, outputs)

            # Create output artifacts
            output_artifacts = None
            if include_outputs:
                output_artifacts = self._create_output_artifacts(item_cfg, item_name, outputs)

            # Create the action
            action = self._action_factory.create_codebuild_action(
                item_name, item_cfg, stage_input, extra_inputs, output_artifacts
            )
            actions.append(action)

        pipeline.add_stage(
            stage_name=self._get_stage_name(stage_type),
            actions=actions
        )

        return outputs

    def _create_release_stage(self, pipeline: codepipeline.Pipeline, 
                             input_artifact: codepipeline.Artifact,
                             available_outputs: Dict[str, codepipeline.Artifact] = None) -> None:
        """Create release stage for artifact publishing and distribution.

        Args:
            pipeline: Pipeline to add release stage to
            input_artifact: Primary input artifact
            available_outputs: Available outputs from previous stages

        Raises:
            ValueError: For unsupported action types

        Side Effects:
            Adds release stage with S3 or CodeBuild actions to pipeline
        """
        actions = []
        stage_config = getattr(self._config.stages, StageType.RELEASE.value)
        pipeline_role = pipeline.role

        for item_name, item_cfg in stage_config.items():
            item_cfg = cast(StageItemConfig, item_cfg)

            # Determine input artifact
            stage_input = self._get_stage_input_artifact(item_cfg, input_artifact, available_outputs)

            # Get extra inputs
            extra_inputs = self._get_extra_input_artifacts(item_cfg, available_outputs, {})

            # Create action based on type
            action_type = getattr(item_cfg, "action_type", ActionType.S3)

            match action_type:
                case ActionType.S3:
                    action = self._action_factory.create_s3_deploy_action(item_name, item_cfg, stage_input, self._config.release_bucket, pipeline_role)
                case ActionType.CODEBUILD:
                    action = self._action_factory.create_codebuild_action(item_name, item_cfg, stage_input, extra_inputs)
                case _:
                    raise ValueError(f"Unsupported action type: {action_type.name} for the {StageType.RELEASE.value} stage")

            actions.append(action)

        pipeline.add_stage(
            stage_name=self._get_stage_name(StageType.RELEASE),
            actions=actions
        )

    def _create_deploy_stage(self, pipeline: codepipeline.Pipeline, 
                            input_artifact: codepipeline.Artifact,
                            available_outputs: Dict[str, codepipeline.Artifact] = None) -> None:
        """Create deploy stage for environment deployment.

        Args:
            pipeline: Pipeline to add deploy stage to
            input_artifact: Primary input artifact
            available_outputs: Available outputs from previous stages

        Raises:
            NotImplementedError: For unsupported ECS deployments
            ValueError: For unsupported action types
            AttributeError: If required stage configuration attributes are missing

        Side Effects:
            Adds deploy stage with CloudFormation, S3, or CodeBuild actions to pipeline
        """
        outputs = {}
        actions = []
        stage_config = getattr(self._config.stages, StageType.DEPLOY.value)

        for item_name, item_cfg in stage_config.items():
            item_cfg = cast(StageItemConfig, item_cfg)

            # Determine input artifact
            stage_input = self._get_stage_input_artifact(item_cfg, input_artifact, available_outputs)

            # Get extra inputs
            extra_inputs = self._get_extra_input_artifacts(item_cfg, available_outputs, outputs)

            # Create output artifact
            outputs[item_name] = codepipeline.Artifact(
                self._get_name_for_resource(f"{item_name}{PipelineConstants.OUTPUT_SUFFIX}", 
                                          max_length=AWSResourceNameLength.ARTIFACT.value)
            )

            # Create action based on type
            action_type = getattr(item_cfg, "action_type", ActionType.S3)

            match action_type:
                case ActionType.S3:
                    action = self._action_factory.create_s3_deploy_action(item_name, item_cfg, stage_input, getattr(item_cfg, 'deploy_bucket'))
                case ActionType.CLOUDFORMATION:
                    action = self._action_factory.create_cloudformation_action(item_name, item_cfg, stage_input)
                case ActionType.CODEBUILD:
                    action = self._action_factory.create_codebuild_action(item_name, item_cfg, stage_input, extra_inputs)
                case ActionType.ECS:
                    raise NotImplementedError("ECS deploy is not yet supported in the CodePipelineBuilder.")
                case _:
                    raise ValueError(f"Unsupported action type: {action_type.name} for the {StageType.DEPLOY.value} stage")

            actions.append(action)

        pipeline.add_stage(
            stage_name=self._get_stage_name(StageType.DEPLOY),
            actions=actions
        )

    def _create_pipeline_role(self, scope: Construct) -> iam.Role:
        """Create IAM role for pipeline execution with required permissions.

        Args:
            scope: CDK construct scope

        Returns:
            IAM role with pipeline execution permissions

        Side Effects:
            Creates IAM role with S3, CodePipeline, KMS, and CodeArtifact permissions
        """
        target_env = self._config.target_env or None
        role_id = self._get_cfn_logical_id("role", git_branch=self._config.git_branch, target_env=target_env)
        role_name = self._get_pipeline_role_name()

        role = iam.Role(
            scope, role_id,
            role_name=role_name,
            assumed_by=iam.ServicePrincipal("codepipeline.amazonaws.com"),
            managed_policies=[
                iam.ManagedPolicy.from_aws_managed_policy_name("AmazonS3FullAccess"),
                iam.ManagedPolicy.from_aws_managed_policy_name("AWSCodePipeline_FullAccess")
            ]
        )

        self._add_pipeline_role_policies(role)
        return role

    def _add_pipeline_role_policies(self, role: iam.Role) -> None:
        """Add additional IAM policies to pipeline role.

        Args:
            role: IAM role to add policies to

        Side Effects:
            Adds KMS and CodeArtifact permissions to the role
        """
        # Add KMS and CodeArtifact permissions
        role.add_to_policy(iam.PolicyStatement(
            actions=PipelineConstants.KMS_ACTIONS,
            resources=["*"]
        ))

        # Add STS permissions for CodeArtifact
        role.add_to_policy(iam.PolicyStatement(
            actions=["sts:GetServiceBearerToken"],
            resources=["*"],
            conditions={
                "StringEquals": {
                    "sts:AWSServiceName": "codeartifact.amazonaws.com"
                }
            }
        ))

    def _create_event_rule_for_commits(self, scope: Construct, pipeline: codepipeline.Pipeline) -> None:
        """Create EventBridge rule for automatic pipeline triggering on git commits.

        Args:
            scope: CDK construct scope
            pipeline: Pipeline to trigger from commits

        Side Effects:
            Creates EventBridge rule monitoring CodeCommit repository
            Creates IAM role for EventBridge pipeline execution
            Configures rule target to trigger pipeline
        """
        event_rule = self._create_codecommit_event_rule(scope)
        event_role = self._create_event_role(scope)

        event_rule.add_target(targets.CodePipeline(
            pipeline=pipeline,
            event_role=event_role
        ))

    def _create_codecommit_event_rule(self, scope: Construct) -> events.Rule:
        """Create EventBridge rule for CodeCommit repository changes.

        Args:
            scope: CDK construct scope

        Returns:
            EventBridge rule monitoring git branch changes
        """
        rule_id = self._get_cfn_logical_id(PipelineConstants.EVENT_RULE_SUFFIX, git_branch=self._config.git_branch)
        rule_name = self._get_name_for_resource(
            PipelineConstants.EVENT_RULE_SUFFIX,
            git_branch=self._config.git_branch,
            max_length=AWSResourceNameLength.EVENTS_RULE.value
        )

        return events.Rule(
            scope, rule_id,
            rule_name=rule_name,
            event_pattern=events.EventPattern(
                source=["aws.codecommit"],
                detail_type=["CodeCommit Repository State Change"],
                resources=[self._config.typed.git_repo.repository_arn],
                detail={
                    "event": ["referenceCreated", "referenceUpdated"],
                    "referenceType": ["branch"],
                    "referenceName": [self._config.git_branch]
                }
            )
        )

    def _create_event_role(self, scope: Construct) -> iam.Role:
        """Create IAM role for EventBridge pipeline triggering.

        Args:
            scope: CDK construct scope

        Returns:
            IAM role for EventBridge service to trigger pipelines
        """
        role_id = self._get_cfn_logical_id(PipelineConstants.EVENT_ROLE_SUFFIX, git_branch=self._config.git_branch)
        role_name = self._get_name_for_resource(
            PipelineConstants.EVENT_ROLE_SUFFIX,
            git_branch=self._config.git_branch,
            max_length=AWSResourceNameLength.ROLE.value
        )

        return iam.Role(
            scope, role_id,
            role_name=role_name,
            assumed_by=iam.ServicePrincipal("events.amazonaws.com"),
            managed_policies=[iam.ManagedPolicy.from_aws_managed_policy_name("AWSCodePipeline_FullAccess")]
        )

    def _get_stage_input_artifact(self, item_cfg: StageItemConfig, 
                                 default_input: codepipeline.Artifact,
                                 available_outputs: Dict[str, codepipeline.Artifact] = None) -> codepipeline.Artifact:
        """Determine input artifact for stage action based on configuration.

        Args:
            item_cfg: Stage item configuration with potential artifact_id specification
            default_input: Default input artifact if no specific artifact_id configured
            available_outputs: Available output artifacts from previous stages (optional)

        Returns:
            Input artifact for the action (either specified or default)
        """
        if available_outputs and item_cfg.artifact_id and item_cfg.artifact_id in available_outputs:
            return available_outputs[item_cfg.artifact_id]
        return default_input

    def _get_extra_input_artifacts(self, item_cfg: StageItemConfig, 
                                  available_outputs: Dict[str, codepipeline.Artifact] = None,
                                  local_outputs: Dict[str, codepipeline.Artifact] = None) -> List[codepipeline.Artifact]:
        """Get additional input artifacts for stage actions.

       Args:
           item_cfg: Stage item configuration with extra_inputs specification
           available_outputs: Available outputs from previous stages
           local_outputs: Outputs from current stage actions

       Returns:
           List of additional input artifacts for the action
       """
        extra_inputs: List[codepipeline.Artifact] = []

        if item_cfg.extra_inputs is not None:
            for extra in item_cfg.extra_inputs:
                source_dict = available_outputs if available_outputs and extra in available_outputs else local_outputs
                if source_dict and extra in source_dict:
                    extra_inputs.append(source_dict[extra])

        return extra_inputs

    def _create_output_artifacts(self, item_cfg: StageItemConfig, item_name: str,
                                outputs: Dict[str, codepipeline.Artifact]) -> List[codepipeline.Artifact]:
        """Create output artifacts for stage actions.

       Args:
           item_cfg: Stage item configuration with output_names specification
           item_name: Name of the stage item for default naming
           outputs: Dictionary to store created artifacts

       Returns:
           List of output artifacts for the action

       Side Effects:
           Adds created artifacts to outputs dictionary
       """
        output_artifacts = []

        if item_cfg.output_names is not None:
            for output_name in item_cfg.output_names:
                outputs[output_name] = codepipeline.Artifact(output_name)
                output_artifacts.append(outputs[output_name])
        else:
            # Default output artifact
            artifact_name = self._get_name_for_resource(
                f"{item_name}{PipelineConstants.OUTPUT_SUFFIX}", 
                max_length=AWSResourceNameLength.ARTIFACT.value
            )
            outputs[item_name] = codepipeline.Artifact(artifact_name)
            output_artifacts.append(outputs[item_name])

        return output_artifacts

    def _get_build_env_vars(self, action_env_vars: Dict[str, codebuild.BuildEnvironmentVariable] = None) -> Dict[str, codebuild.BuildEnvironmentVariable]:
        """Get environment variables for CodeBuild actions.

       Args:
           action_env_vars: Custom environment variables from action configuration

       Returns:
           Dictionary of environment variables including defaults and custom vars
       """
        env_vars = {
            "IS_DEV_BRANCH": codebuild.BuildEnvironmentVariable(
                value="true" if self._config.is_dev_branch else "false",
                type=codebuild.BuildEnvironmentVariableType.PLAINTEXT
            ),
            "GIT_BRANCH": codebuild.BuildEnvironmentVariable(
                value=self._config.git_branch,
                type=codebuild.BuildEnvironmentVariableType.PLAINTEXT
            )
        }

        if action_env_vars is not None:
            env_vars.update(action_env_vars)

        return env_vars

    def _get_stage_name(self, stage_type: StageType) -> str:
        """Generate stage name from stage type.

       Args:
           stage_type: Type of stage

       Returns:
           Formatted stage name for pipeline
       """
        return f"{stage_type.name.replace("_", "-")}-STAGE"

    def _generate_pipeline_id_and_name(self) -> Tuple[str, str]:
        """Generate pipeline CloudFormation ID and AWS name.

       Returns:
           Tuple of (pipeline_id, pipeline_name) following naming conventions
       """
        base_name = self._config.target_env if self._config.target_env is not None else None

        pipeline_id = self._get_cfn_logical_id(base_name, git_branch=self._config.git_branch)
        pipeline_name = self._get_name_for_resource(
            base_name, 
            git_branch=self._config.git_branch,
            max_length=AWSResourceNameLength.CODEBUILD_PIPELINE_PROJECT.value
        )

        return pipeline_id, pipeline_name

    def _get_pipeline_role_name(self) -> str:
        """Generate IAM role name for pipeline execution.

        Returns:
            IAM role name following organizational naming conventions
        """
        target_env = self._config.target_env or self._application_helper.get_target_env()
        base_name = self._get_name_for_resource(
            target_env,
            git_branch=self._config.git_branch,
            max_length=AWSResourceNameLength.ROLE.value
        )
        return base_name + PipelineConstants.ROLE_SUFFIX

    def _set_config(self) -> None:
        """Create and validate the CodePipeline configuration.

       Merges builder configuration with usage to create a validated
       CodePipelineConfig object for pipeline creation.

       Raises:
           ValidationError: If the CodePipelineConfig validation fails

       Side Effects:
           Sets self._config to validated CodePipelineConfig instance
       """
        try:
            self._config = CodePipelineConfig(**{
                **self._builder_config,
                "usage": self._usage
            })
        except ValidationError as e:
            self._log_validation_error(e, CodePipelineConfig)
            raise

    def _control_consistency(self) -> None:
        """Validate builder configuration and internal state consistency.

       Ensures usage is set and validates the complete configuration
       using the CodePipelineConfig model.

       Raises:
           ValidationError: If configuration validation fails

       Side Effects:
           Calls _set_config to validate and set configuration
       """
        super()._control_consistency()
        self._set_config()

Attributes

_requires_usage = True class-attribute instance-attribute

_resource_type = AWSResourceType.CODEPIPELINE class-attribute instance-attribute

Functions

_add_pipeline_role_policies(role)

Add additional IAM policies to pipeline role.

Parameters:

Name	Type	Description	Default
role	Role	IAM role to add policies to	required

Side Effects

Adds KMS and CodeArtifact permissions to the role

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _add_pipeline_role_policies(self, role: iam.Role) -> None:
    """Add additional IAM policies to pipeline role.

    Args:
        role: IAM role to add policies to

    Side Effects:
        Adds KMS and CodeArtifact permissions to the role
    """
    # Add KMS and CodeArtifact permissions
    role.add_to_policy(iam.PolicyStatement(
        actions=PipelineConstants.KMS_ACTIONS,
        resources=["*"]
    ))

    # Add STS permissions for CodeArtifact
    role.add_to_policy(iam.PolicyStatement(
        actions=["sts:GetServiceBearerToken"],
        resources=["*"],
        conditions={
            "StringEquals": {
                "sts:AWSServiceName": "codeartifact.amazonaws.com"
            }
        }
    ))

_build_pipeline_stages(scope, pipeline)

Build pipeline stages based on usage configuration.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope	required
pipeline	Pipeline	Pipeline to configure with stages	required

Side Effects

Delegates to orchestrator to create appropriate stage configuration

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _build_pipeline_stages(self, scope: Construct, pipeline: codepipeline.Pipeline) -> None:
    """Build pipeline stages based on usage configuration.

    Args:
        scope: CDK construct scope
        pipeline: Pipeline to configure with stages

    Side Effects:
        Delegates to orchestrator to create appropriate stage configuration
    """
    match self._config.usage:
        case Usage.BUILD_RELEASE:
            self._orchestrator.create_build_release_pipeline(scope, pipeline)
        case Usage.BUILD_DEPLOY:
            self._orchestrator.create_build_deploy_pipeline(scope, pipeline)
        case Usage.DEPLOY:
            self._orchestrator.create_deploy_pipeline(scope, pipeline)
        case Usage.BUILD_ASSESS:
            self._orchestrator.create_build_assess_pipeline(scope, pipeline)

_control_consistency()

Validate builder configuration and internal state consistency.

Ensures usage is set and validates the complete configuration using the CodePipelineConfig model.

Raises:

Type	Description
ValidationError	If configuration validation fails

Side Effects

Calls _set_config to validate and set configuration

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _control_consistency(self) -> None:
    """Validate builder configuration and internal state consistency.

   Ensures usage is set and validates the complete configuration
   using the CodePipelineConfig model.

   Raises:
       ValidationError: If configuration validation fails

   Side Effects:
       Calls _set_config to validate and set configuration
   """
    super()._control_consistency()
    self._set_config()

_create_approval_stage(pipeline)

Create manual approval stage for pipeline gating.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline to add approval stage to	required

Side Effects

Adds manual approval stage to pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_approval_stage(self, pipeline: codepipeline.Pipeline) -> None:
    """Create manual approval stage for pipeline gating.

    Args:
        pipeline: Pipeline to add approval stage to

    Side Effects:
        Adds manual approval stage to pipeline
    """
    action = self._action_factory.create_manual_approval_action()
    pipeline.add_stage(
        stage_name=self._get_stage_name(StageType.MANUAL_APPROVAL),
        actions=[action]
    )

_create_base_pipeline(scope)

Create base CodePipeline with core configuration.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope	required

Returns:

Type	Description
Pipeline	Basic pipeline with role and configuration, without stages

Side Effects

Creates IAM role for pipeline execution

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_base_pipeline(self, scope: Construct) -> codepipeline.Pipeline:
    """Create base CodePipeline with core configuration.

    Args:
        scope: CDK construct scope

    Returns:
        Basic pipeline with role and configuration, without stages

    Side Effects:
        Creates IAM role for pipeline execution
    """
    pipeline_id, pipeline_name = self._generate_pipeline_id_and_name()
    pipeline_role = self._create_pipeline_role(scope)

    return codepipeline.Pipeline(
        scope, pipeline_id,
        pipeline_name=pipeline_name,
        pipeline_type=codepipeline.PipelineType.V2,
        role=pipeline_role,
        restart_execution_on_update=self._config.restart_execution_on_update,
        cross_account_keys=True,
        artifact_bucket=self._config.artifact_bucket
    )

_create_build_stage(pipeline, input_artifact)

Create build stage for compilation and testing.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline to add build stage to	required
input_artifact	Artifact	Input artifact from source stage	required

Returns:

Type	Description
Dict[str, Artifact]	Dictionary of output artifacts from build actions

Side Effects

Adds build stage with CodeBuild actions to pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_build_stage(self, pipeline: codepipeline.Pipeline, 
                       input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
    """Create build stage for compilation and testing.

    Args:
        pipeline: Pipeline to add build stage to
        input_artifact: Input artifact from source stage

    Returns:
        Dictionary of output artifacts from build actions

    Side Effects:
        Adds build stage with CodeBuild actions to pipeline
    """
    return self._create_codebuild_stage(pipeline, StageType.BUILD, input_artifact)

_create_codebuild_stage(pipeline, stage_type, input_artifact, available_outputs=None, include_outputs=True)

Create generic CodeBuild stage with multiple actions.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline to add stage to	required
stage_type	StageType	Type of stage being created	required
input_artifact	Artifact	Primary input artifact	required
available_outputs	Dict[str, Artifact]	Available outputs from previous stages	None
include_outputs	bool	Whether to create output artifacts	True

Returns:

Type	Description
Dict[str, Artifact]	Dictionary of output artifacts from stage actions

Side Effects

Adds stage with configured CodeBuild actions to pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_codebuild_stage(self, pipeline: codepipeline.Pipeline, stage_type: StageType,
                           input_artifact: codepipeline.Artifact, 
                           available_outputs: Dict[str, codepipeline.Artifact] = None,
                           include_outputs: bool = True) -> Dict[str, codepipeline.Artifact]:
    """Create generic CodeBuild stage with multiple actions.

    Args:
        pipeline: Pipeline to add stage to
        stage_type: Type of stage being created
        input_artifact: Primary input artifact
        available_outputs: Available outputs from previous stages
        include_outputs: Whether to create output artifacts

    Returns:
        Dictionary of output artifacts from stage actions

    Side Effects:
        Adds stage with configured CodeBuild actions to pipeline
    """
    outputs: Dict[str, codepipeline.Artifact] = {}
    actions: List[Any] = []

    stage_config: StagesConfig = getattr(self._config.stages, stage_type.value, None)

    for item_name, item_cfg in stage_config.items():
        item_cfg = cast(StageItemConfig, item_cfg)

        # Determine input artifact
        stage_input = self._get_stage_input_artifact(item_cfg, input_artifact, available_outputs)

        # Get extra inputs
        extra_inputs = self._get_extra_input_artifacts(item_cfg, available_outputs, outputs)

        # Create output artifacts
        output_artifacts = None
        if include_outputs:
            output_artifacts = self._create_output_artifacts(item_cfg, item_name, outputs)

        # Create the action
        action = self._action_factory.create_codebuild_action(
            item_name, item_cfg, stage_input, extra_inputs, output_artifacts
        )
        actions.append(action)

    pipeline.add_stage(
        stage_name=self._get_stage_name(stage_type),
        actions=actions
    )

    return outputs

_create_codecommit_event_rule(scope)

Create EventBridge rule for CodeCommit repository changes.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope	required

Returns:

Type	Description
Rule	EventBridge rule monitoring git branch changes

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_codecommit_event_rule(self, scope: Construct) -> events.Rule:
    """Create EventBridge rule for CodeCommit repository changes.

    Args:
        scope: CDK construct scope

    Returns:
        EventBridge rule monitoring git branch changes
    """
    rule_id = self._get_cfn_logical_id(PipelineConstants.EVENT_RULE_SUFFIX, git_branch=self._config.git_branch)
    rule_name = self._get_name_for_resource(
        PipelineConstants.EVENT_RULE_SUFFIX,
        git_branch=self._config.git_branch,
        max_length=AWSResourceNameLength.EVENTS_RULE.value
    )

    return events.Rule(
        scope, rule_id,
        rule_name=rule_name,
        event_pattern=events.EventPattern(
            source=["aws.codecommit"],
            detail_type=["CodeCommit Repository State Change"],
            resources=[self._config.typed.git_repo.repository_arn],
            detail={
                "event": ["referenceCreated", "referenceUpdated"],
                "referenceType": ["branch"],
                "referenceName": [self._config.git_branch]
            }
        )
    )

_create_control_stage(pipeline, input_artifact)

Create control stage for pipeline validation and gating.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline to add control stage to	required
input_artifact	Artifact	Input artifact from previous stage	required

Returns:

Type	Description
Dict[str, Artifact]	Dictionary of output artifacts from control stage actions

Side Effects

Adds control stage with CodeBuild actions to pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_control_stage(self, pipeline: codepipeline.Pipeline, 
                         input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
    """Create control stage for pipeline validation and gating.

    Args:
        pipeline: Pipeline to add control stage to
        input_artifact: Input artifact from previous stage

    Returns:
        Dictionary of output artifacts from control stage actions

    Side Effects:
        Adds control stage with CodeBuild actions to pipeline
    """
    return self._create_codebuild_stage(pipeline, StageType.CONTROL, input_artifact)

_create_deploy_stage(pipeline, input_artifact, available_outputs=None)

Create deploy stage for environment deployment.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline to add deploy stage to	required
input_artifact	Artifact	Primary input artifact	required
available_outputs	Dict[str, Artifact]	Available outputs from previous stages	None

Raises:

Type	Description
NotImplementedError	For unsupported ECS deployments
ValueError	For unsupported action types
AttributeError	If required stage configuration attributes are missing

Side Effects

Adds deploy stage with CloudFormation, S3, or CodeBuild actions to pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_deploy_stage(self, pipeline: codepipeline.Pipeline, 
                        input_artifact: codepipeline.Artifact,
                        available_outputs: Dict[str, codepipeline.Artifact] = None) -> None:
    """Create deploy stage for environment deployment.

    Args:
        pipeline: Pipeline to add deploy stage to
        input_artifact: Primary input artifact
        available_outputs: Available outputs from previous stages

    Raises:
        NotImplementedError: For unsupported ECS deployments
        ValueError: For unsupported action types
        AttributeError: If required stage configuration attributes are missing

    Side Effects:
        Adds deploy stage with CloudFormation, S3, or CodeBuild actions to pipeline
    """
    outputs = {}
    actions = []
    stage_config = getattr(self._config.stages, StageType.DEPLOY.value)

    for item_name, item_cfg in stage_config.items():
        item_cfg = cast(StageItemConfig, item_cfg)

        # Determine input artifact
        stage_input = self._get_stage_input_artifact(item_cfg, input_artifact, available_outputs)

        # Get extra inputs
        extra_inputs = self._get_extra_input_artifacts(item_cfg, available_outputs, outputs)

        # Create output artifact
        outputs[item_name] = codepipeline.Artifact(
            self._get_name_for_resource(f"{item_name}{PipelineConstants.OUTPUT_SUFFIX}", 
                                      max_length=AWSResourceNameLength.ARTIFACT.value)
        )

        # Create action based on type
        action_type = getattr(item_cfg, "action_type", ActionType.S3)

        match action_type:
            case ActionType.S3:
                action = self._action_factory.create_s3_deploy_action(item_name, item_cfg, stage_input, getattr(item_cfg, 'deploy_bucket'))
            case ActionType.CLOUDFORMATION:
                action = self._action_factory.create_cloudformation_action(item_name, item_cfg, stage_input)
            case ActionType.CODEBUILD:
                action = self._action_factory.create_codebuild_action(item_name, item_cfg, stage_input, extra_inputs)
            case ActionType.ECS:
                raise NotImplementedError("ECS deploy is not yet supported in the CodePipelineBuilder.")
            case _:
                raise ValueError(f"Unsupported action type: {action_type.name} for the {StageType.DEPLOY.value} stage")

        actions.append(action)

    pipeline.add_stage(
        stage_name=self._get_stage_name(StageType.DEPLOY),
        actions=actions
    )

_create_event_role(scope)

Create IAM role for EventBridge pipeline triggering.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope	required

Returns:

Type	Description
Role	IAM role for EventBridge service to trigger pipelines

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_event_role(self, scope: Construct) -> iam.Role:
    """Create IAM role for EventBridge pipeline triggering.

    Args:
        scope: CDK construct scope

    Returns:
        IAM role for EventBridge service to trigger pipelines
    """
    role_id = self._get_cfn_logical_id(PipelineConstants.EVENT_ROLE_SUFFIX, git_branch=self._config.git_branch)
    role_name = self._get_name_for_resource(
        PipelineConstants.EVENT_ROLE_SUFFIX,
        git_branch=self._config.git_branch,
        max_length=AWSResourceNameLength.ROLE.value
    )

    return iam.Role(
        scope, role_id,
        role_name=role_name,
        assumed_by=iam.ServicePrincipal("events.amazonaws.com"),
        managed_policies=[iam.ManagedPolicy.from_aws_managed_policy_name("AWSCodePipeline_FullAccess")]
    )

_create_event_rule_for_commits(scope, pipeline)

Create EventBridge rule for automatic pipeline triggering on git commits.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope	required
pipeline	Pipeline	Pipeline to trigger from commits	required

Side Effects

Creates EventBridge rule monitoring CodeCommit repository Creates IAM role for EventBridge pipeline execution Configures rule target to trigger pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_event_rule_for_commits(self, scope: Construct, pipeline: codepipeline.Pipeline) -> None:
    """Create EventBridge rule for automatic pipeline triggering on git commits.

    Args:
        scope: CDK construct scope
        pipeline: Pipeline to trigger from commits

    Side Effects:
        Creates EventBridge rule monitoring CodeCommit repository
        Creates IAM role for EventBridge pipeline execution
        Configures rule target to trigger pipeline
    """
    event_rule = self._create_codecommit_event_rule(scope)
    event_role = self._create_event_role(scope)

    event_rule.add_target(targets.CodePipeline(
        pipeline=pipeline,
        event_role=event_role
    ))

_create_manual_source_stage(pipeline, input_artifact)

Create manual source stage for artifact-based deployments.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline to add manual source stage to	required
input_artifact	Artifact	Input artifact from source stage	required

Returns:

Type	Description
Dict[str, Artifact]	Dictionary of output artifacts from manual source actions

Side Effects

Adds manual source stage with CodeBuild actions to pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_manual_source_stage(self, pipeline: codepipeline.Pipeline, 
                               input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
    """Create manual source stage for artifact-based deployments.

    Args:
        pipeline: Pipeline to add manual source stage to
        input_artifact: Input artifact from source stage

    Returns:
        Dictionary of output artifacts from manual source actions

    Side Effects:
        Adds manual source stage with CodeBuild actions to pipeline
    """
    return self._create_codebuild_stage(pipeline, StageType.MANUAL_SOURCE, input_artifact)

_create_output_artifacts(item_cfg, item_name, outputs)

Create output artifacts for stage actions.

Parameters:

Name	Type	Description	Default
item_cfg	StageItemConfig	Stage item configuration with output_names specification	required
item_name	str	Name of the stage item for default naming	required
outputs	Dict[str, Artifact]	Dictionary to store created artifacts	required

Returns:

Type	Description
List[Artifact]	List of output artifacts for the action

Side Effects

Adds created artifacts to outputs dictionary

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_output_artifacts(self, item_cfg: StageItemConfig, item_name: str,
                            outputs: Dict[str, codepipeline.Artifact]) -> List[codepipeline.Artifact]:
    """Create output artifacts for stage actions.

   Args:
       item_cfg: Stage item configuration with output_names specification
       item_name: Name of the stage item for default naming
       outputs: Dictionary to store created artifacts

   Returns:
       List of output artifacts for the action

   Side Effects:
       Adds created artifacts to outputs dictionary
   """
    output_artifacts = []

    if item_cfg.output_names is not None:
        for output_name in item_cfg.output_names:
            outputs[output_name] = codepipeline.Artifact(output_name)
            output_artifacts.append(outputs[output_name])
    else:
        # Default output artifact
        artifact_name = self._get_name_for_resource(
            f"{item_name}{PipelineConstants.OUTPUT_SUFFIX}", 
            max_length=AWSResourceNameLength.ARTIFACT.value
        )
        outputs[item_name] = codepipeline.Artifact(artifact_name)
        output_artifacts.append(outputs[item_name])

    return output_artifacts

_create_pipeline_role(scope)

Create IAM role for pipeline execution with required permissions.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope	required

Returns:

Type	Description
Role	IAM role with pipeline execution permissions

Side Effects

Creates IAM role with S3, CodePipeline, KMS, and CodeArtifact permissions

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_pipeline_role(self, scope: Construct) -> iam.Role:
    """Create IAM role for pipeline execution with required permissions.

    Args:
        scope: CDK construct scope

    Returns:
        IAM role with pipeline execution permissions

    Side Effects:
        Creates IAM role with S3, CodePipeline, KMS, and CodeArtifact permissions
    """
    target_env = self._config.target_env or None
    role_id = self._get_cfn_logical_id("role", git_branch=self._config.git_branch, target_env=target_env)
    role_name = self._get_pipeline_role_name()

    role = iam.Role(
        scope, role_id,
        role_name=role_name,
        assumed_by=iam.ServicePrincipal("codepipeline.amazonaws.com"),
        managed_policies=[
            iam.ManagedPolicy.from_aws_managed_policy_name("AmazonS3FullAccess"),
            iam.ManagedPolicy.from_aws_managed_policy_name("AWSCodePipeline_FullAccess")
        ]
    )

    self._add_pipeline_role_policies(role)
    return role

_create_publish_stage(pipeline, input_artifact)

Create publish stage for package/image publishing.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline to add publish stage to	required
input_artifact	Artifact	Input artifact from previous stage	required

Returns:

Type	Description
Dict[str, Artifact]	Dictionary of output artifacts from publish actions

Side Effects

Adds publish stage with CodeBuild actions to pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_publish_stage(self, pipeline: codepipeline.Pipeline, 
                         input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
    """Create publish stage for package/image publishing.

    Args:
        pipeline: Pipeline to add publish stage to
        input_artifact: Input artifact from previous stage

    Returns:
        Dictionary of output artifacts from publish actions

    Side Effects:
        Adds publish stage with CodeBuild actions to pipeline
    """
    return self._create_codebuild_stage(pipeline, StageType.PUBLISH, input_artifact)

_create_release_stage(pipeline, input_artifact, available_outputs=None)

Create release stage for artifact publishing and distribution.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline to add release stage to	required
input_artifact	Artifact	Primary input artifact	required
available_outputs	Dict[str, Artifact]	Available outputs from previous stages	None

Raises:

Type	Description
ValueError	For unsupported action types

Side Effects

Adds release stage with S3 or CodeBuild actions to pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_release_stage(self, pipeline: codepipeline.Pipeline, 
                         input_artifact: codepipeline.Artifact,
                         available_outputs: Dict[str, codepipeline.Artifact] = None) -> None:
    """Create release stage for artifact publishing and distribution.

    Args:
        pipeline: Pipeline to add release stage to
        input_artifact: Primary input artifact
        available_outputs: Available outputs from previous stages

    Raises:
        ValueError: For unsupported action types

    Side Effects:
        Adds release stage with S3 or CodeBuild actions to pipeline
    """
    actions = []
    stage_config = getattr(self._config.stages, StageType.RELEASE.value)
    pipeline_role = pipeline.role

    for item_name, item_cfg in stage_config.items():
        item_cfg = cast(StageItemConfig, item_cfg)

        # Determine input artifact
        stage_input = self._get_stage_input_artifact(item_cfg, input_artifact, available_outputs)

        # Get extra inputs
        extra_inputs = self._get_extra_input_artifacts(item_cfg, available_outputs, {})

        # Create action based on type
        action_type = getattr(item_cfg, "action_type", ActionType.S3)

        match action_type:
            case ActionType.S3:
                action = self._action_factory.create_s3_deploy_action(item_name, item_cfg, stage_input, self._config.release_bucket, pipeline_role)
            case ActionType.CODEBUILD:
                action = self._action_factory.create_codebuild_action(item_name, item_cfg, stage_input, extra_inputs)
            case _:
                raise ValueError(f"Unsupported action type: {action_type.name} for the {StageType.RELEASE.value} stage")

        actions.append(action)

    pipeline.add_stage(
        stage_name=self._get_stage_name(StageType.RELEASE),
        actions=actions
    )

_create_self_mutate_stage(pipeline, input_artifact)

Create self-mutate stage for pipeline updates.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline to add self-mutate stage to	required
input_artifact	Artifact	Input artifact from source stage	required

Returns:

Type	Description
Dict[str, Artifact]	Dictionary of output artifacts (empty for self-mutate)

Side Effects

Adds self-mutate stage with CodeBuild actions to pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_self_mutate_stage(self, pipeline: codepipeline.Pipeline, 
                             input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
    """Create self-mutate stage for pipeline updates.

    Args:
        pipeline: Pipeline to add self-mutate stage to
        input_artifact: Input artifact from source stage

    Returns:
        Dictionary of output artifacts (empty for self-mutate)

    Side Effects:
        Adds self-mutate stage with CodeBuild actions to pipeline
    """
    return self._create_codebuild_stage(pipeline, StageType.SELF_MUTATE, input_artifact, include_outputs=False)

_create_source_stage(pipeline)

Create source stage for git repository integration.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline to add source stage to	required

Returns:

Type	Description
Artifact	Source output artifact for downstream stages

Side Effects

Adds source stage to pipeline with CodeCommit or configured action

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_source_stage(self, pipeline: codepipeline.Pipeline) -> codepipeline.Artifact:
    """Create source stage for git repository integration.

    Args:
        pipeline: Pipeline to add source stage to

    Returns:
        Source output artifact for downstream stages

    Side Effects:
        Adds source stage to pipeline with CodeCommit or configured action
    """
    source_output = codepipeline.Artifact(
        self._get_name_for_resource("Source-Output", max_length=AWSResourceNameLength.ARTIFACT.value)
    )

    stage_config = getattr(self._config.stages, StageType.SOURCE.value, None)

    if stage_config is None:
        # Default CodeCommit source action
        action = self._action_factory.create_codecommit_source_action(source_output)
    else:
        action = self._action_factory.create_configured_source_action(stage_config, source_output)

    pipeline.add_stage(
        stage_name=self._get_stage_name(StageType.SOURCE),
        actions=[action]
    )

    return source_output

_create_synth_stage(pipeline, input_artifact)

Create synth stage for CDK synthesis and template generation.

Parameters:

Name	Type	Description	Default
pipeline	Pipeline	Pipeline to add synth stage to	required
input_artifact	Artifact	Input artifact from source stage	required

Returns:

Type	Description
Dict[str, Artifact]	Dictionary of output artifacts from synth stage actions

Side Effects

Adds synth stage with CodeBuild actions to pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _create_synth_stage(self, pipeline: codepipeline.Pipeline, 
                       input_artifact: codepipeline.Artifact) -> Dict[str, codepipeline.Artifact]:
    """Create synth stage for CDK synthesis and template generation.

    Args:
        pipeline: Pipeline to add synth stage to
        input_artifact: Input artifact from source stage

    Returns:
        Dictionary of output artifacts from synth stage actions

    Side Effects:
        Adds synth stage with CodeBuild actions to pipeline
    """
    return self._create_codebuild_stage(pipeline, StageType.SYNTH, input_artifact)

_generate_pipeline_id_and_name()

Generate pipeline CloudFormation ID and AWS name.

Returns:

Type	Description
Tuple[str, str]	Tuple of (pipeline_id, pipeline_name) following naming conventions

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _generate_pipeline_id_and_name(self) -> Tuple[str, str]:
    """Generate pipeline CloudFormation ID and AWS name.

   Returns:
       Tuple of (pipeline_id, pipeline_name) following naming conventions
   """
    base_name = self._config.target_env if self._config.target_env is not None else None

    pipeline_id = self._get_cfn_logical_id(base_name, git_branch=self._config.git_branch)
    pipeline_name = self._get_name_for_resource(
        base_name, 
        git_branch=self._config.git_branch,
        max_length=AWSResourceNameLength.CODEBUILD_PIPELINE_PROJECT.value
    )

    return pipeline_id, pipeline_name

_get_build_env_vars(action_env_vars=None)

Get environment variables for CodeBuild actions.

Parameters:

Name	Type	Description	Default
action_env_vars	Dict[str, BuildEnvironmentVariable]	Custom environment variables from action configuration	None

Returns:

Type	Description
Dict[str, BuildEnvironmentVariable]	Dictionary of environment variables including defaults and custom vars

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _get_build_env_vars(self, action_env_vars: Dict[str, codebuild.BuildEnvironmentVariable] = None) -> Dict[str, codebuild.BuildEnvironmentVariable]:
    """Get environment variables for CodeBuild actions.

   Args:
       action_env_vars: Custom environment variables from action configuration

   Returns:
       Dictionary of environment variables including defaults and custom vars
   """
    env_vars = {
        "IS_DEV_BRANCH": codebuild.BuildEnvironmentVariable(
            value="true" if self._config.is_dev_branch else "false",
            type=codebuild.BuildEnvironmentVariableType.PLAINTEXT
        ),
        "GIT_BRANCH": codebuild.BuildEnvironmentVariable(
            value=self._config.git_branch,
            type=codebuild.BuildEnvironmentVariableType.PLAINTEXT
        )
    }

    if action_env_vars is not None:
        env_vars.update(action_env_vars)

    return env_vars

_get_extra_input_artifacts(item_cfg, available_outputs=None, local_outputs=None)

Get additional input artifacts for stage actions.

Parameters:

Name	Type	Description	Default
item_cfg	StageItemConfig	Stage item configuration with extra_inputs specification	required
available_outputs	Dict[str, Artifact]	Available outputs from previous stages	None
local_outputs	Dict[str, Artifact]	Outputs from current stage actions	None

Returns:

Type	Description
List[Artifact]	List of additional input artifacts for the action

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _get_extra_input_artifacts(self, item_cfg: StageItemConfig, 
                              available_outputs: Dict[str, codepipeline.Artifact] = None,
                              local_outputs: Dict[str, codepipeline.Artifact] = None) -> List[codepipeline.Artifact]:
    """Get additional input artifacts for stage actions.

   Args:
       item_cfg: Stage item configuration with extra_inputs specification
       available_outputs: Available outputs from previous stages
       local_outputs: Outputs from current stage actions

   Returns:
       List of additional input artifacts for the action
   """
    extra_inputs: List[codepipeline.Artifact] = []

    if item_cfg.extra_inputs is not None:
        for extra in item_cfg.extra_inputs:
            source_dict = available_outputs if available_outputs and extra in available_outputs else local_outputs
            if source_dict and extra in source_dict:
                extra_inputs.append(source_dict[extra])

    return extra_inputs

_get_pipeline_role_name()

Generate IAM role name for pipeline execution.

Returns:

Type	Description
str	IAM role name following organizational naming conventions

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _get_pipeline_role_name(self) -> str:
    """Generate IAM role name for pipeline execution.

    Returns:
        IAM role name following organizational naming conventions
    """
    target_env = self._config.target_env or self._application_helper.get_target_env()
    base_name = self._get_name_for_resource(
        target_env,
        git_branch=self._config.git_branch,
        max_length=AWSResourceNameLength.ROLE.value
    )
    return base_name + PipelineConstants.ROLE_SUFFIX

_get_stage_input_artifact(item_cfg, default_input, available_outputs=None)

Determine input artifact for stage action based on configuration.

Parameters:

Name	Type	Description	Default
item_cfg	StageItemConfig	Stage item configuration with potential artifact_id specification	required
default_input	Artifact	Default input artifact if no specific artifact_id configured	required
available_outputs	Dict[str, Artifact]	Available output artifacts from previous stages (optional)	None

Returns:

Type	Description
Artifact	Input artifact for the action (either specified or default)

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _get_stage_input_artifact(self, item_cfg: StageItemConfig, 
                             default_input: codepipeline.Artifact,
                             available_outputs: Dict[str, codepipeline.Artifact] = None) -> codepipeline.Artifact:
    """Determine input artifact for stage action based on configuration.

    Args:
        item_cfg: Stage item configuration with potential artifact_id specification
        default_input: Default input artifact if no specific artifact_id configured
        available_outputs: Available output artifacts from previous stages (optional)

    Returns:
        Input artifact for the action (either specified or default)
    """
    if available_outputs and item_cfg.artifact_id and item_cfg.artifact_id in available_outputs:
        return available_outputs[item_cfg.artifact_id]
    return default_input

_get_stage_name(stage_type)

Generate stage name from stage type.

Parameters:

Name	Type	Description	Default
stage_type	StageType	Type of stage	required

Returns:

Type	Description
str	Formatted stage name for pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _get_stage_name(self, stage_type: StageType) -> str:
    """Generate stage name from stage type.

   Args:
       stage_type: Type of stage

   Returns:
       Formatted stage name for pipeline
   """
    return f"{stage_type.name.replace("_", "-")}-STAGE"

_set_config()

Create and validate the CodePipeline configuration.

Merges builder configuration with usage to create a validated CodePipelineConfig object for pipeline creation.

Raises:

Type	Description
ValidationError	If the CodePipelineConfig validation fails

Side Effects

Sets self._config to validated CodePipelineConfig instance

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def _set_config(self) -> None:
    """Create and validate the CodePipeline configuration.

   Merges builder configuration with usage to create a validated
   CodePipelineConfig object for pipeline creation.

   Raises:
       ValidationError: If the CodePipelineConfig validation fails

   Side Effects:
       Sets self._config to validated CodePipelineConfig instance
   """
    try:
        self._config = CodePipelineConfig(**{
            **self._builder_config,
            "usage": self._usage
        })
    except ValidationError as e:
        self._log_validation_error(e, CodePipelineConfig)
        raise

build(scope)

Build and return the configured CodePipeline.

Creates pipeline with base configuration, builds stages according to usage pattern, and applies organizational tags.

Parameters:

Name	Type	Description	Default
scope	Construct	CDK construct scope where the pipeline will be created	required

Returns:

Type	Description
IPipeline	Configured CodePipeline instance

Raises:

Type	Description
ValidationError	If pipeline configuration validation fails

Side Effects

Creates CodePipeline with all configured stages and actions Applies organizational tags to the pipeline

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def build(self, scope: Construct) -> codepipeline.IPipeline:
    """Build and return the configured CodePipeline.

    Creates pipeline with base configuration, builds stages according to usage pattern,
    and applies organizational tags.

    Args:
        scope: CDK construct scope where the pipeline will be created

    Returns:
        Configured CodePipeline instance

    Raises:
        ValidationError: If pipeline configuration validation fails

    Side Effects:
        Creates CodePipeline with all configured stages and actions
        Applies organizational tags to the pipeline
    """
    super().build()

    pipeline = self._create_base_pipeline(scope)
    self._build_pipeline_stages(scope, pipeline)
    self._tag_resource(pipeline)

    return pipeline

reset()

Reset the builder to its initial state.

Side Effects

Resets internal builder state via parent class

Source code in mare_aws_common_lib/builders/code_pipeline_builder.py

def reset(self) -> None:
    """Reset the builder to its initial state.

    Side Effects:
        Resets internal builder state via parent class
    """
    super().reset()
    self._orchestrator = PipelineOrchestrator(self)
    self._action_factory = ActionFactory(self)