基于REPO_SOURCES的source_type动态选择CDK流水线源操作的实现逻辑

实现思路 #

遍历REPO_SOURCES字典列表中的每一项配置，通过source_type字段判断仓库类型，动态实例化对应的源操作类，最终将所有源操作实例组成的列表传入CDK Pipeline的source_action参数。

完整可运行代码 #

from aws_cdk import core, aws_codepipeline as codepipeline, aws_codepipeline_actions as cpactions, aws_codecommit as codecommit
from aws_cdk import pipelines

# 可按需扩展多个仓库配置
REPO_SOURCES = [
    {"repo": "repo-name-1", "branch": "master", "source_type": "github"},
    {"repo": "repo-name-2", "branch": "main", "source_type": "codecommit"}
]

class YourPipelineStack(core.Stack):
    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)
        
        cloud_assembly_artifact = codepipeline.Artifact("CloudAssembly")
        source_artifacts = []
        source_actions = []

        for source_config in REPO_SOURCES:
            # 每个源生成独立输出artifact，避免冲突
            source_artifact = codepipeline.Artifact(f"Source-{source_config['repo']}")
            source_artifacts.append(source_artifact)
            
            if source_config["source_type"] == "github":
                # 实例化GitHub源操作
                action = cpactions.GitHubSourceAction(
                    action_name=f"GitHub-{source_config['repo']}",
                    output=source_artifact,
                    oauth_token=core.SecretValue.secrets_manager("github-token"),
                    owner="你的GitHub组织/用户名", # 也可配置到REPO_SOURCES的字段中灵活取值
                    repo=source_config["repo"],
                    branch=source_config["branch"],
                    trigger=cpactions.GitHubTrigger.POLL
                )
            else:
                # 实例化CodeCommit源操作
                action = cpactions.CodeCommitSourceAction(
                    repository=codecommit.Repository.from_repository_name(
                        self,
                        id=f"CodeCommitRepo-{source_config['repo']}",
                        repository_name=source_config["repo"]
                    ),
                    branch=source_config["branch"],
                    action_name=f"CodeCommit-{source_config['repo']}",
                    trigger=cpactions.CodeCommitTrigger.POLL,
                    output=source_artifact
                )
            source_actions.append(action)
        
        # 初始化CDK Pipeline
        pipeline = pipelines.CdkPipeline(
            self, "CdkPipeline",
            cloud_assembly_artifact=cloud_assembly_artifact,
            pipeline_name="自定义Pipeline名称",
            source_actions=source_actions
            # 其余业务参数按需求补充即可
        )

注意事项 #

• 多仓库配置场景必须为每个源生成独立的Artifact实例，否则会出现输出冲突导致Pipeline运行失败
• CodeCommit资源ID、源操作的action_name需全局唯一，代码中通过拼接仓库名的方式实现，避免CDK部署时报资源重复错误
• GitHub个人访问令牌需要提前存入AWS Secrets Manager，存储名称和代码中secrets_manager的参数保持一致
• 如果不同GitHub仓库属于不同所有者，可在REPO_SOURCES中新增owner字段，代码中对应替换取值即可

内容的提问来源于stack exchange，提问作者Sameed Usmani