OpenResty Edge™ Python SDK 用户手册

欢迎查阅 OpenResty Edge™ Python SDK 的用户手册。

# 安装

OpenResty Edge 的 Python SDK 依赖 Python3，请确保您已经安装好 3.6 及以上版本的 Python3。 您可以像安装其他 Python3 模块那样安装此 Python SDK：

1. 打开 Linux 终端或 Windows 命令提示符窗口

2. 输入以下命令安装 Python SDK：

   pip3 install openresty-edge-sdk --index-url https://pypi.org/simple
   

如果您要安装特定版本的 Python SDK，则可以在模块名称后加上版本号：

pip3 install openresty-edge-sdk==1.2.39 --index-url https://pypi.org/simple

安装命令执行成功后，您可以像下面这样验证是否已经正确安装：

$ python3
>>> from edge2client import Edge2Client
>>> client = Edge2Client('https://EDGE_ADMIN_HOST:EDGE_ADMIN_PORT', 'USERNAME', 'PASSWORD')
>>> client.login()
True
>>> client.get_version()
{'waf': {'waf_version': '3.3.4-1' ...}
>>> exit()

键入在 client.login() 后，如果输出 True，则表示已经正确安装，如果您的 Edge Admin 使用的是自签名证书（默认是自签名证书），那么你还需要在 client.login() 之前调用 client.set_ssl_verify(False)。 接着你还可以使用 client.get_version() 来查看 OpenResty Edge 的版本信息。

此外，你可以在这里看到此 Python SDK 的变更日志。

至此，安装已经全部完成，您可以参考本文档中后续的内容，以实现更为复杂的功能。

# 概述

import sys
import time
from edge2client import Edge2Client

def get_client():
    client = Edge2Client(host, username, password)
    # 如果是自签证书导致 SSL 校验失败，可以去掉下面的注释
    # client.set_ssl_verify(False)
    ok = client.login()

    return client

if __name__ == '__main__':
    try:
        client = get_client()
    except Exception as e:
        print("failed to get client: ", str(e))
        sys.exit(1)

    try:
        app_ids = client.get_all_apps()
    except Exception as e:
        print("failed to get all apps: ", str(e))
        sys.exit(1)

    for app_id in app_ids:
        client.use_app(app_id)
        if client.pending_changes() > 0:
            ret = client.new_release()
            if ret:
                print('release ok app id: ' + str(app_id))
            else:
                print('release failed: ' + str(app_id))
            time.sleep(1)

# 方法列表

# 方法详情

# Edge2Client

client = Edge2Client(host, username, password, api_token=None)

此方法用于初始化 SDK 客户端。

参数

返回值

• 返回一个 edge2client.Edge2Client 对象。

示例

1. 当 Edge Admin 的地址为 https://127.0.0.1:443，且使用用户名 username 和密码 password 初始化 Edge2Client 对象时：

   from edge2client import Edge2Client
   
   client = Edge2Client('https://127.0.0.1:443', 'username', 'password')
   

2. 使用 API Token api_token 来初始化 Edge2Client 对象时：

   from edge2client import Edge2Client
   
   client = Edge2Client('https://127.0.0.1:443', None, None, "api_token")
   

# set_ssl_verify

set_ssl_verify(verify)

设置访问 Edge Admin 时，是否进行 SSL 验证，默认会进行验证。 当您的 Edge Admin 使用的是自签名的证书时，那么需要把 SSL 验证关闭。

参数

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.set_ssl_verify(False)
ok = client.login()

# login

ok = login()

此方法用于登录 OpenResty Edge，以便进行后续操作。

返回值

• True：表示登录成功。
• False：表示登录失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)

ok = client.login()

# use_app

ok = use_app(app_id)

此方法用于指定当前实例操作的应用 ID。

返回值

• True：应用 ID 指定成功。
• False：应用 ID 指定失败。

示例

下面的示例演示如何将当前实例操作的应用 ID 更改为 10。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 10
client.use_app(app_id)

# new_app

app_id = new_app(domains, label)

此方法用于新增 HTTP 应用。

参数

返回值

• app_id：返回新建应用的 ID，其数据类型为 number。

示例

以下示例展示了如何创建一个域名为 orig.foo.com，标签为 origin site for foo.com 的 HTTP 应用，且该应用所在的集群组为 [1]。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = client.new_app(
    domains=['orig.foo.com'],
    label='origin site for foo.com',
    cluster_groups = [1])

# put_app

ok = put_app(app_id, domains, label, cluster_groups=None, offline=None)

此方法用于修改 HTTP 应用的基本信息。

参数

返回值

• True：表示修改成功。
• False：表示修改失败。

示例

以下示例展示了如何将已创建的应用（其域名为 orig.foo.com）的域名修改为 *.foo.com。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 10
client.use_app(app_id)

ok = client.put_app(app_id, domains=['*.foo.com'], label='foo.com')

# get_app

data = get_app()

此方法用于获取 HTTP 应用的相关信息，包括域名、标签以及所属群组等。

返回值

• data：返回一个 dict 类型的数据。

示例

以下示例演示了如何获取 app_id 为 1 的应用信息。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

data = client.get_app()

# put_app_config

put_app_config(app_id, limiter)

此方法用于修改 HTTP 应用的配置。

参数

返回值

• True：配置修改成功。
• False：配置修改失败。

示例

以下示例展示了如何修改 app_id 为 1 的应用配置。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1

res = client.put_app_config(app_id=app_id,
                            limiter={
                                        'max_uri_args': 130,
                                        'max_post_args': 140,
                                        'max_req_headers': 150,
                                        'check_post_args_type': True
                                    },
                            enable_websocket=True)

# get_app_config

data = get_app_config(app_id)

此方法用于获取指定 HTTP 应用的配置信息。

参数

返回值

• data：返回一个字典（dict）类型的数据。

示例

下例展示了如何获取 id 为 1 的应用的配置信息：

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1

data = client.get_app_config(app_id)

# del_app

success = del_app(app_id)

此方法根据提供的 app_id 删除指定的 HTTP 应用。

参数

返回值

• True：表示应用删除成功。
• False：表示应用删除失败。

示例

以下示例展示了如何删除 app_id 为 1 的应用。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1

success = client.del_app(app_id)

# search_app

data = search_app(app_domain)

此方法用于根据指定域名搜索 HTTP 应用。

参数

返回值

• data：返回一个 list 类型的结果，其中每个元素都是应用的相关信息。

示例

以下示例以 orig.foo.com 为域名来搜索 HTTP 应用。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.search_app(app_domain='orig.foo.com')

# search_http_app_by_keyword

data = search_http_app_by_keyword(keyword, page, pagesize)

此方法用于根据域名或应用名称搜索 HTTP 应用。

参数

返回值

• data：此返回值为 list 类型，每个元素包含应用的相关信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.search_http_app_by_keyword('orig.foo.com')
data = client.search_http_app_by_keyword('orig.foo.com', 1, 2)
data = client.search_http_app_by_keyword('origin')

# search_upstream_by_ip

data = search_upstream_by_ip(ip, page=1, pagesize=20)

此方法用于通过 IP 地址搜索上游信息。

参数

返回值

• data：返回一个list类型的数据，列表中的每一个元素代表一个上游的详细信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.search_upstream_by_ip('1.1.1.1')

# search_k8s_upstream_by_ip

data = search_k8s_upstream_by_ip(ip, page, pagesize)

这个方法用于根据 IP 搜索 Kubernetes 上游。

参数

返回值

• data：一个list类型的变量，其中包含了 Kubernetes 上游的信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.search_k8s_upstream_by_ip('1.1.1.1')

# search_upstream_by_name

data = search_upstream_by_name(name, page, pagesize)

此方法用于根据名称搜索上游。

参数

返回值

• data：此为list 类型，每个元素都代表了一条上游信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.search_upstream_by_name('test')

# search_k8s_upstream_by_name

data = search_k8s_upstream_by_name(name, page, pagesize)

此方法用于通过指定名称搜索 Kubernetes 上游。

参数

返回值

• data：此为 list 类型，包含多个 Kubernetes 上游的信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.search_k8s_upstream_by_name('test')

# search_k8s_upstream

data = search_k8s_upstream(namespace, service, port, page, pagesize, type_list)

此方法允许用户通过指定命名空间，服务名称和端口号来搜索 Kubernetes 上游。

参数

返回值

• data：list 类型，列表中每个元素都是 Kubernetes 上游的信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = self.client.search_k8s_upstream(port = 80, service = "test-hello", namespace = "default")

# search_k8s_upstream_history

data = search_k8s_upstream_history(page, pagesize, start_time, end_time)

此方法根据命名空间、服务名称和端口号搜索 Kubernetes 上游历史。

参数

返回值

• data：返回值为 list 类型，每个元素代表一个 Kubernetes 上游的更新历史操作。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.search_k8s_upstream_history(port = 80, service = "test-hello", namespace = "default")

# get_all_apps

data = get_all_apps(detail)

此方法用于获取所有应用的信息。

参数

返回值

• 当 detail 参数为 False 时，函数返回一个包含所有的 app id 的 list 类型的 Python 对象，包含所有应用的基本信息。
• 当 detail 参数为 True 时，函数返回一个以 app id 作为键值的 dict 类型的 Python 对象，包含所有应用的详细信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

# 获取基本信息
apps = client.get_all_apps()

# 获取详细信息
apps = client.get_all_apps(detail=True)

# new_upstream

up_id = new_upstream(**kwargs)

此方法用于新增上游，返回值为新增上游的 id。

参数

servers 参数

返回值

• up_id：表示创建成功的上游的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

up_id = client.new_upstream(
    name='origin-upstream',
    servers=[
        {'ip': '172.22.31.1', 'port': 80},
        {'ip': '172.22.31.2', 'port': 80, 'weight': 2}
    ])

# put_upstream

ok = put_upstream(**kwargs)

此方法用于修改上游信息。

参数

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

up_id = 1
ok = client.put_upstream(up_id,
    name='origin-upstream',
    servers=[
        {'domain': 'test.com', 'port': 8080},
        {'ip': '172.22.31.2', 'port': 8080, 'weight': 1}
    ],
    health_checker={
        'type': 'http',
        'http_req_uri': '/status',
        'http_req_host': 'test.com',
        'interval': 3,
        'interval_unit': 'sec',
        'timeout': 1,
        'fall': 3,
        'rise': 2,
        'valid_statuses': [200, 302],
        'report_interval': 3,
        'report_interval_unit': 'min'
    })

# get_upstream

data = get_upstream(up_id)

此方法用于获取上游信息。

参数

返回值

• data：返回类型为 dict 的上游信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

up_id = 1
data = client.get_upstream(up_id)

# del_upstream

del_upstream(up_id)

此方法用于删除上游。

参数

返回值

• 返回 True：表示上游删除成功。
• 返回 False：表示上游删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

up_id = 1

ok = client.del_upstream(up_id)

在这个示例中，我们首先导入了 Edge2Client，然后创建了一个 Edge2Client 对象，使用 host，username 和 password 进行登录。然后，我们尝试用 del_upstream 方法删除标识符为 1 的上游。

# get_all_upstreams

upstreams = client.get_all_upstreams()

此方法用于获取所有上游的详细信息。

返回值

• upstreams：返回一组list类型的数据，包含所有上游信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

upstreams = client.get_all_upstreams()

# new_k8s_upstream

up_id = new_k8s_upstream(**kwargs)

本函数用于新增全局 Kubernetes 上游，将返回新创建的 k8s 上游的 ID。

参数

k8s_services 参数

返回值

• up_id：新创建的 k8s 上游的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

k8s_up_id = client.new_k8s_upstream(
    name = 'k8s-upstream',
    k8s_services = [
        {
            'k8s' : 1,
            'k8s_namespace' : 'default',
            'k8s_service' : 'foo',
            'k8s_service_port' : 80,
        }
    ])

# copy_upstream_to_k8s_upstream

k8s_up_id = copy_upstream_to_k8s_upstream(up_id, k8s_services, rules)

此方法用于将一个常规上游复制为一个 Kubernetes 上游。

参数

返回值

• up_id：新创建的 Kubernetes 上游的 ID。

示例

示例一：

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = client.new_app(domains = ['orig.foo.com'], label = 'origin site for foo.com')

up_id = client.new_upstream(name = 'origin-upstream',
                            servers = [
                                {'ip': '172.22.31.1', 'port': 80},
                                {'ip': '172.22.31.2', 'port': 80, 'weight': 2}
                            ])

k8s_up_id = client.copy_upstream_to_k8s_upstream(up_id,
                            k8s_services = [
                                {
                                    'k8s' : 1,
                                    'k8s_namespace' : 'default',
                                    'k8s_service' : 'foo',
                                    'k8s_service_port' : 80,
                                }
                            ])

示例二：

client.use_app(app_id)

rules = client.get_all_rules()

k8s_up_id = client.copy_upstream_to_k8s_upstream(up_id = up_id,
                            k8s_services = [
                                {
                                    'k8s' : 1,
                                    'k8s_namespace' : 'default',
                                    'k8s_service' : 'foo',
                                    'k8s_service_port' : 80,
                                }
                            ],
                            rules = rules)

# put_k8s_upstream

ok = put_k8s_upstream(**kwargs)

此方法用于修改 Kubernetes 上游信息。

参数

返回值

• True：表示修改成功。
• False：表示修改失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

up_id = 1
ok = client.put_k8s_upstream(up_id,
    name='k8s-upstream',
    k8s_services = [
        {
            'k8s' : 1,
            'k8s_namespace' : 'default',
            'k8s_service' : 'foo',
            'k8s_service_port' : 80,
        }
    ],
    health_checker={
        'type': 'http',
        'http_req_uri': '/status',
        'http_req_host': 'test.com',
        'interval': 3,
        'interval_unit': 'sec',
        'timeout': 1,
        'fall': 3,
        'rise': 2,
        'valid_statuses': [200, 302],
        'report_interval': 3,
        'report_interval_unit': 'min'
    })

# get_k8s_upstream

data = get_k8s_upstream(up_id)

此方法用于获取 k8s 上游的信息。

参数

返回值

• data：返回一个 dict 类型的数据。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

up_id = 1
data = client.get_k8s_upstream(up_id)

# del_k8s_upstream

success = del_k8s_upstream(upstream_id)

此方法用于删除 k8s（Kubernetes）上游。

参数

返回值

• True：表示删除成功。
• False：表示删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

upstream_id = '1'
success = client.del_k8s_upstream(upstream_id)

# get_all_k8s_upstreams

upstreams = client.get_all_k8s_upstreams()

此方法用于获取所有 Kubernetes 上游的信息。

返回值

• upstreams：返回一个 list 类型的 Kubernetes 上游列表。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

upstreams = client.get_all_k8s_upstreams()

# new_rule

rule_id = new_rule(**kwargs)

此方法用于新建页面规则，包含各种规则类型，如：通用规则，代理规则，WAF 规则和缓存规则。你也可以指定规则执行的前提条件。

参数

condition 参数

conseq 参数

conseq 可以是 dict 类型或者 list 类型，如下是一个 dict 类型的例子。

waf 参数

proxy 参数

proxy 数组的描述如下：

upstreams 参数

upstreams 与 backup_upstreams 数组的描述如下：

cache 参数

返回值

• rule_id：新规则的 ID。

示例

示例一：使用 dict 类型的条件和动作来新建规则。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

condition = [
    {
        'var': 'host',
        'val': 'test.openresty.com'
    },
    {
        'var': ['req-header', 'Referer'],
        'vals': [
            'foo.com',
            ['foo\d+', 'rx'],
            ['foo*', 'wc']
        ]
    }
]

conseq = {
    'enable-websocket': {},
    'redirect': {'url': '/cn/2017/', 'code': 302}
}

rule_id = client.new_rule(condition=condition, conseq=conseq)

由于 dict 类型不能自行排序，并且不能设置重复名称的动作， 所以推荐下面这种使用 list 类型代替 dict 类型的方式。

示例二：使用 list 类型的条件和动作来新建规则。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

condition = [
    {'var': 'host', 'val': 'test.openresty.com'},
    {'var': ['req-header', 'Referer'], 'vals': [['foo\d+', 'rx'], 'foo.com']}
]

conseq = [
    {'set-proxy-header': {
        'header': 'Foo',
        'value': 'default'
    }},
    {'set-proxy-header': {
        'header': 'Host',
        'value': "orig"
    }},
    {'print': {
        'msg': 'hello'
    }}
]

rule_id = client.new_rule(condition=condition, conseq=conseq)

示例三：启用 WAF 规则。

创建自定义的 WAF 规则，并在应用中使用该规则。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

global_waf_rule_id = client.new_global_waf_rule(
    name='foo',
    code="uri-arg('foo') => waf-mark-risk(level: 'definite', msg: 'found foo');")

condition = [{'var': 'uri', 'op': 'prefix', 'val': '/foo'}]

waf_rule = {
    'rule_sets': [global_waf_rule_id],
    'action': '403 Forbidden', 'threshold': 'low'}

rule_id = client.new_rule(condition=condition, waf=waf_rule)

示例四：启用代理规则。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

up_id = client.new_upstream(
    name='origin-upstream',
    servers=[
        {'ip': '172.22.31.1', 'port': 80},
        {'ip': '172.22.31.2', 'port': 80, 'weight': 2}
    ])

k8s_up_id = client.new_k8s_upstream(
    name='k8s-upstream',
    k8s_services = [
        {
            'k8s' : k8s_id,
            'k8s_namespace' : 'default',
            'k8s_service' : 'test-hello',
            'k8s_service_port' : 80,
        }
    ])

ok = client.put_upstream(up_id,
    name='origin-upstream',
    servers=[
      {'ip': '172.22.31.1', 'port': 8080},
      {'ip': '172.22.31.2', 'port': 8080, 'weight': 1}
    ])

backup_up_id = client.new_upstream(
    name='backup-upstream',
    servers=[
        {'ip': '172.22.31.3', 'port': 80},
        {'ip': '172.22.31.4', 'port': 80, 'weight': 3}
    ])

proxy_rule = {
    'upstreams': [{'upstream': up_id, 'weight': 2}],
    'backup_upstreams': [{'upstream': backup_up_id}],
    'upstream_el_code': '',
    'retries': 2,
    "retry_condition": ["invalid_header", "http_500"],
    "balancer_algorithm": "hash",
    "balancer_vars": [
        {"name":"uri"},
        {
            "name":"uri-arg",
            "args":"foo"
        }
    ]
}

rule_id = client.new_rule(proxy = proxy_rule)

k8s_proxy_rule = {
    'upstreams': [{'k8s_upstream': k8s_up_id, 'weight': 2}],
    'upstream_el_code': '',
    'retries': 2,
    "retry_condition": ["invalid_header", "http_500"],
    "balancer_algorithm": "hash",
    "balancer_vars": [
        {"name":"uri"},
        {
            "name":"uri-arg",
            "args":"foo"
        }
    ]
}

k8s_rule_id = client.new_rule(proxy = k8s_proxy_rule)

示例五：启用缓存。

在该示例中，当访问来源的 IP 位于中国（CN）时，将会实施缓存规则。具体的缓存 key 包含 URL、查询字符串以及客户端来源的城市。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

condition = [{'var': 'client-country', 'op': 'eq', 'val': 'CN'}]
# cache_key = ['uri', 'query-string', 'client-city']
# or
cache_key = [
    {'name': 'uri'},
    {'name':'query-string'},
    {'name': 'client-city', 'args': 'first-x-forwarded-addr'}
]

default_ttls = [
    {
        'ttl_unit': "min", 'status': 200, 'ttl': 1
    }, {
        'ttl_unit': "min", 'status': 301, 'ttl': 1
    }
]
cache_rule = {'cache_key': cache_key, 'default_ttls': default_ttls}

rule_id = client.new_rule(condition=condition, cache=cache_rule)

# put_rule

ok = put_rule(**kwargs)

此方法用于修改页面规则。

参数

返回值

• True：表示成功。
• False：表示失败。

示例

from edge2client import Edge2Client
from edge2client.constants import OFF, ON

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

conseq = {
    'enable-websocket': {},
    'redirect': {'url': '/cn/2017/', 'code': 302},
    'user-code': {'el': 'true => say(\"hello\");'}
}

conseq['redirect']['url'] = '/cn/2018/'

ok = client.put_rule(rule_id=rule_id, condition=condition, conseq=conseq)

# 关闭页面规则中的条件
ok = client.put_rule(rule_id=rule_id, condition=OFF, conseq=conseq)

# get_rule

data = get_rule(rule_id)

此方法用于根据规则 ID 获取页面规则。

参数

返回值

• data：返回一个 dict 类型的页面规则信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

rule_id = 1
data = client.get_rule(rule_id)

# del_rule

success = del_rule(rule_id)

此方法用于删除与 rule_id 对应的页面规则。

参数

返回值

• True：表示页面规则成功删除。
• False：表示页面规则删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

rule_id = 1
success = client.del_rule(rule_id)

# get_all_rules

rules = get_all_rules(app_id)

此方法用于获取指定应用 ID (app_id) 的所有页面规则。

参数

返回值

• 返回一个规则列表，该列表中的每个元素的类型与 get_rule 函数的返回值类型一致。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

data = client.get_all_rules()

# reorder_rules

ok = reorder_rules(orders)

此方法用于调整规则的顺序。

参数

返回值

• True：如果成功调整规则顺序。
• False：如果调整规则顺序失败。

示例

以下示例展示了如何调换规则 ID 2 和 3 的顺序。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

new_orders = {
    1: 1,
    2: 3,
    3: 2,
}
data = client.reorder_rules(new_orders)

# get_all_rules_by_app_domain

rules = get_all_rules_by_app_domain(domain)

此方法用于根据指定的应用域名获取对应的页面规则。

参数

返回值

• rules：list 类型的规则列表。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

data = client.get_all_rules_by_app_domain('orig.foo.com')

# get_all_rules_by_upstream_ip

rules = get_all_rules_by_upstream_ip(upstream_ip)

此方法用于根据上游 IP 地址来检索所有关联的页面规则。

参数

返回值

• rules：这是一个包含所有相关规则的 list 类型列表。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

upstream_ip = '1.2.3.4'
data = client.get_all_rules_by_upstream_ip(upstream_ip)

for app_id, rules in data.items():
    print(app_id)

# get_all_waf_rules

rules = get_all_waf_rules(app_id)

此方法用于根据特定的应用 id (app_id) 获取对应的 WAF 规则。

参数

返回值

• rules：返回一种 list 类型的规则列表。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

data = client.get_all_waf_rules()

# get_global_actions_used_in_app

rule_ids = get_global_actions_used_in_app(app_id)

此方法用以获取与应用 ID (app_id) 相关联的自定义规则 ID。

参数

返回值

• rule_ids：返回一个列表，其中元素为各自规则的唯一 ID。如果需要获取规则的详细信息，可以依据这些规则 ID 进行进一步查询，详情请参阅 get_global_rule() 方法。

示例

以下示例展示了如何获取所有全局规则的 ID。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
rule_ids = client.get_global_actions_used_in_app(app_id)

# set_cert_key

cert_id = set_cert_key(**kwargs)

此方法用于新增 SSL 证书并提供手动上传证书的选项。

参数

返回值

• cert_id：返回新添加的证书的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

key_file = os.path.join('tests', 'key.pem')
cert_file = os.path.join('tests', 'cert.pem')

with open(key_file) as f:
    key = f.read()
with open(cert_file) as f:
    cert = f.read()

cert_id = client.set_cert_key(key=key, cert=cert)

# set_le_cert

cert_id = set_le_cert(**kwargs)

此方法用于添加 Let’s Encrypt 证书或其他证书提供商（如 ZeroSSL）的证书。

参数

返回值

• cert_id：新创建的 Let’s Encrypt 证书的 ID

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

cert_id = client.set_le_cert(domains=["orig.foo.com"])

# put_cert_key

ok = put_cert_key(**kwargs)

此方法用于更新 SSL 证书（手动上传方式）。

参数

返回值

• True：表示更新成功。
• False：表示更新失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

new_key_file = os.path.join('tests', 'new_key.pem')
new_cert_file = os.path.join('tests', 'new_cert.pem')
with open(new_key_file) as f:
    new_key = f.read()
with open(new_cert_file) as f:
    new_cert = f.read()

cert_id = 1
ok = client.put_cert_key(cert_id=cert_id, key=new_key, cert=new_cert)

# put_le_cert

ok = put_le_cert(**kwargs)

此方法用于更新 Let’s Encrypt 证书或其他证书提供商（如 ZeroSSL）提供的证书。

参数

返回值

• True：表示成功。
• False：表示失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

cert_id = 1
ok = client.put_le_cert(cert_id=cert_id, domains=["xx.foo.com"], resign=False)

# get_le_cert

data = get_le_cert(cert_id)

此方法用于获取 Let’s Encrypt 证书的签发状态。

参数

返回值

• data：返回一个 dict 类型的数据，包含以下字段：

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

cert_id = 1
data = client.get_le_cert(cert_id)

# get_cert_key

cert_data = get_cert_key(cert_id)

此方法用于获取 SSL 证书内容。

参数

返回值

• 返回值为 cert_data，表示证书的内容，数据类型为 dict。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

cert_id = 1
data = client.get_cert_key(cert_id)

# get_all_cert_keys

cert_data = get_all_cert_keys()

此方法用于获取当前应用的全部证书信息。

返回值

• cert_data：以 dict 类型返回的证书数据。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

data = client.get_all_cert_keys()

# del_cert_key

success = del_cert_key(certificate_id)

删除与给定的 certificate_id 相关联的证书信息。

参数

返回值

• True：删除成功。
• False：删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

application_id = 1
client.use_application(application_id)

certificate_id = 1
success = client.del_cert_key(certificate_id)

# new_el

ok = new_el(**kwargs)

此方法用于添加自定义的 edge 语言代码。

参数

返回值

• True：表示成功。
• False：表示失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

code = "true => print('hello, world');"

ok = client.new_el(phase='req-rewrite', code=code, pre=True)

# new_release

ok = new_release()

此方法用于发布当前应用的更新。

返回值

• True：表示更新成功。
• False：表示更新失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

ok = client.new_release()

此示例展示了如何使用 Edge2Client 的 new_release 方法来发布应用的更新。

# pending_changes

changes = pending_changes()

此方法用于查询待发布的更改数量。

返回值

• changes：这是一个整数，表示待发布的更改数量。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

changes = client.pending_changes()

在上述示例中，首先创建了一个 Edge2Client 实例并完成登录。随后，将应用的 id 设置为 1，并通过 pending_changes 函数查询待发布的更改数量。

# sync_status

total, synced = sync_status()

查询当前的同步状态。

返回值

• total：Edge Node 服务器的总数量。
• synced：已成功同步的 Edge Node 服务器的数量。

示例

创建多个规则，并确认它们是否已成功发布到全网。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

total, synced = client.sync_status()
if total == synced:
    print('新的发布已同步到所有节点')

# node_sync_status

data = node_sync_status()

此方法用于获取各节点的同步详情。

返回值

• data：dict 类型，它返回每个节点的 node_id 以及该节点上一次发布到现在的延迟时间。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.node_sync_status()
for node_id, delay in data.items():
    if delay > 0:
        print('警告：节点 id {}: 落后 {}'
        .format(str(node_id), str(delay)))

# get_healthcheck_status

data = get_healthcheck_status(node_id, page=1, page_size=1000)

这段代码用于获取各节点的健康检查信息。

返回值

• data：这是一个dict类型的数据，它返回每个节点的 node_id 以及对应节点的健康状态。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.node_sync_status()

for node_id, node_info in data.items():
    node_id = int(node_id)
    data = client.get_healthcheck_status(node_id, 1, 1000)

# add_app_user

user_id = add_app_user(name, read, write, release, dns_read, dns_write)

此方法用于新增应用级用户。

参数

返回值

• user_id：新添加用户的 ID。

请注意：此处的 user_id 与全局用户的 user_id 不同。

示例

以下示例展示了如何新增一个应用级用户，用户名为 only_reader，并且没有写权限和发布更动权限。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

user_id = client.add_app_user(name='only_reader', write=False, release=False)

# update_app_user

success = update_app_user(id, name, read, write, release, dns_read, dns_write)

此方法用于更新应用级的用户信息。

参数

返回值

• True：用户信息更新成功。
• False：用户信息更新失败。

示例

以下示例展示了如何使用此方法更新 id 为 1 的用户的权限，将用户名设置为 master，并赋予写入和发布更改的权限。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

uid = 1
success = client.update_app_user(id=uid, name='master', write=True, release=True)

# get_app_user

data = get_app_user(id=None, name=None, app_id=None, user_id=None)

此方法用于获取应用级别的用户信息，包括用户名、权限等。

参数

返回值

• data：返回一个字典类型的数据，包含用户名、权限等用户信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

username = 'only_reader'
data = client.get_app_user(name=username)

# del_app_user

ok = del_app_user(id, name, app_id, user_id)

此方法用于删除特定应用的用户。

参数

返回值

• 如果操作成功，则返回 True。
• 如果操作失败，则返回 False。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

username = 'only_reader'
ok = client.del_app_user(name=username)

在这个例子中，我们首先创建了一个 Edge2Client 实例并登录。然后，我们选择了一个应用，最后我们通过传入用户名来删除应用的用户。

# get_all_app_users

data = get_all_app_users(app_id)

此方法用于获取所有当前应用级别的用户，也可以通过指定 app_id 来获取特定应用的用户。

参数

返回值

• data：返回值为list类型，列表中的每个元素都是 dict 类型。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

data = client.get_all_app_users()

# add_user_for_all_apps

ok = add_user_for_all_apps(name, read, write, release, dns_read, dns_write)

此方法用于为所有应用添加用户。

参数

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

# 利用 add_global_user() 函数创建名为 'write' 的用户。
username='write'

data = client.add_user_for_all_apps(name=username)

# add_all_users_for_app

ok = add_all_users_for_app(app_id, read, write, release, dns_read, dns_write)

此方法用于将所有全局用户添加到指定应用。

参数

返回值

• True：表示添加成功。
• False：表示添加失败。

示例

以下示例展示了如何为 ID 为 1 的应用添加用户，其中所有全局用户都将添加在该应用下，且 release (发布更动) 权限为 True，其他权限保持默认值。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

data = client.add_all_users_for_app(release=True)

# new_cache_purge_task

task_id = client.new_cache_purge_task(condition=condition)

此方法用于新增条件刷新任务。任务的触发条件以数组形式传递，支持设定多个触发条件。只有当所有设定的条件同时满足时，才会触发对应的刷新任务。

参数

返回值

• task_id：返回已创建的刷新任务的 ID，可通过此 ID 查询刷新任务的执行详情。

示例

以下示例中，新建的条件刷新任务的触发条件设定为：“当访问的 Host 为 con.foo.com，并且请求头中 Referer 的值匹配到正则表达式 foo\d+ 时。”

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

condition = [
    {'var': 'host', 'val': 'con.foo.com'},
    {'var': ['req-header', 'Referer'], 'vals': [['foo\d+', 'rx'], 'foo.com']}
]

rule_id = client.new_cache_purge_task(condition=condition)

# get_cache_purge_task

data = client.get_cache_purge_task(task_id)

此方法通过指定的刷新任务 id 来查询任务的详细信息。

参数

返回值

• data：返回的是一个包含任务详情的字典（dict）类型的数据。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

task_id = 1

data = client.get_cache_purge_task(task_id)

# get_all_cache_purge_tasks

data = client.get_all_cache_purge_tasks()

此方法用于获取所有缓存刷新任务的详细信息。

返回值

• data：返回一个包含所有刷新任务的列表。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

data = client.get_all_cache_purge_tasks()

# del_cache_purge_task

ok = client.del_cache_purge_task(task_id)

此方法用于删除指定 ID 的刷新任务。

参数

返回值

• True：表示成功删除。
• False：表示删除失败。

示例

以下示例展示如何删除 ID 为 1 的刷新缓存任务。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

task_id = 1
ok = client.del_cache_purge_task(task_id)

# new_waf_whitelist

rule_id = new_waf_whitelist(**kwargs)

此方法用于新增 WAF 白名单规则。

参数

WAF Condition 参数

返回值

• rule_id：新创建的 WAF 白名单规则的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

condition = [
    {'var': 'host', 'val': 'con.foo.com'},
    {'var': ['req-header', 'Referer'], 'vals': [['foo\d+', 'rx'], 'foo.com']}
]

id = client.new_waf_whitelist(condition=condition, rule_sets=[1, 2])

# update_waf_whitelist

ok = update_waf_whitelist(**kwargs)

此方法用于更新 WAF 白名单规则。

参数

返回值

• 若成功，返回 True。
• 若失败，返回 False。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

condition = [
    {'var': 'host', 'val': 'con.foo.com'},
    {'var': ['req-header', 'Referer'], 'vals': [['foo\d+', 'rx'], 'foo.com']}
]

waf_whitelist_id = 1
ok = client.update_waf_whitelist(
    whitelist_id=waf_whitelist_id, condition=condition, rule_sets=[1])

# get_waf_whitelist

data = get_waf_whitelist(whitelist_id)

此方法用于获取 WAF 的白名单规则。

参数

返回值

• data：返回一个 dict 类型，包含 WAF 白名单的规则数据。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

data = client.get_waf_whitelist(waf_whitelist_id)

# del_waf_whitelist

ok = del_waf_whitelist(whitelist_id)

执行此操作会删除指定的 WAF 白名单规则。

参数

返回值

• True：表示删除成功。
• False：表示删除失败。

示例

from edge2client import Edge2Client

# 初始化客户端并登录
client = Edge2Client(host, username, password)
client.login()

# 指定 APP ID
app_id = 1
client.use_app(app_id)

# 删除特定 WAF 白名单规则
ok = client.del_waf_whitelist(waf_whitelist_id)

# get_all_waf_whitelists

data = get_all_waf_whitelists()

此方法用于获取所有的 WAF 白名单规则。

返回值

• data：返回一个 list 类型的 WAF 白名单数据。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

# 选择应用
app_id = 1
client.use_app(app_id)

# 获取所有的 WAF 白名单
data = client.get_all_waf_whitelists()

# new_app_dymetrics

id = new_app_dymetrics(app_id=None, name=None, note=None, interval=60, sql=None)

创建新的应用动态指标。如果想要开启 OpenResty Edge 的内置动态指标，可以从此文档获取内置动态指标的名称及 SQL。

参数

返回值

• id：新创建的动态指标的唯一标识符。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

sql = 'select status, count(*) from reqs group by status'

dymetrics_id = client.new_app_dymetrics(app_id=1, name='test', sql=sql)

# put_app_dymetrics

ok = put_app_dymetrics(app_id=None, id=None, name=None, note=None, interval=60, sql=None)

此方法用于修改应用的动态指标。

参数

返回值

• True：表示修改成功。
• False：表示修改失败。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

id=1
sql = 'select status, count(*) from reqs group by status'
data = client.put_app_dymetrics(app_id=1, id=id, name='test', sql=sql)

# del_app_dymetrics

ok = del_app_dymetrics(app_id, id)

此方法用于删除指定应用的动态指标。

参数

返回值

• True：表示删除成功。
• False：表示删除失败。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

ok = client.del_app_dymetrics(app_id, dymetrics_id)

# get_app_dymetrics

data = get_app_dymetrics(app_id, id)

此方法用于获取指定应用的动态指标。

参数

返回值

• data：返回一个 dict 类型的动态指标数据。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

data = client.get_app_dymetrics(app_id, dymetrics_id)

# get_all_app_dymetrics

data = get_all_app_dymetrics(app_id)

此方法用于获取指定应用的全部动态指标信息。

参数

返回值

• data：返回的数据为一个包含所有动态指标信息的 list 类型列表。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

data = client.get_all_app_dymetrics(app_id)

# get_app_dymetrics_data

data = get_app_dymetrics_data(app_id, id, chart_type='line', start_time=None, end_time=None, node_id=None)

此方法用于获取应用的动态指标数据，默认返回最近 30 分钟的数据。

参数

返回数据

• data：返回的动态指标数据，其数据类型为list。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

app_id = 1
dymetrics_id = 1
data = client.get_app_dymetrics_data(app_id, dymetrics_id)

# get_app_metrics

data = get_app_metrics(id, start_time=None, end_time=None)

此方法用于获取应用的 HTTP 请求状态指标和网络收发指标，默认情况下，它会获取最近 30 分钟的数据。

参数

返回值

• data：这是一个包含指标数据的 list 类型。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

data = client.get_app_metrics(app_id)

# new_ip_list

id = new_ip_list(name=None, type='ipv4', items=None)

添加新的应用级 IP 列表。

参数

items 参数

返回值

• id：新创建的 IP 列表的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

ok = client.new_ip_list(name = 'ip-list-1',
                        type = 'ipv4',
                        items = [
                            {'ip': '127.0.0.0/24'},
                            {'ip': '192.168.1.0/24'},
                            {'ip': '192.168.2.0/24', 'exptime': 1725000000},
                            {'ip': '192.168.3.0/24', 'exptime': 1735000000}
                        ])

# put_ip_list

ok = put_ip_list(rule_id=None, name=None, type='ipv4', items=None)

此方法用于修改指定的 IP 列表。

参数

返回值

• True：表示成功修改指定的 IP 列表。
• False：表示修改指定的 IP 列表失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

ok = client.put_ip_list(rule_id = 1, items = [ {'ip': '127.0.0.2'} ])

# del_ip_list

ok = del_ip_list(rule_id=None)

此方法用于删除指定的 IP 列表。

参数

返回值

• True：表示删除成功。
• False：表示删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

ok = client.del_ip_list(rule_id=1)

在此示例中，我们首先使用 Edge2Client 的实例登陆，然后选择应用程序，最后删除指定 ID 的 IP 列表。

# get_ip_list

data = get_ip_list(rule_id=None)

此方法用于查看特定的 IP 列表。

参数

返回值

• data：返回一个字典类型的 IP 列表数据。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

data = client.get_ip_list(rule_id=1)

# append_to_ip_list

ok = append_to_ip_list(rule_id, items)

此方法用于向指定的 IP 列表追加 IP。

参数

返回值

• True：表示成功修改指定的 IP 列表。
• False：表示修改指定的 IP 列表失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

ok = client.append_to_ip_list(rule_id = 1, items = [ {'ip': '127.0.0.3'} ])

ok = client.append_to_ip_list(rule_id = 1, items = [ {'ip': '127.0.0.3'} ], remove_expired = True)

# remove_from_ip_list

ok = remove_from_ip_list(rule_id, items)

此方法用于从指定的 IP 列表移除 IP。

参数

返回值

• True：表示成功修改指定的 IP 列表。
• False：表示修改指定的 IP 列表失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

app_id = 1
client.use_app(app_id)

ok = client.remove_from_ip_list(rule_id = 1, items = [ {'ip': '127.0.0.3'} ])

# node_monitor

data = node_monitor(node_id, start_time=None, end_time=None, step=60)

此方法用于获取网络节点的 CPU 和内存负载相关信息。

参数

返回值

• data：返回一个包含监控数据的列表。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

node_id = 1
data = client.node_monitor(node_id)

# new_global_dymetrics

id = new_global_dymetrics(name=None, note=None, interval=60, sql=None)

此方法用于创建全局动态指标。

参数

返回值

• id：新创建的全局动态指标的唯一标识符 ID。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

sql = 'select status, count(*) from reqs group by status'

dymetrics_id = client.new_global_dymetrics(name='test', sql=sql)

# put_global_dymetrics

ok = put_global_dymetrics(id=None, name=None, note=None, interval=60, sql=None)

此方法用于修改全局动态指标。

参数

返回值

• True：表示成功。
• False：表示失败。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

id=1
sql = 'select status, count(*) from reqs group by status'
data = client.put_global_dymetrics(id=id, name='test', sql=sql)

# del_global_dymetrics

ok = del_global_dymetrics(id)

此方法用于删除指定的全局动态指标。

参数

返回值

• True：表示成功删除全局动态指标。
• False：表示删除全局动态指标失败。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

ok = client.del_global_dymetrics(dymetrics_id)

# get_global_dymetrics

data = get_global_dymetrics(id)

此方法用于获取指定的全局动态指标。

参数

返回值

• data：返回的是一个类型为 dict 的全局动态指标信息。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

data = client.get_global_dymetrics(dymetrics_id)

# get_all_global_dymetrics

data = get_all_global_dymetrics()

此方法用于获取所有全局动态指标。

返回值

• data：返回一个包含全局动态指标信息的 list 类型数据。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

data = client.get_all_global_dymetrics()

# get_global_dymetrics_data

data = get_global_dymetrics_data(id, chart_type='line', start_time=None, end_time=None, node_id=None)

此方法用于获取全局动态指标数据，默认范围为最近的 30 分钟。

参数

返回值

• data：返回值是一个列表，包含全局动态指标数据。

示例

from edge2client import Edge2Client

# 创建并登录客户端
client = Edge2Client(host, username, password)
client.login()

dymetrics_id = 1
data = client.get_global_dymetrics_data(dymetrics_id)

# use_dns_app

ok = use_dns_app(dns_id)

此方法用于切换当前实例所操作的 DNS 应用。

参数

返回值

• True：表示切换成功。
• False：表示切换失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

dns_id = 1
client.use_dns_app(dns_id)

# new_dns_app

dns_id = new_dns_app(**kwargs)

此方法用于创建新的 DNS 应用。

参数

返回内容

• dns_id：返回新创建的 DNS 应用的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

authority = [
    {'domain': 'ns1.foo.com', 'ttl': '2 [hour]'},
    {'domain': 'ns2.foo.com', 'ttl': '1 [day]'}
]

soa_email = 'admin@foo.com'

dns_id = client.new_dns_app(
    authority=authority,
    soa_email=soa_email,
    zone='foo.com')

# put_dns_app

ok = put_dns_app(**kwargs)

此方法用于修改 DNS 应用的相关信息。

参数

返回值

• True：表示信息修改成功。
• False：表示信息修改失败。

示例

以下示例展示如何修改 dns_id 为 1 的应用。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

dns_id = 1

authority = [
    {'domain': 'ns1.foo.com', 'ttl': '4 [hour]'},
    {'domain': 'ns2.foo.com', 'ttl': '1 [day]'}
]

ok = client.put_dns_app(
    dns_id=dns_id,
    authority=authority,
    soa_email=soa_email,
    zone='foo.com')

# get_dns_app

data = get_dns_app(dns_id)

此方法用于获取 DNS 应用的相关信息。

参数

返回值描述

• data：一个 dict 类型的数据，包含 DNS 应用的详细信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

dns_id = 1

data = client.get_dns_app(dns_id)

# del_dns_app

ok = del_dns_app(dns_id)

此方法用于删除指定的 DNS 应用。

参数

下表列出了此方法所需的参数。

返回值

• True：表示成功删除了 DNS 应用。
• False：表示删除 DNS 应用失败。

示例

以下示例展示了如何删除指定的 DNS 应用。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

dns_id = 1

data = client.del_dns_app(dns_id)

# new_dns_record

record_id = new_dns_record(**kwargs)

此方法用于新增 DNS 记录。

参数

返回值

• record_id：新增的 DNS 记录的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

dns_id = 1
use_dns_app(dns_id)

record_id = client.new_dns_record(
    sub_domain='bar', record_type='TXT', text='text string')
client.new_dns_record(
    sub_domain='bar', record_type='A', ip='127.0.0.1')
client.new_dns_record(
    sub_domain='bar', record_type='CNAME', domain='test.com')
client.new_dns_record(
    sub_domain='bar', record_type='MX', domain='test.com', priority=3)

# put_dns_record

ok = put_dns_record(**kwargs)

此方法用于修改 DNS 记录。

参数

返回值

• True：表示成功。
• False：表示失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

dns_id = 1
use_dns_app(dns_id)

record_id = 1

ok = client.put_dns_record(
    record_id=record_id,
    sub_domain='a.bar',
    record_type='TXT',
    text='text string')

# get_dns_record

data = get_dns_record(record_id)

此方法用于获取 DNS 记录。

参数

返回值

• data：返回一个 dict 类型的 DNS 记录信息。

示例

如需获取 record_id 为 1 的 DNS 记录信息，可以参考以下步骤。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

dns_id = 1
use_dns_app(dns_id)

record_id = 1

data = client.get_dns_record(record_id)

# del_dns_record

ok = del_dns_record(record_id)

此方法用于删除 DNS 记录。

参数

返回值

• 若成功删除，则返回 True。
• 若删除失败，则返回 False。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

dns_id = 1
use_dns_app(dns_id)

record_id = 1

ok = client.del_dns_record(record_id)

在该示例中，我们首先创建了一个 Edge2Client 客户端，并用特定的主机，用户名和密码进行登录。然后，我们选择使用具有特定 id 的 DNS 应用。最后，我们尝试使用该客户端删除具有特定 id 的 DNS 记录，并将结果保存在 ok 中。

# new_global_upstream

up_id = new_global_upstream(**kwargs)

此方法用于新增上游，并返回新创建的全局上游的 id。

参数

返回值

• up_id：新建上游的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

up_id = client.new_global_upstream(
    name='origin-upstream',
    servers=[
        {'ip': '172.22.31.1', 'port': 80},
        {'ip': '172.22.31.2', 'port': 80, 'weight': 2}
    ])

# update_global_upstream

ok = update_global_upstream(**kwargs)

此方法用于更新全局上游信息。

参数

返回值

• True：操作成功。
• False：操作失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

up_id = 1

ok = client.update_global_upstream(up_id,
    name='origin-upstream',
    servers=[
        {'domain': 'test.com', 'port': 8080},
        {'ip': '172.22.31.2', 'port': 8080, 'weight': 1}
    ],
    health_checker={
        'type': 'http',
        'http_req_uri': '/status',
        'http_req_host': 'test.com',
        'interval': 3,
        'interval_unit': 'sec',
        'timeout': 1,
        'fall': 3,
        'rise': 2,
        'valid_statuses': [200, 302],
        'report_interval': 3,
        'report_interval_unit': 'min'
    })

# get_global_upstream

data = get_global_upstream(up_id)

此方法用于获取全局上游信息。

参数

返回值

• True：当获取信息成功时。
• False：当获取信息失败时。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

up_id = 1

data = client.get_global_upstream(up_id)

# del_global_upstream

del_global_upstream(up_id)

此方法用于删除全局上游。

参数

返回值

• True：表示删除成功。
• False：表示删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

up_id = 1

operation_status = client.del_global_upstream(up_id)

# get_all_global_upstreams

upstreams = client.get_all_global_upstreams()

此方法用于获取全部的上游信息。

返回值

• upstreams：返回一个包含所有全局上游信息的 list 类型变量。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

upstreams = client.get_all_global_upstreams()

# new_global_k8s_upstream

up_id = new_global_k8s_upstream(**kwargs)

此方法用于创建新的全局 Kubernetes 上游，并返回新创建的上游的 ID。

参数

返回值

• up_id: 新创建的 k8s 上游的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

k8s_up_id = client.new_global_k8s_upstream(
    name = 'k8s-upstream',
    k8s_services = [
        {
            'k8s' : 1,
            'k8s_namespace' : 'default',
            'k8s_service' : 'foo',
            'k8s_service_port' : 80,
        }
    ])

# put_global_k8s_upstream

ok = put_global_k8s_upstream(**kwargs)

此方法用于修改全局 Kubernetes 上游的信息。

参数

返回值

• True：表示修改成功。
• False：表示修改失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

up_id = 1

ok = client.put_global_k8s_upstream(up_id,
    name='k8s-upstream',
    k8s_services = [
        {
            'k8s' : 1,
            'k8s_namespace' : 'default',
            'k8s_service' : 'foo',
            'k8s_service_port' : 80,
        }
    ],
    health_checker={
        'type': 'http',
        'http_req_uri': '/status',
        'http_req_host': 'test.com',
        'interval': 3,
        'interval_unit': 'sec',
        'timeout': 1,
        'fall': 3,
        'rise': 2,
        'valid_statuses': [200, 302],
        'report_interval': 3,
        'report_interval_unit': 'min'
    })

# fetch_global_k8s_upstream

data = fetch_global_k8s_upstream(up_id)

此方法用于获取全局 Kubernetes 上游的信息。

参数

返回值

• data：返回一个包含 Kubernetes 上游信息的字典类型值。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

up_id = '1'

data = client.fetch_global_k8s_upstream(up_id)

# del_global_k8s_upstream

del_global_upstream(up_id)

此方法用于删除全局 k8s 上游。

参数

返回值

• 如果删除成功，返回 True。
• 如果删除失败，返回 False。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

up_id = '1'

ok = client.del_global_k8s_upstream(up_id)

# get_all_global_k8s_upstreams

upstreams = client.get_all_global_k8s_upstreams()

此方法用于获取所有的 Kubernetes 上游信息。

返回值

• upstreams：返回 Kubernetes 上游信息的列表。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

upstreams = client.get_all_global_k8s_upstreams()

# set_global_cert_key

cert_id = set_global_cert_key(**kwargs)

用于添加全局 SSL 证书。

参数

返回值

• cert_id：新建全局证书的 ID。

示例

以下示例展示了如何上传本地的 key.pem 和 cert.pem 文件以创建全局证书。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

key_file = os.path.join('tests', 'key.pem')
cert_file = os.path.join('tests', 'cert.pem')
with open(key_file) as f:
    key = f.read()
with open(cert_file) as f:
    cert = f.read()

cert_id = client.set_global_cert_key(key=key, cert=cert)

# put_global_cert_key

ok = put_global_cert_key(**kwargs)

此方法用于修改全局 SSL 证书。

参数

返回值

• True：表示成功。
• False：表示失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

cert_id = 1

new_key_file = os.path.join('tests', 'new_key.pem')
new_cert_file = os.path.join('tests', 'new_cert.pem')
with open(new_key_file) as f:
    new_key = f.read()
with open(new_cert_file) as f:
    new_cert = f.read()

ok = client.put_global_cert_key(cert_id=cert_id, key=new_key, cert=new_cert)

# get_global_cert_key

data = get_global_cert_key(cert_id)

此方法用于获取与 cert_id 相关联的 SSL 证书信息。

参数

返回值

• data：返回一个 dict 类型数据，其中包含了全局证书的所有信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

cert_id = 1

data = client.get_global_cert_key(cert_id)

# del_global_cert_key

ok = del_global_cert_key(cert_id)

此方法用于删除指定的全局 SSL 证书。

参数

返回值

• True：表示删除成功。
• False：表示删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

cert_id = 1

ok = client.del_global_cert_key(cert_id)

# add_global_user

uid = add_global_user(name, pwd, gid)

此方法用于新增全局用户。

参数

返回值

• uid：新增的全局用户的唯一识别 ID。

注意：全局用户的 ID 与应用级别的用户 ID 需要区分。

示例

以下示例展示了如何创建一个名为 global_demo_user，具有创建新应用权限，且其角色为普通用户的全局用户。

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

username='global_demo_user'

uid = client.add_global_user(name=username, pwd=pwd, gid=[1])

# search_global_user

user_info = search_global_user(name)

此方法用于根据用户名搜索全局用户信息。

参数

返回值

• user_info：dict 类型，包含搜索到的用户信息及其相关权限信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

username = 'global_demo_user'

data = client.search_global_user(username)
user_id = data.get('id')

# get_all_global_users

data = get_all_global_users(detail)

本函数用于获取所有全局用户的信息。

参数

返回值

• data：返回一个 list 类型的数据，其每一个元素均为 dict 类型，包含对应全局用户的信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.get_all_global_users()
data = client.get_all_global_users(True)

# new_global_rule

rule_id = new_global_rule(**kwargs)

此方法用于新增全局规则。

参数

返回值

• rule_id：成功创建的全局规则的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

condition = [
    {'var': 'host', 'val': 'con.' + APEX},
    {'var': ['req-header', 'Referer'], 'vals': [['foo\d+', 'rx'], 'foo.com']}
]

conseq = {
    'enable-websocket': {},
    'redirect': {'url': '/cn/2017/', 'code': 302}
}

rule_id = client.new_global_rule(condition=condition, conseq=conseq,gid=[1])

# put_global_rule

ok = put_global_rule(**kwargs)

此方法用于修改全局规则。

参数

返回值

• True：表示成功。
• False：表示失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

rule_id = 1

conseq['redirect']['url'] = '/cn/2018/'

ok = client.put_global_rule(
    rule_id=rule_id, condition=condition, conseq=conseq)

# get_global_rule

data = get_global_rule(rule_id)

此方法用于获取全局规则信息。

参数

返回值

• data：返回值是一个dict类型，包含全局规则的详细信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

rule_id = 1

data = client.get_global_rule(rule_id)

这个示例展示了如何使用 Edge2Client 的实例获取特定 id 的全局规则信息。

# get_global_action_by_name

rule_id = get_global_action_by_name(name)

此方法用于根据给定的应用名来获取相应的全局自定义动作。

参数

返回值

• data：返回一个包含全局规则信息的 dict 类型数据。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

rule_name = 'test-actions'

rule_id = client.get_global_action_by_name(rule_name)

# get_all_global_rules

data = get_all_global_rules()

此方法用于获取所有的全局规则。

返回值

• data：这是一个 list 类型的全局规则信息列表。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.get_all_global_rules()

# del_global_rule

ok = del_global_rule(rule_id)

此方法用于删除全局规则。

参数

返回值

• 返回 True：表示规则成功删除。
• 返回 False：表示规则删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

rule_id = 1

ok = client.del_global_rule(rule_id)

此示例展示了如何使用 Edge2Client 对象的 del_global_rule 方法来删除指定的全局规则。

# new_global_var

var_id = new_global_var(**kwargs)

此方法用于新增全局变量。

参数

返回值

• var_id：若创建成功，则返回新建变量的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

var_id = client.new_global_var(
    name='is-whitelist', var_type='string', default='no')

# put_global_var

ok = put_global_var(**kwargs)

此方法用于修改全局变量。

返回值

• True：表示全局变量修改成功。
• False：表示全局变量修改失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

var_id = 1

ok = client.put_global_var(
    var_id=var_id,
    name='is-whitelist',
    var_type='string',
    default='yes',
    gid=[1])

在此示例中，我们首先创建了一个 Edge2Client 对象并进行登录。然后，使用 put_global_var() 函数，通过指定对应的参数，修改 ID 为 1 的全局变量名称为’is-whitelist’，类型为’string’，默认值为’yes’，并将其关联到 ID 为 1 的用户组。

# get_global_var

data = get_global_var(var_id)

此方法用于获取全局变量。

参数

返回值

• data：返回值为一个 dict 类型，包含全局变量的信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

var_id = 1

data = client.get_global_var(var_id)

# get_all_global_vars

data = get_all_global_vars()

此方法用于获取所有全局变量。

返回值

• data：此为一个 list 类型的对象，其中每一个元素都是一个 dict 类型，包含了全局变量的相关信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password) # 创建一个 Edge2Client 实例
client.login() # 登陆

data = client.get_all_global_vars() # 获取所有全局变量

# del_global_var

ok = del_global_var(var_id)

此方法用于根据全局变量的 ID 进行删除操作。

参数

| 名称 | 数据类型 | 是否必须 | 描述信息 | | var_id | number | 是 | 指定需要删除的全局变量的 ID |

返回值

• True：表示删除成功。
• False：表示删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

var_id = 1

ok = client.del_global_var(var_id)

# get_global_ngx_config

data = get_global_ngx_config()

此方法用于获取全局的 nginx 配置信息。

返回值

• data：返回一个dict类型的全局 ngx 配置信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.get_global_ngx_config()

# set_global_ngx_config

data = set_global_ngx_config(opts)

此方法设定全局的 nginx 配置信息。

参数

access_log_formats 参数

返回值

• True：设定成功。
• False：设定失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

opts = {'keepalive_timeout': 10, 'enable_open_file_cache': False}
ok = client.set_global_ngx_config(opts)

# get_global_misc_config

data = get_global_misc_config()

此方法用于获取全局的杂项配置信息。

返回值

• data：返回一个 dict 类型，其中包含了杂项的配置信息。

示例

from edge2client import Edge2Client

# 创建客户端并登陆
client = Edge2Client(host, username, password)
client.login()

# 获取全局杂项配置信息
data = client.get_global_misc_config()

# set_global_misc_config

ok = set_global_misc_config(opts)

此方法用于设定全局杂项配置信息。

参数

| 参数名 | 数据类型 | 是否必须 | 描述 | | opts | 字典类型（dict） | 否 | 杂项（misc）配置项 |

返回值

• True：表示设定成功。
• False：表示设定失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

opts = {'enabled_req_id': True}
ok = client.set_global_misc_config(opts)

# get_request_id_status

status = get_request_id_status()

此方法用于查询请求 id 的状态。

返回值

• True：表示请求 id 已启用。
• False：表示请求 id 已禁用。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

status = client.get_request_id_status()

# enable_request_id

ok = enable_request_id()

这个方法用于激活请求 id 功能。

返回值

• True：表示激活成功。
• False：表示激活失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

ok = client.enable_request_id()

# disable_request_id

ok = disable_request_id()

此方法用于禁用请求 ID 功能。

返回值

• True：表示请求 ID 功能成功被禁用。
• False：表示请求 ID 功能禁用失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

ok = client.disable_request_id()

# new_global_waf_rule

rule_id = new_global_waf_rule(**kwargs)

此方法用于新增全局 WAF 规则。

参数

返回值

• rule_id：返回新创建的全局 WAF 规则集的唯一 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

waf_rule_id = client.new_global_waf_rule(
    name='foo',
    code="uri-arg('foo') => waf-mark-risk(level: 'definite', msg: 'found foo');")

在这个例子中，我们首先创建了一个 Edge2Client 实例，然后使用 new_global_waf_rule 方法创建了一个新的全局 WAF 规则，并将其规则 ID 存储在 waf_rule_id 中。规则名称为 ‘foo’，edge 语言代码为 “uri-arg(‘foo’) => waf-mark-risk(level: ‘definite’, msg: ‘found foo’);"。

# put_global_waf_rule

ok = put_global_waf_rule(**kwargs)

此方法用于修改全局 WAF 规则。

参数

返回值

• True：表示修改成功。
• False：表示修改失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

waf_rule_id = 1

ok = client.put_global_waf_rule(
    rule_id=waf_rule_id,
    name='bar',
    code="uri-arg('bar') => waf-mark-risk(level: 'definite', msg: 'found bar');")

在此示例中，我们通过 put_global_waf_rule 函数修改了 id 为 1 的全局 WAF 规则。

# get_global_waf_rule

data = get_global_waf_rule(rule_id)

此方法通过提供的规则集 ID（rule_id）来获取全局的 WAF 规则集。

参数

| 名称 | 数据类型 | 是否必选 | 描述信息 | | rule_id | number | 是 | WAF 规则集的唯一标识符 |

返回值

• data：返回一个包含 WAF 规则集详细信息的字典。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

waf_rule_id = 1

data = client.get_global_waf_rule(waf_rule_id)

# del_global_waf_rule

ok = del_global_waf_rule(rule_id)

此方法用于删除全局 WAF 规则。

参数

| 名称 | 数据类型 | 是否必选 | 描述 | | rule_id | number | 是 | 规则集的 ID |

返回值

• True：删除成功。
• False：删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

waf_rule_id = 1

ok = client.del_global_waf_rule(waf_rule_id)

# get_all_global_waf_rules

data = get_all_global_waf_rules(detail=False)

此方法用于获取所有全局 WAF 规则。

参数

返回值

• data：返回一个 list 类型的 WAF 规则集信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.get_all_global_waf_rules()

# new_global_action

rule_id = new_global_action(**kwargs)

此方法用于添加新的全局自定义动作。

参数

返回值

• rule_id：返回新创建的全局自定义动作的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

name = 'test'
condition = [
    {'var': 'req-method', 'vals': ['GET', 'POST', 'HEAD']}
]

conseq = [
    {
        'exit': {'code': 403}
    }
]

action_id = client.new_global_action(
    name=name, condition=condition, conseq=conseq, gid=[1, 2])

# put_global_action

ok = put_global_action(**kwargs)

此方法用于修改全局自定义动作。

参数

返回值

• True：代表操作成功。
• False：代表操作失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

action_id = 1

ok = client.put_global_action(
    name=name,
    action_id=action_id,
    condition=condition,
    conseq=conseq,
    gid=[1])

# get_global_action

data = get_global_action(action_id)

此方法用于获取全局自定义动作。

参数

返回值

• data：返回的数据为字典类型，包含全局自定义动作的相关信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

action_id = 1

data = client.get_global_action(action_id)

# del_global_action

ok = del_global_action(action_id)

此方法用于删除指定的全局自定义动作。

参数

返回值

• 返回 True：表示删除成功。
• 返回 False：表示删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

action_id = 1

ok = client.del_global_action(action_id)

# count_global_actions

count = count_global_actions()

此方法用于获取全局自定义动作的总数。

返回值

• count：返回全局自定义动作的总数。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

count = client.count_global_actions()

# get_all_global_actions

data = get_all_global_actions()

此方法用于获取所有全局自定义动作。

参数

返回值

• data：返回一个包含所有全局自定义动作信息的列表。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.get_all_global_actions()

# upload_static_file

file_id = upload_static_file(**kwargs)

此方法用于上传静态文件。

参数

返回值

• file_id：返回新上传的静态文件的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

file_id = client.upload_static_file(content='test_content', label='500.html', gid=[1])

# get_static_file

data = get_static_file(file_id)

此方法用于获取指定的静态文件配置。

参数

返回值

• data：返回一个 dict 类型的对象，包含所获取的静态文件的信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

file_id = 1

data = client.get_static_file(file_id)

在这个示例中，我们首先从 edge2client 模块导入了 Edge2Client 类，然后创建了一个 Edge2Client 对象并进行登录。之后，我们调用了其 get_static_file 方法，获取了 ID 为 1 的静态文件的配置信息，并将其存储在 data 变量中。

# del_static_file

ok = del_static_file(file_id)

此方法用于删除已上传的静态文件。

参数

返回值

• True：表示文件删除成功。
• False：表示文件删除失败。

示例

from edge2client import Edge2Client

# 创建 Edge2Client 实例
client = Edge2Client(host, username, password)

# 登录
client.login()

# 设定要删除的文件 ID
file_id = 1

# 删除指定 ID 的静态文件
ok = client.del_static_file(file_id)

# new_cluster_group

group_id = new_cluster_group(group_name)

此方法用于添加新的集群组。

参数

返回值

• id：返回新建的集群组的唯一标识符 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

group_id = client.new_cluster_group('foo')

# put_cluster_group

ok = put_cluster_group(**kwargs)

此方法用于修改集群组信息。

参数

返回值

• True：表示成功修改集群组信息。
• False：表示修改集群组信息失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

group_id = 1

ok = client.put_cluster_group(group_id=group_id, group_name='bar')

此示例演示了如何使用 put_cluster_group 函数来修改具有指定 group_id 的集群组名称为 ‘bar’。

# get_cluster_group

data = get_cluster_group(group_id)

此方法用于获取集群组的详细信息。

参数

返回值

• data：返回一个dict类型的集群组详细信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

group_id = 1

data = client.get_cluster_group(group_id)

# get_all_cluster_groups

data = get_all_cluster_groups()

此方法用于获取所有集群组的信息。

返回值

• data：这是一个 list 类型，包含了所有集群组的信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.get_all_cluster_groups()

# del_cluster_group

ok = del_cluster_group(group_id)

此方法用于删除集群组信息。

参数

返回值

• 返回 True：表示集群组信息删除成功。
• 返回 False：表示集群组信息删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

group_id = 1

ok = client.del_cluster_group(group_id)

在以上示例中，首先初始化一个 Edge2Client 对象并登陆，然后指定要删除的集群组的编号，并调用 del_cluster_group 方法进行删除操作。

# put_proxy_rule

ok = put_proxy_rule(**kwargs)

此方法用于修改上游代理规则。

参数

返回值

• True：表示规则修改成功。
• False：表示规则修改失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

rule_id = 1
proxy_rule = []

ok = client.put_proxy_rule(rule_id=rule_id, proxy=proxy_rule)

# new_global_k8s

k8s_id = new_global_k8s(name=None, host=None, port=None, ssl_verify=True, token=None)

此方法用于添加新的 k8s 集群连接参数。

参数

返回值

• k8s_id：新建立的 k8s 集群的 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

k8s_id = client.new_global_k8s(name = "k8s", host = "192.168.1.1", port = 443, token = "xxxxx")

# put_global_k8s

ok = put_global_k8s(**kwargs)

此方法用于修改 Kubernetes 集群的连接参数。

参数

返回值

• 返回 True：表示成功。
• 返回 False：表示失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

ok = client.put_global_k8s(k8s_id = 1, name = "k8s", host = "192.168.1.1", port = 443, token = "xxxxx")

# get_global_k8s

data = get_global_k8s(k8s_id)

此方法用于获取 Kubernetes 集群的连接参数信息。

参数

返回值

• data：返回一个 dict 类型的数据，包含 Kubernetes 集群的详细信息。

示例

from edge2client import Edge2Client

# 创建一个 Edge2Client 实例
client = Edge2Client(host, username, password)

# 登录客户端
client.login()

# 使用 get_global_k8s 方法获取 Kubernetes 集群信息
data = client.get_global_k8s(k8s_id = 1)

# get_k8s_service_detail

data = get_k8s_service_detail()

此方法用于获取 k8s 集群中定义的服务相关信息。

参数

返回值

• 返回一个 dict 类型的数据，包含 k8s 集群的详细信息。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.get_k8s_service_detail(k8s_id = 1)

# get_all_global_k8s

data = get_all_global_k8s()

此方法用于获取所有全局 Kubernetes 集群的 id。

返回值

• data：返回一个 list 类型的 Kubernetes 集群信息。

应用示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

data = client.get_all_k8s()

# del_global_k8s

ok = del_global_k8s(k8s_id)

此方法用于删除 k8s 集群的连接参数。

参数

返回值

• True：表示删除成功。
• False：表示删除失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

ok = client.del_global_k8s(k8s_id = 1)

以上代码展示了如何使用 Edge2Client 删除特定的 k8s 集群连接参数。

# update_proxy_rule

result = update_proxy_rule(**kwargs)

更新指定的上游代理规则。

参数

返回值

• True：代表成功更新代理规则。
• False：代表更新代理规则失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

rule_id = 1
proxy_rule = [...]

result = client.update_proxy_rule(rule_id=rule_id, proxy=proxy_rule)

此方法用于更新全局级别的 IP 规则列表。

# new_global_ip_list

data = new_global_ip_list(name=None, type='ipv4', items=None)

此方法用于创建新的全局 IP 列表。

参数

items 参数

返回值

• id：新创建的全局 IP 列表的唯一标识 ID。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

ok = client.new_global_ip_list(name = 'ip-list-1',
                               type = 'ipv4',
                               items = [
                                    {'ip': '127.0.0.0/24'},
                                    {'ip': '192.168.1.0/24'},
                                    {'ip': '192.168.2.0/24', 'exptime': 1725000000},
                                    {'ip': '192.168.3.0/24', 'exptime': 1735000000}
                               ])

# put_global_ip_list

data = put_global_ip_list(rule_id=None, name=None, type='ipv4', items=None)

此方法用于修改指定的全局级别 IP 列表。

参数

返回值

• True：表明修改成功。
• False：表明修改失败。

示例

from edge2client import Edge2Client

client = Edge2Client(host, username, password)
client.login()

ok = client.put_global_ip_list(rule_id = 1, items = [ {'ip': '127.0.0.2'} ])

# del_global_ip_list

ok = del_global_ip_list(rule_id=None)