网站首页编程语言正文

使用python sdk添加删除阿里云pvc路由

作者：lv2 更新时间： 2023-07-07 编程语言

1. 前言

由于线路供应商sdwan存在单点问题，需要实现线路高可用解决方案，需要设计自动切换阿里云vpc路由解决方案。通过阿里云文档了解，可通过阿里云专有网络Python SDK，通过sdk实现创建、删除、查询等vpc网络相关操作（创建/删除路由表、创建/删除自定义路由等）。

环境准备

Python 2.7及3.x环境（这里使用python3.9.6)
配置阿里云账号和访问密钥（AccessKey），权限：AliyunVPCFullAccess
安装Alibaba Cloud SDK for Python

2. 安装

安装Alibaba Cloud SDK

# 安装模块
pip install alibabacloud_tea_openapi
pip install alibabacloud_vpc20160428==2.0.0
pip install aliyun-python-sdk-vpc==3.0.25

# 安装sdk示例库,完成环境初始化
git clone https://github.com/aliyun/aliyun-openapi-python-sdk-examples.git
cd aliyun-openapi-python-sdk-examples/sdk_examples/
python setup.py install

安装效果如下

3. 使用方法

在解压的目录.sdk_examples/examples/，可以看到除了vpc的py脚本，还有slb，eip等，根据需要可自行查看修改，这里使用sdk_examples/examples/vpc/vpc_route_entry.py进行修改，原代码为

#encoding=utf-8
import sys
import json
import time

from aliyunsdkcore.acs_exception.exceptions import ServerException, ClientException
from aliyunsdkvpc.request.v20160428 import CreateRouteEntryRequest
from aliyunsdkvpc.request.v20160428 import DeleteRouteEntryRequest
from aliyunsdkvpc.request.v20160428 import DescribeRouteTablesRequest
from sdk_lib.exception import ExceptionHandler
from sdk_lib.check_status import CheckStatus
from sdk_lib.common_util import CommonUtil
from sdk_lib.sdk_vswitch import VSwitch
from sdk_lib.sdk_route_table import RouteTable
from sdk_lib.consts import *

class RouteEntry(object):
    def __init__(self, client):
        self.client = client

    def create_route_entry(self, params):
        """
        create_route_entry: 创建route_entry路由条目
        官网API参考链接: https://help.aliyun.com/document_detail/36012.html
        """
        try:
            request = CreateRouteEntryRequest.CreateRouteEntryRequest()
            # 路由表ID
            request.set_RouteTableId(params['route_table_id'])
            # 自定义路由条目的目标网段
            request.set_DestinationCidrBlock(params['destination_cidr_block'])
            # 下一跳的类型
            request.set_NextHopType(params['nexthop_type'])
            # 下一跳实例的ID
            request.set_NextHopId(params['nexthop_id'])
            response = self.client.do_action_with_exception(request)
            response_json = json.loads(response)
            # 判断router entry状态是否可用
            if CheckStatus.check_status(TIME_DEFAULT_OUT, DEFAULT_TIME,
                                        self.describe_route_entry_status,
                                        AVAILABLE, params['route_table_id']):
                return response_json
        except ServerException as e:
            ExceptionHandler.server_exception(e)
        except ClientException as e:
            ExceptionHandler.client_exception(e)

    def delete_route_entry(self, params):
        """
        delete_route_entry: 删除route_entry路由条目
        官网API参考链接: https://help.aliyun.com/document_detail/36013.html
        """
        try:
            request = DeleteRouteEntryRequest.DeleteRouteEntryRequest()
            # 路由条目所在的路由表的ID
            request.set_RouteTableId(params['route_table_id'])
            # 路由条目的目标网段
            request.set_DestinationCidrBlock(params['destination_cidr_block'])
            # 下一跳实例的ID
            request.set_NextHopId(params['nexthop_id'])
            response = self.client.do_action_with_exception(request)
            response_json = json.loads(response)
            time.sleep(DEFAULT_TIME)
            return response_json
        except ServerException as e:
            ExceptionHandler.server_exception(e)
        except ClientException as e:
            ExceptionHandler.client_exception(e)

    def describe_route_entry_vrouter(self, params):
        """
        describe_route_entry_vrouter: 查询route_entry路由条目
        官网API参考链接: https://help.aliyun.com/document_detail/36014.html
        """
        try:
            request = DescribeRouteTablesRequest.DescribeRouteTablesRequest()
            # 路由表所属的VPC路由器或边界路由器的ID
            request.set_VRouterId(params['vrouter_id'])
            # 路由表的ID
            request.set_RouteTableId(params['route_table_id'])
            response = self.client.do_action_with_exception(request)
            response_json = json.loads(response)
            return response_json
        except ServerException as e:
            ExceptionHandler.server_exception(e)
        except ClientException as e:
            ExceptionHandler.client_exception(e)

    def describe_route_entry(self, route_table_id):
        """
        describe_route_entry: 查询route_entry路由条目
        官网API参考链接: https://help.aliyun.com/document_detail/36014.html
        """
        try:
            request = DescribeRouteTablesRequest.DescribeRouteTablesRequest()
            request.set_RouteTableId(route_table_id)
            response = self.client.do_action_with_exception(request)
            response_json = json.loads(response)
            return response_json
        except ServerException as e:
            ExceptionHandler.server_exception(e)
        except ClientException as e:
            ExceptionHandler.client_exception(e)

    def describe_route_entry_status(self, route_table_id):
        """
        describe_route_entry_status: 查询route_entry路由条目当前状态
        """
        response = self.describe_route_entry(route_table_id)
        return response["RouteTables"]["RouteTable"][0]["RouteEntrys"]["RouteEntry"][0]["Status"]

def main():
    client = ACS_CLIENT
    vswitch = VSwitch(client)
    route_table = RouteTable(client)
    route_entry = RouteEntry(client)

    params = {}
    params['route_table_name'] = "sdk_route_table"
    params['destination_cidr_block'] = "0.0.0.0/0"
    params['nexthop_id'] = "i-xxx"
    params['nexthop_type'] = "Instance"

    params['vpc_id'] = "vpc-xxx"
    params['vswitch_id'] = "vsw-xxx"

    #创建route table
    route_table_json = route_table.create_route_table(params)
    CommonUtil.log("create_route_table", route_table_json)

    #查询vswitch
    vswitch_json = vswitch.describe_vswitch_attribute(params)
    CommonUtil.log("describe_vswitch_attribute", vswitch_json)

    #route table绑定vswitch
    params['route_table_id'] = route_table_json['RouteTableId']
    associate_json = route_table.associate_route_table(params)
    CommonUtil.log("associate_route_table", associate_json)

    #创建路由条目
    create_route_entry_json = route_entry.create_route_entry(params)
    CommonUtil.log("create_route_entry", create_route_entry_json)
    
    #删除路由条目
    delete_route_entry_json = route_entry.delete_route_entry(params)
    CommonUtil.log("delete_route_entry", delete_route_entry_json)

    #route table解绑vswitch
    unassociate_json = route_table.unassociate_route_table(params)
    CommonUtil.log("unassociate_route_table", unassociate_json)

    #删除route table
    delete_route_table_json = route_table.delete_route_table(params)
    CommonUtil.log("delete_route_table", delete_route_table_json)


if __name__ == "__main__":
    sys.exit(main())

改为两个py脚本，一个添加路由，一个删除路由

添加路由脚本：add_route.py

#encoding=utf-8
import sys
import json
import time

from aliyunsdkcore.acs_exception.exceptions import ServerException, ClientException
from aliyunsdkvpc.request.v20160428 import CreateRouteEntryRequest
from aliyunsdkvpc.request.v20160428 import DeleteRouteEntryRequest
from aliyunsdkvpc.request.v20160428 import DescribeRouteTablesRequest
from sdk_lib.exception import ExceptionHandler
from sdk_lib.check_status import CheckStatus
from sdk_lib.common_util import CommonUtil
from sdk_lib.sdk_vswitch import VSwitch
from sdk_lib.sdk_route_table import RouteTable
from sdk_lib.consts import *

class RouteEntry(object):
    def __init__(self, client):
        self.client = client
    
    def create_route_entry(self, params):
        """
        create_route_entry: 创建route_entry路由条目

        """
        try:
            request = CreateRouteEntryRequest.CreateRouteEntryRequest()
            # 路由表ID
            request.set_RouteTableId(params['route_table_id'])
            # 自定义路由条目的目标网段
            request.set_DestinationCidrBlock(params['destination_cidr_block'])
            # 下一跳的类型
            request.set_NextHopType(params['nexthop_type'])
            # 下一跳实例的ID
            request.set_NextHopId(params['nexthop_id'])
            response = self.client.do_action_with_exception(request)
            response_json = json.loads(response)
            # 判断router entry状态是否可用
            if CheckStatus.check_status(TIME_DEFAULT_OUT, DEFAULT_TIME,
                                        self.describe_route_entry_status,
                                        AVAILABLE, params['route_table_id']):
                return response_json
        except ServerException as e:
            ExceptionHandler.server_exception(e)
        except ClientException as e:
            ExceptionHandler.client_exception(e)

    def describe_route_entry_status(self, route_table_id):
        """
        describe_route_entry_status: 查询route_entry路由条目当前状态

        """
        response = self.describe_route_entry(route_table_id)
        return response["RouteTables"]["RouteTable"][0]["RouteEntrys"]["RouteEntry"][0]["Status"]

    def describe_route_entry_vrouter(self, params):
        """
        describe_route_entry_vrouter: 查询route_entry路由条目

        """
        try:
            request = DescribeRouteTablesRequest.DescribeRouteTablesRequest()
            # 路由表所属的VPC路由器或边界路由器的ID
            request.set_VRouterId(params['vrouter_id'])
            # 路由表的ID
            request.set_RouteTableId(params['route_table_id'])
            response = self.client.do_action_with_exception(request)
            response_json = json.loads(response)
            return response_json
        except ServerException as e:
            ExceptionHandler.server_exception(e)
        except ClientException as e:
            ExceptionHandler.client_exception(e)

    def describe_route_entry(self, route_table_id):
        """
        describe_route_entry: 查询route_entry路由条目

        """
        try:
            request = DescribeRouteTablesRequest.DescribeRouteTablesRequest()
            request.set_RouteTableId(route_table_id)
            response = self.client.do_action_with_exception(request)
            response_json = json.loads(response)
            return response_json
        except ServerException as e:
            ExceptionHandler.server_exception(e)
        except ClientException as e:
            ExceptionHandler.client_exception(e)

def main():
    # 这里填上上面创建的阿里云帐号accesskey和secretkey，以及vpc地域
    client = AcsClient(
            '***',
            '***',
            'cn-guangzhou')
    vswitch = VSwitch(client)
    route_table = RouteTable(client)
    route_entry = RouteEntry(client)
    # 变量params参数，根据需要添加路由,pvc id，switch id和路由表名称
    params = {}
    params['route_table_name'] = "sdk_route_table"
    params['destination_cidr_block'] = "0.0.0.0/0"
    params['nexthop_id'] = "i-xxx"
    params['nexthop_type'] = "Instance"

    params['vpc_id'] = "vpc-xxx"
    params['vswitch_id'] = "vsw-xxx"


    #创建路由条目
    create_route_entry_json = route_entry.create_route_entry(params)
    CommonUtil.log("create_route_entry", create_route_entry_json)

if __name__ == "__main__":
    sys.exit(main())

使用方法：

python add_route.py

删除路由脚本：del_route.py

#encoding=utf-8
import sys
import json
import time

from aliyunsdkcore.acs_exception.exceptions import ServerException, ClientException
from aliyunsdkvpc.request.v20160428 import CreateRouteEntryRequest
from aliyunsdkvpc.request.v20160428 import DeleteRouteEntryRequest
from aliyunsdkvpc.request.v20160428 import DescribeRouteTablesRequest
from sdk_lib.exception import ExceptionHandler
from sdk_lib.check_status import CheckStatus
from sdk_lib.common_util import CommonUtil
from sdk_lib.sdk_vswitch import VSwitch
from sdk_lib.sdk_route_table import RouteTable
from sdk_lib.consts import *

class RouteEntry(object):
    def __init__(self, client):
        self.client = client
    
    def delete_route_entry(self, params):
        """
        delete_route_entry: 删除route_entry路由条目

        """
        try:
            request = DeleteRouteEntryRequest.DeleteRouteEntryRequest()
            # 路由条目所在的路由表的ID
            request.set_RouteTableId(params['route_table_id'])
            # 路由条目的目标网段
            request.set_DestinationCidrBlock(params['destination_cidr_block'])
            # 下一跳实例的ID
            request.set_NextHopId(params['nexthop_id'])
            response = self.client.do_action_with_exception(request)
            response_json = json.loads(response)
            time.sleep(DEFAULT_TIME)
            return response_json
        except ServerException as e:
            ExceptionHandler.server_exception(e)
        except ClientException as e:
            ExceptionHandler.client_exception(e)

 

def main():
    # 这里填上上面创建的阿里云帐号accesskey和secretkey，以及vpc地域
    client = AcsClient(
            '***',
            '***',
            'cn-guangzhou')
    vswitch = VSwitch(client)
    route_table = RouteTable(client)
    route_entry = RouteEntry(client)
    # 变量params参数，根据需要添加路由,pvc id，switch id和路由表名称
    params = {}
    params['route_table_name'] = "sdk_route_table"
    params['destination_cidr_block'] = "0.0.0.0/0"
    params['nexthop_id'] = "i-xxx"
    params['nexthop_type'] = "Instance"

    params['vpc_id'] = "vpc-xxx"
    params['vswitch_id'] = "vsw-xxx"


    #删除路由条目
    delete_route_entry_json = route_entry.delete_route_entry(params)
    CommonUtil.log("delete_route_entry", delete_route_entry_json)

if __name__ == "__main__":
    sys.exit(main())

使用方法：

python del_route.py

原文链接：https://blog.csdn.net/manwufeilong/article/details/131477339

网站首页 编程语言 正文

使用python sdk添加删除阿里云pvc路由

1. 前言

2. 安装

3. 使用方法

相关推荐

网站首页编程语言正文