背景

我的环境中有一个用python写的升级工具，然后现在有个需求是要动态创建和删除alb的转发规则，带有特定cookie的转发到newnginx-albtest服务器组，不带的转发到oldnginx-albtest服务器组。因为要集成到升级工具代码中，所以采用了调用官网SDK的方式来实现。

前提

已经创建了alb，并配置了监听，用于获取监听id
最好有一个专属的ram账号用于创建和删除alb规则，用于获取ak和sk
已经配置好了用于访问不同版本服务的服务器组

官网API调试地址：
https://api.aliyun.com/api/Alb/2020-06-16/CreateRule
https://api.aliyun.com/api/Alb/2020-06-16/ListRules
https://api.aliyun.com/api/Alb/2020-06-16/DeleteRule

create alb rule

规则内容是带有cookie：companyid=123的，路径是/，转发到newnginx-albtest服务器组。

# -*- coding: utf-8 -*-
import sys
from typing import List, Optional

from alibabacloud_alb20200616 import models as alb_20200616_models
from alibabacloud_alb20200616.client import Client as Alb20200616Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient

# ==================== 配置常量 ====================
# AK/SK认证信息
ACCESS_KEY_ID = "xxx" # AK
ACCESS_KEY_SECRET = "xxx" # SK
ENDPOINT = "alb.cn-hangzhou.aliyuncs.com"

# ALB资源ID
LISTENER_ID = "xxx" # alb监听的ID
SERVER_GROUP_ID = "xxx"  # newnginx-albtest服务器组ID

# 规则配置
RULE_NAME = "cookie_alb_rule"
PRIORITY = 1
COOKIE_KEY = "companyid"
COOKIE_VALUE = "123"
PATH = "/"
TAG_KEY = "cookie"
TAG_VALUE = "true"

# ==================== 配置常量结束 ====================

class ALBCookieRuleCreator:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> Alb20200616Client:
        """
        创建ALB客户端
        @return: Client
        """
        config = open_api_models.Config()
        config.access_key_id = ACCESS_KEY_ID
        config.access_key_secret = ACCESS_KEY_SECRET
        config.endpoint = ENDPOINT
        return Alb20200616Client(config)

    @staticmethod
    def create_cookie_path_rule(client: Alb20200616Client) -> Optional[str]:
        """
        创建包含Cookie和路径条件的规则
        @param client: ALB客户端
        @return: 规则ID，如果创建失败则返回None
        """
        try:
            print(f"开始创建ALB转发规则: {RULE_NAME}")
            print(f"  监听器ID: {LISTENER_ID}")
            print(f"  优先级: {PRIORITY}")
            print(f"  条件: Cookie {COOKIE_KEY}={COOKIE_VALUE} AND 路径={PATH}")
            print(f"  转发到服务器组: {SERVER_GROUP_ID}")
            print(f"  标签: {TAG_KEY}={TAG_VALUE}")

            # 创建标签
            tag_0 = alb_20200616_models.CreateRuleRequestTag(
                key=TAG_KEY,
                value=TAG_VALUE
            )

            # 创建Cookie条件
            rule_conditions_0_cookie_config_values_values_0 = alb_20200616_models.CreateRuleRequestRuleConditionsCookieConfigValues(
                key=COOKIE_KEY,
                value=COOKIE_VALUE
            )

            rule_conditions_0_cookie_config = alb_20200616_models.CreateRuleRequestRuleConditionsCookieConfig(
                values=[rule_conditions_0_cookie_config_values_values_0]
            )

            rule_conditions_0 = alb_20200616_models.CreateRuleRequestRuleConditions(
                type='Cookie',
                cookie_config=rule_conditions_0_cookie_config
            )

            # 创建路径条件
            rule_conditions_1_path_config = alb_20200616_models.CreateRuleRequestRuleConditionsPathConfig(
                values=[PATH]
            )

            rule_conditions_1 = alb_20200616_models.CreateRuleRequestRuleConditions(
                path_config=rule_conditions_1_path_config,
                type='Path'
            )

            # 创建服务器组配置
            rule_actions_0_forward_group_config_server_group_tuples_server_group_tuples_0 = alb_20200616_models.CreateRuleRequestRuleActionsForwardGroupConfigServerGroupTuples(
                server_group_id=SERVER_GROUP_ID
            )

            rule_actions_0_forward_group_config = alb_20200616_models.CreateRuleRequestRuleActionsForwardGroupConfig(
                server_group_tuples=[
                    rule_actions_0_forward_group_config_server_group_tuples_server_group_tuples_0
                ]
            )

            rule_actions_0 = alb_20200616_models.CreateRuleRequestRuleActions(
                forward_group_config=rule_actions_0_forward_group_config,
                order=1,
                type='ForwardGroup'
            )

            # 创建请求
            create_rule_request = alb_20200616_models.CreateRuleRequest(
                listener_id=LISTENER_ID,
                priority=PRIORITY,
                rule_actions=[rule_actions_0],
                rule_conditions=[
                    rule_conditions_0,
                    rule_conditions_1
                ],
                rule_name=RULE_NAME,
                tag=[tag_0]
            )

            runtime = util_models.RuntimeOptions()

            # 调用API创建规则
            response = client.create_rule_with_options(create_rule_request, runtime)

            print(f"ALB转发规则创建成功！")
            print(f"  规则ID: {response.body.rule_id}")

            return response.body.rule_id

        except Exception as error:
            print(f"ALB转发规则创建失败！错误信息: {error.message}")
            if hasattr(error, 'data') and error.data:
                print(f"诊断地址: {error.data.get('Recommend')}")
            UtilClient.assert_as_string(error.message)
            return None

    @staticmethod
    def main(
            args: List[str],
    ) -> None:
        # 初始化客户端
        client = ALBCookieRuleCreator.create_client()

        print("=" * 60)
        print("开始创建ALB转发规则")
        print("=" * 60)

        # 创建Cookie+路径规则（条件：Cookie companyid=123 AND 路径=/）
        rule_id = ALBCookieRuleCreator.create_cookie_path_rule(client)

        print("=" * 60)
        if rule_id:
            print(f"规则创建成功！规则ID: {rule_id}")
        else:
            print("规则创建失败！")

    @staticmethod
    async def main_async(
            args: List[str],
    ) -> None:
        # 异步版本的main函数
        client = ALBCookieRuleCreator.create_client()

        try:
            print(f"开始创建ALB转发规则: {RULE_NAME}")

            # 创建标签
            tag_0 = alb_20200616_models.CreateRuleRequestTag(
                key=TAG_KEY,
                value=TAG_VALUE
            )

            # 创建Cookie条件
            rule_conditions_0_cookie_config_values_values_0 = alb_20200616_models.CreateRuleRequestRuleConditionsCookieConfigValues(
                key=COOKIE_KEY,
                value=COOKIE_VALUE
            )

            rule_conditions_0_cookie_config = alb_20200616_models.CreateRuleRequestRuleConditionsCookieConfig(
                values=[rule_conditions_0_cookie_config_values_values_0]
            )

            rule_conditions_0 = alb_20200616_models.CreateRuleRequestRuleConditions(
                type='Cookie',
                cookie_config=rule_conditions_0_cookie_config
            )

            # 创建路径条件
            rule_conditions_1_path_config = alb_20200616_models.CreateRuleRequestRuleConditionsPathConfig(
                values=[PATH]
            )

            rule_conditions_1 = alb_20200616_models.CreateRuleRequestRuleConditions(
                path_config=rule_conditions_1_path_config,
                type='Path'
            )

            # 创建服务器组配置
            rule_actions_0_forward_group_config_server_group_tuples_server_group_tuples_0 = alb_20200616_models.CreateRuleRequestRuleActionsForwardGroupConfigServerGroupTuples(
                server_group_id=SERVER_GROUP_ID
            )

            rule_actions_0_forward_group_config = alb_20200616_models.CreateRuleRequestRuleActionsForwardGroupConfig(
                server_group_tuples=[
                    rule_actions_0_forward_group_config_server_group_tuples_server_group_tuples_0
                ]
            )

            rule_actions_0 = alb_20200616_models.CreateRuleRequestRuleActions(
                forward_group_config=rule_actions_0_forward_group_config,
                order=1,
                type='ForwardGroup'
            )

            # 创建请求
            create_rule_request = alb_20200616_models.CreateRuleRequest(
                listener_id=LISTENER_ID,
                priority=PRIORITY,
                rule_actions=[rule_actions_0],
                rule_conditions=[
                    rule_conditions_0,
                    rule_conditions_1
                ],
                rule_name=RULE_NAME,
                tag=[tag_0]
            )

            runtime = util_models.RuntimeOptions()

            response = await client.create_rule_with_options_async(create_rule_request, runtime)

            print(f"ALB转发规则创建成功！")
            print(f"规则ID: {response.body.rule_id}")

        except Exception as error:
            print(f"ALB转发规则创建失败！错误信息: {error.message}")
            if hasattr(error, 'data') and error.data:
                print(f"诊断地址: {error.data.get('Recommend')}")
            UtilClient.assert_as_string(error.message)

if __name__ == '__main__':
    ALBCookieRuleCreator.main(sys.argv[1:])

执行脚本：

规则创建成功：

测试访问：

delete alb rule

先查询带有cookie=true标签的规则，获取rule id，然后删除这个规则。

# -*- coding: utf-8 -*-
# This file is auto-generated, don't edit it. Thanks.
import sys
from typing import List, Dict

from alibabacloud_alb20200616 import models as alb_20200616_models
from alibabacloud_alb20200616.client import Client as Alb20200616Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient

# ==================== 配置常量 ====================
# AK/SK认证信息
ACCESS_KEY_ID = "xxx"
ACCESS_KEY_SECRET = "xxx"
ENDPOINT = "alb.cn-hangzhou.aliyuncs.com"

# 监听器ID（可选）
LISTENER_ID = "xxx"

# 标签配置
TAG_KEY = "cookie"  # 标签键
TAG_VALUE = "true"  # 标签值

# ==================== 配置常量结束 ====================

class ALBRuleDeleter:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> Alb20200616Client:
        """
        创建ALB客户端
        @return: Client
        """
        config = open_api_models.Config()
        config.access_key_id = ACCESS_KEY_ID
        config.access_key_secret = ACCESS_KEY_SECRET
        config.endpoint = ENDPOINT
        return Alb20200616Client(config)

    @staticmethod
    def list_rules_by_tag(client: Alb20200616Client) -> List[Dict]:
        """
        根据标签查询规则
        @param client: ALB客户端
        @return: 规则列表，每个规则包含rule_id和rule_name
        """
        # 创建标签条件
        tag_0 = alb_20200616_models.ListRulesRequestTag(
            key=TAG_KEY,
            value=TAG_VALUE
        )

        # 创建查询请求
        list_rules_request = alb_20200616_models.ListRulesRequest(
            tag=[tag_0]
        )

        # 如果配置了监听器ID，则添加到查询条件中
        if LISTENER_ID:
            list_rules_request.listener_ids = [LISTENER_ID]

        runtime = util_models.RuntimeOptions()

        try:
            # 查询规则
            response = client.list_rules_with_options(list_rules_request, runtime)

            if response and response.body and response.body.rules:
                rules = []
                print(f"找到 {len(response.body.rules)} 个带有标签 '{TAG_KEY}={TAG_VALUE}' 的规则")

                for rule in response.body.rules:
                    rule_info = {
                        "rule_id": rule.rule_id,
                        "rule_name": rule.rule_name,
                        "listener_id": rule.listener_id
                    }
                    rules.append(rule_info)
                    print(f"  - 规则ID: {rule.rule_id}, 名称: {rule.rule_name}")

                return rules
            else:
                print(f"未找到带有标签 '{TAG_KEY}={TAG_VALUE}' 的规则")
                return []

        except Exception as error:
            print(f"查询规则失败！错误信息: {error.message}")
            if hasattr(error, 'data') and error.data:
                print(f"诊断地址: {error.data.get('Recommend')}")
            UtilClient.assert_as_string(error.message)
            return []

    @staticmethod
    def delete_rule(client: Alb20200616Client, rule_id: str) -> bool:
        """
        删除规则
        @param client: ALB客户端
        @param rule_id: 规则ID
        @return: 是否成功
        """
        try:
            # 创建删除请求
            delete_rule_request = alb_20200616_models.DeleteRuleRequest(
                rule_id=rule_id
            )

            runtime = util_models.RuntimeOptions()

            # 调用API删除规则
            response = client.delete_rule_with_options(delete_rule_request, runtime)

            print(f"规则删除成功: {rule_id}")
            return True

        except Exception as error:
            print(f"删除规则失败: {rule_id}, 错误信息: {error.message}")
            if hasattr(error, 'data') and error.data:
                print(f"诊断地址: {error.data.get('Recommend')}")
            UtilClient.assert_as_string(error.message)
            return False

    @staticmethod
    def delete_rules_by_tag(client: Alb20200616Client) -> None:
        """
        根据标签删除规则（直接删除，无需确认）
        @param client: ALB客户端
        """
        # 1. 查询规则
        print(f"查询带有标签 '{TAG_KEY}={TAG_VALUE}' 的规则...")
        rules = ALBRuleDeleter.list_rules_by_tag(client)

        if not rules:
            print("没有找到需要删除的规则。")
            return

        # 2. 批量删除规则
        print(f"\n开始删除 {len(rules)} 个规则...")
        success_count = 0
        failed_rules = []

        for rule in rules:
            if ALBRuleDeleter.delete_rule(client, rule['rule_id']):
                success_count += 1
            else:
                failed_rules.append(rule['rule_id'])

        # 3. 显示删除结果
        print(f"\n删除完成!")
        print(f"成功删除: {success_count} 个规则")
        if failed_rules:
            print(f"删除失败: {len(failed_rules)} 个规则")
            for rule_id in failed_rules:
                print(f"  - {rule_id}")

    @staticmethod
    def main(
            args: List[str],
    ) -> None:
        # 初始化客户端
        client = ALBRuleDeleter.create_client()

        print("=" * 60)
        print(f"开始删除带有标签 '{TAG_KEY}={TAG_VALUE}' 的ALB规则")
        if LISTENER_ID:
            print(f"监听器ID: {LISTENER_ID}")
        print("=" * 60)

        # 直接删除规则（无需确认）
        ALBRuleDeleter.delete_rules_by_tag(client)

        print("=" * 60)

    @staticmethod
    async def main_async(
            args: List[str],
    ) -> None:
        # 异步版本的main函数
        client = ALBRuleDeleter.create_client()

        print("正在查询规则...")

        # 创建标签条件
        tag_0 = alb_20200616_models.ListRulesRequestTag(
            key=TAG_KEY,
            value=TAG_VALUE
        )

        # 创建查询请求
        list_rules_request = alb_20200616_models.ListRulesRequest(
            tag=[tag_0]
        )

        # 如果配置了监听器ID，则添加到查询条件中
        if LISTENER_ID:
            list_rules_request.listener_ids = [LISTENER_ID]

        runtime = util_models.RuntimeOptions()

        try:
            # 查询规则
            response = await client.list_rules_with_options_async(list_rules_request, runtime)

            if response and response.body and response.body.rules:
                rules = []
                print(f"找到 {len(response.body.rules)} 个带有标签 '{TAG_KEY}={TAG_VALUE}' 的规则:")

                for rule in response.body.rules:
                    rules.append(rule.rule_id)
                    print(f"  - 规则ID: {rule.rule_id}")

                # 删除规则
                print(f"\n开始删除规则...")
                success_count = 0

                for rule_id in rules:
                    delete_rule_request = alb_20200616_models.DeleteRuleRequest(
                        rule_id=rule_id
                    )

                    try:
                        await client.delete_rule_with_options_async(delete_rule_request, runtime)
                        print(f"规则删除成功: {rule_id}")
                        success_count += 1
                    except Exception as error:
                        print(f"删除规则失败: {rule_id}, 错误信息: {error.message}")

                print(f"\n删除完成! 成功删除: {success_count}/{len(rules)} 个规则")
            else:
                print(f"未找到带有标签 '{TAG_KEY}={TAG_VALUE}' 的规则")

        except Exception as error:
            print(f"查询规则失败！错误信息: {error.message}")
            if hasattr(error, 'data') and error.data:
                print(f"诊断地址: {error.data.get('Recommend')}")
            UtilClient.assert_as_string(error.message)

if __name__ == '__main__':
    ALBRuleDeleter.main(sys.argv[1:])

执行脚本：