11# SPDX-License-Identifier: MPL-2.0
22# Copyright 2020-2025 John Mille <[email protected] > 3-
3+ import botocore . client
44from boto3 .session import Session
5- from compose_x_common .aws .arns import ARNS_PER_TAGGINGAPI_TYPE
6- from compose_x_common .compose_x_common import keyisset
5+ from compose_x_common .aws .arns import ARNS_PER_CFN_TYPE , ARNS_PER_TAGGINGAPI_TYPE
6+ from compose_x_common .compose_x_common import keyisset , set_else_none
77
88from ecs_composex .common .aws import find_aws_resource_arn_from_tags_api
99from ecs_composex .common .logging import LOG
@@ -82,6 +82,55 @@ def validate_subnets_belong_with_vpc(
8282 )
8383
8484
85+ def lookup_vpc_id (vpc_id_details : dict , lookup_session : Session ) -> str :
86+ """
87+ Function to find the VPC either by ID, Arn or Tags. Arn takes priority, then ID, then Tags
88+ """
89+ vpc_id = set_else_none ("Identifier" , vpc_id_details )
90+ vpc_arn = set_else_none ("Arn" , vpc_id_details )
91+ vpc_tags = set_else_none (TAGS_KEY , vpc_id_details )
92+ arn_from_arn = True if vpc_arn and not vpc_id else False
93+
94+ if vpc_arn :
95+ vpc_re = ARNS_PER_CFN_TYPE ["AWS::EC2::VPC" ]
96+ if not vpc_re .match (vpc_arn ):
97+ raise ValueError (f"{ vpc_arn } is not a valid VPC ARN" )
98+ vpc_id = vpc_re .match (vpc_arn ).group ("id" )
99+
100+ if vpc_id :
101+ cloud_control_client = lookup_session .client ("cloudcontrol" )
102+ try :
103+ cloud_control_client .get_resource (
104+ TypeName = "AWS::EC2::VPC" ,
105+ Identifier = vpc_id ,
106+ )
107+ except botocore .client .ClientError :
108+ LOG .error (f"{ vpc_id } is not a valid VPC ID" )
109+ return None
110+ if arn_from_arn :
111+ return vpc_arn
112+ else :
113+ ec2_client = lookup_session .client ("ec2" )
114+ sts_client = lookup_session .client ("sts" )
115+ account_id = sts_client .get_caller_identity ()["Account" ]
116+ return (
117+ f"arn:aws:ec2:{ ec2_client .meta .region_name } :{ account_id } :vpc/{ vpc_id } "
118+ )
119+
120+ elif vpc_tags :
121+ return find_aws_resource_arn_from_tags_api (
122+ vpc_id_details ,
123+ lookup_session ,
124+ "ec2:vpc" ,
125+ allow_multi = False ,
126+ )
127+ raise LookupError (
128+ "Failed to find VPC with given details: {}" .format (
129+ vpc_id or vpc_arn or vpc_tags
130+ )
131+ )
132+
133+
85134def lookup_x_vpc_settings (vpc_resource ):
86135 """
87136 Method to set VPC settings from x-vpc
@@ -103,11 +152,9 @@ def lookup_x_vpc_settings(vpc_resource):
103152 APP_SUBNETS .title ,
104153 STORAGE_SUBNETS .title ,
105154 ]
106- vpc_arn = find_aws_resource_arn_from_tags_api (
155+ vpc_arn = lookup_vpc_id (
107156 vpc_resource .lookup [VPC_ID .title ],
108157 vpc_resource .lookup_session ,
109- vpc_type ,
110- allow_multi = False ,
111158 )
112159 vpc_re = ARNS_PER_TAGGINGAPI_TYPE [vpc_type ]
113160 vpc_settings = {
0 commit comments