使用 Python 管理 Snowflake 网络策略

您可以使用 Python 来管理 Snowflake 网络策略，这些策略可用于控制对 Snowflake 服务和内部暂存区的入站访问。有关更多信息，请参阅 使用网络策略控制网络流量。

备注

ALTER NETWORK POLICY 目前不受支持。

Snowflake Python APIs 使用两种不同类型来表示网络策略：

  • NetworkPolicy：显示网络策略的属性，例如其名称、网络规则以及允许和阻止的 IP 列表。

  • NetworkPolicyResource：显示可用于提取相应 NetworkPolicy 对象和删除网络策略的方法。

先决条件

在本主题中的示例中，假设您已添加了用来连接 Snowflake 和创建 Root 对象以使用 Snowflake Python APIs 的代码。

例如，以下代码使用配置文件中定义的连接参数来创建与 Snowflake 的连接：

from snowflake.core import Root
from snowflake.snowpark import Session

session = Session.builder.config("connection_name", "myconnection").create()
root = Root(session)
Copy

代码可通过生成的 Session 对象创建 Root 对象，从而使用 API 的类型和方法。有关更多信息，请参阅 使用 Snowflake Python APIs 连接到 Snowflake。

创建网络策略

要创建网络策略，请先创建一个 NetworkPolicy 对象，然后根据 API Root 对象创建 NetworkPolicyCollection 对象。使用 NetworkPolicyCollection.create，将新的网络策略添加到 Snowflake。

以下示例中的代码会创建一个 NetworkPolicy 对象，该对象表示一项名为 my_network_policy 的网络策略，其中包含指定的允许和阻止的网络规则及 IP 地址：

from snowflake.core.network_policy import NetworkPolicy

my_network_policy = NetworkPolicy(
  name = 'my_network_policy',
  allowed_network_rule_list = ['allowed_network_rule1','allowed_network_rule2'],
  blocked_network_rule_list = ['blocked_network_rule1','blocked_network_rule2'],
  allowed_ip_list=['8.8.8.8'],
  blocked_ip_list=['192.100.123.0'],
)

root.network_policies.create(my_network_policy)
Copy

获取网络策略详细信息

您可以通过调用 NetworkPolicyResource.fetch 方法来获取关于网络策略的信息，该方法会返回一个 NetworkPolicy 对象。

以下示例中的代码会获取有关名为 my_network_policy 的网络策略的信息：

my_network_policy = root.network_policies["my_network_policy"].fetch()
print(my_network_policy.to_dict())
Copy

列出网络策略

您可以使用 NetworkPolicyCollection.iter 方法列出网络策略，该方法会返回 NetworkPolicy 对象的 PagedIter 迭代器。

以下示例中的代码会列出名称中以 my 为开头的网络策略，并打印每个网络策略的名称：

network_policy_iter = root.network_policies.iter(like="my%")  # returns a PagedIter[NetworkPolicy]
for network_policy_obj in network_policy_iter:
  print(network_policy_obj.name)
Copy

删除网络策略

您可以使用 NetworkPolicyResource 对象删除网络策略。

以下示例中的代码会获取 my_network_policy 网络策略资源对象，然后删除网络策略。

my_network_policy_res = root.network_policies["my_network_policy"]
my_network_policy_res.drop()
Copy
语言: 中文