You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

AWS CDK:创建工具类以避免跨栈引用问题

当然可以,完全不需要把这个逻辑做成CDK Stack。你可以把它封装成独立的Construct类(符合CDK最佳实践)或者纯工具函数,既能从SSM Parameter Store读取配置,又能避免跨引用问题,还能在多个仓库的Stack中复用。


方案一:封装为独立的Construct类(推荐)

这个类继承cdk.Construct而非cdk.Stack,职责是从SSM读取VPC相关参数、生成ec2.IVpc实例供其他Stack使用,本身不会部署任何资源。

from aws_cdk import aws_ec2 as ec2
from aws_cdk import aws_ssm as ssm
from aws_cdk import core as cdk

class VpcLookupUtility(cdk.Construct):
    def __init__(self, scope: cdk.Construct, construct_id: str, env_name: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        # 从SSM按环境路径读取参数(假设参数路径格式为 /{env}/vpc/xxx)
        self.vpc_id = ssm.StringParameter.from_string_parameter_attributes(
            self, "VpcIdParam",
            parameter_name=f"/{env_name}/vpc/id"
        ).string_value
        
        self.vpc_cidr = ssm.StringParameter.from_string_parameter_attributes(
            self, "VpcCidrParam",
            parameter_name=f"/{env_name}/vpc/cidr"
        ).string_value
        
        # 子网ID、可用区、路由表ID按逗号分隔存储,转为列表
        subnet_ids_str = ssm.StringParameter.from_string_parameter_attributes(
            self, "PrivateSubnetIdsParam",
            parameter_name=f"/{env_name}/vpc/private-subnet-ids"
        ).string_value
        self.subnet_list = subnet_ids_str.split(",")
        
        azs_str = ssm.StringParameter.from_string_parameter_attributes(
            self, "AvailabilityZonesParam",
            parameter_name=f"/{env_name}/vpc/availability-zones"
        ).string_value
        self.availability_zone_list = azs_str.split(",")
        
        route_table_ids_str = ssm.StringParameter.from_string_parameter_attributes(
            self, "RouteTableIdsParam",
            parameter_name=f"/{env_name}/vpc/private-route-table-ids"
        ).string_value
        self.route_table_id_list = route_table_ids_str.split(",")
        
        # 生成VPC引用对象
        self.vpc = ec2.Vpc.from_vpc_attributes(
            self, "LookupVpc",
            vpc_id=self.vpc_id,
            vpc_cidr_block=self.vpc_cidr,
            availability_zones=self.availability_zone_list,
            private_subnet_ids=self.subnet_list,
            private_subnet_route_table_ids=self.route_table_id_list
        )
        
        # 按需添加其他需要的属性,比如安全组ID
        self.vpc_api_gw_security_group_id = ssm.StringParameter.from_string_parameter_attributes(
            self, "ApiGwSgIdParam",
            parameter_name=f"/{env_name}/vpc/api-gw-sg-id"
        ).string_value

在其他CDK Stack中调用

在任意Stack里实例化这个工具类,直接使用它暴露的属性即可:

from aws_cdk import core as cdk
from aws_cdk import aws_ecs as ecs
from path.to.your.utility import VpcLookupUtility

class MyApplicationStack(cdk.Stack):
    def __init__(self, scope: cdk.Construct, construct_id: str, env_name: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        # 实例化工具类,传入环境名称
        vpc_util = VpcLookupUtility(self, "VpcLookup", env_name=env_name)
        
        # 使用获取到的VPC创建资源,比如ECS集群
        ecs_cluster = ecs.Cluster(
            self, "MyCluster",
            vpc=vpc_util.vpc,
            cluster_name=f"{env_name}-ecs-cluster"
        )
        
        # 直接使用其他属性,比如安全组ID
        api_gw_sg = ec2.SecurityGroup.from_security_group_id(
            self, "ApiGwSg",
            security_group_id=vpc_util.vpc_api_gw_security_group_id
        )

方案二:纯工具函数(更轻量)

如果不需要封装成Construct,也可以写一个纯函数完成VPC查找:

from aws_cdk import aws_ec2 as ec2
from aws_cdk import aws_ssm as ssm
from aws_cdk import core as cdk

def get_vpc_from_ssm(scope: cdk.Construct, env_name: str) -> ec2.IVpc:
    # 读取SSM参数
    vpc_id = ssm.StringParameter.from_string_parameter_attributes(
        scope, f"VpcId-{env_name}",
        parameter_name=f"/{env_name}/vpc/id"
    ).string_value
    
    vpc_cidr = ssm.StringParameter.from_string_parameter_attributes(
        scope, f"VpcCidr-{env_name}",
        parameter_name=f"/{env_name}/vpc/cidr"
    ).string_value
    
    subnet_ids = ssm.StringParameter.from_string_parameter_attributes(
        scope, f"SubnetIds-{env_name}",
        parameter_name=f"/{env_name}/vpc/private-subnet-ids"
    ).string_value.split(",")
    
    azs = ssm.StringParameter.from_string_parameter_attributes(
        scope, f"Azs-{env_name}",
        parameter_name=f"/{env_name}/vpc/availability-zones"
    ).string_value.split(",")
    
    route_table_ids = ssm.StringParameter.from_string_parameter_attributes(
        scope, f"RouteTableIds-{env_name}",
        parameter_name=f"/{env_name}/vpc/private-route-table-ids"
    ).string_value.split(",")
    
    # 返回VPC引用
    return ec2.Vpc.from_vpc_attributes(
        scope, f"LookupVpc-{env_name}",
        vpc_id=vpc_id,
        vpc_cidr_block=vpc_cidr,
        availability_zones=azs,
        private_subnet_ids=subnet_ids,
        private_subnet_route_table_ids=route_table_ids
    )

调用方式

在Stack中直接调用函数:

class MyAnotherStack(cdk.Stack):
    def __init__(self, scope: cdk.Construct, construct_id: str, env_name: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)
        
        vpc = get_vpc_from_ssm(self, env_name=env_name)
        
        # 使用vpc创建后续资源...

关键注意事项

  • SSM权限:运行CDK synth/deploy的用户/角色需要具备ssm:GetParameter权限,才能读取指定的SSM参数。
  • 参数格式:确保SSM中存储的子网ID、可用区等是逗号分隔的字符串,方便直接转为列表。
  • 无跨引用问题:这个工具类/函数只是引用已存在的资源,不会创建新的部署资源,因此不会产生跨Stack的依赖关系,多个仓库的Stack都可以直接引入使用。

内容的提问来源于stack exchange,提问作者Codistan

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.19 18:05:27