使用 Python 管理 Snowflake 网络策略¶

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

备注

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)

代码可通过生成的 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)

获取网络策略详细信息¶

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

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

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

列出网络策略¶

您可以使用 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)

删除网络策略¶

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

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

my_network_policy_res = root.network_policies["my_network_policy"]
my_network_policy_res.drop()