You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何通过boto3获取VPC对等连接中接受方的区域信息?

我之前也踩过这个坑!describe_vpc_peering_connections接口本身确实不会直接返回接受方VPC的可用区、子网这些细节,但它其实已经提供了关键的区域信息——你可能没注意到AccepterVpcInfo里的Region字段?如果你的SDK版本足够新,这个字段是默认返回的。如果找不到,大概率是boto3版本太旧,先升级再说。

第一步:提取接受方的区域信息

先检查你的返回结果,每个对等连接的AccepterVpcInfoRequesterVpcInfo里都包含Region字段,直接提取即可:

connections = self.client.describe_vpc_peering_connections()["VpcPeeringConnections"]
for conn in connections:
    accepter_region = conn['AccepterVpcInfo']['Region']
    requester_region = conn['RequesterVpcInfo']['Region']
    print(f"对等连接 {conn['VpcPeeringConnectionId']}: 请求方区域 {requester_region},接受方区域 {accepter_region}")

如果这里找不到Region字段,执行pip install --upgrade boto3升级SDK,这个字段在较新的版本里是默认返回的。

第二步:获取可用区和子网信息

拿到区域后,要获取可用区和子网,就得针对每个VPC所在的区域调用对应的EC2接口:

  • 子网属于特定VPC,用describe_subnets通过VPC ID过滤查询;
  • 可用区可以从子网的AvailabilityZone字段提取(每个子网绑定一个可用区),或者直接调用describe_availability_zones获取区域下的所有可用区。

下面是完整的实现代码,能帮你生成完整的VPC拓扑:

def get_full_vpc_peering_topology(self):
    # 获取所有对等连接
    peering_conns = self.client.describe_vpc_peering_connections()["VpcPeeringConnections"]
    topology = []

    for conn in peering_conns:
        conn_id = conn["VpcPeeringConnectionId"]
        # 提取双方VPC基础信息
        requester_vpc = conn["RequesterVpcInfo"]
        accepter_vpc = conn["AccepterVpcInfo"]

        # 获取区域(兼容旧SDK版本,默认用当前客户端区域)
        req_region = requester_vpc.get("Region", self.client.meta.region_name)
        acc_region = accepter_vpc.get("Region", self.client.meta.region_name)

        # 创建对应区域的EC2客户端
        req_client = self._get_ec2_client(req_region)
        acc_client = self._get_ec2_client(acc_region)

        # 查询请求方VPC的子网和可用区
        req_subnets = req_client.describe_subnets(
            Filters=[{"Name": "vpc-id", "Values": [requester_vpc["VpcId"]]}]
        )["Subnets"]
        req_azs = list({sub["AvailabilityZone"] for sub in req_subnets})

        # 查询接受方VPC的子网和可用区
        acc_subnets = acc_client.describe_subnets(
            Filters=[{"Name": "vpc-id", "Values": [accepter_vpc["VpcId"]]}]
        )["Subnets"]
        acc_azs = list({sub["AvailabilityZone"] for sub in acc_subnets})

        # 整理拓扑数据
        topology.append({
            "vpc_peering_id": conn_id,
            "requester_vpc": {
                "vpc_id": requester_vpc["VpcId"],
                "region": req_region,
                "availability_zones": req_azs,
                "subnets": [sub["SubnetId"] for sub in req_subnets]
            },
            "accepter_vpc": {
                "vpc_id": accepter_vpc["VpcId"],
                "region": acc_region,
                "availability_zones": acc_azs,
                "subnets": [sub["SubnetId"] for sub in acc_subnets]
            }
        })

    return topology

# 辅助方法:获取指定区域的EC2客户端,避免重复创建
def _get_ec2_client(self, region):
    if region == self.client.meta.region_name:
        return self.client
    return boto3.client("ec2", region_name=region)

关键注意事项

  • 权限问题:确保你的IAM角色拥有所有涉及区域的ec2:DescribeVpcsec2:DescribeSubnetsec2:DescribeVpcPeeringConnections权限,如果是跨账户对等连接,还需要接受方账户授予对应的跨账户访问权限;
  • 跨区域处理:上面的代码自动处理跨区域的情况,会为不同区域创建独立的EC2客户端;
  • 去重可用区:用集合去重是因为一个VPC可能在同一个可用区有多个子网,我们只需要列出一次可用区即可。

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

火山引擎 最新活动