使用 Python 管理 Snowflake 用户、角色和授权

您可以使用 Python 来管理 Snowflake 用户、角色和授权。有关在 Snowflake 中管理用户及其权限的更多信息，请参阅 用户管理。

先决条件

在本主题中的示例中，假设您已添加了用来连接 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。

管理用户

您可以在 Snowflake 中管理用户。在 Snowflake 中，用户是账户级对象。Snowflake Python APIs 使用两种不同类型来表示用户：

  • User：显示用户的属性，例如其名称。

  • UserResource：显示可用于提取相应 User 对象和删除用户的方法。

创建用户

您可以通过调用 UserCollection.create 方法并传递一个表示要创建的用户的 User 对象来创建用户。要创建用户，请首先创建一个指定用户名的 User 对象。

以下示例中的代码会创建一个表示用户 my_user 的 User 对象，然后通过将该 User 对象传递给 UserCollection.create 方法来创建该用户：

from snowflake.core.user import User

my_user = User(name="my_user")
root.users.create(my_user)

获取用户详细信息

您可以通过调用 UserResource.fetch 方法来获取关于某个用户的信息，该方法会返回一个 User 对象。

以下示例中的代码会获取有关名为 my_user 的用户的信息：

my_user = root.users["my_user"].fetch()
print(my_user.to_dict())

创建或更改用户

您可以设置 User 对象的属性并将其传递给 UserResource.create_or_alter 方法以创建用户（如果用户不存在）或者根据用户定义更改用户（如果用户存在）。create_or_alter 的行为旨在实现幂等性，这意味着无论在调用该方法之前用户是否存在，生成的用户对象都是一样的。

对于任何没有显式定义的 用户 属性，create_or_alter 会使用默认值。例如，如果您没有设置 snowflake_support，则即使用户以前以不同的值存在，其值也会默认为 False。

备注

create_or_alter 方法目前不支持更改现有用户的 password。您仅可在创建新用户时设置密码。

以下示例中的代码更新了 my_user 用户的名字、姓氏和 must_change_password 的属性，然后在 Snowflake 上更改用户：

user_parameters = root.users["my_user"].fetch()
user_parameters.first_name="Snowy"
user_parameters.last_name="User"
user_parameters.must_change_password=False
root.users["my_user"].create_or_alter(user_parameters)

列出用户

您可以使用 iter 方法列出用户，该方法会返回一个 PagedIter 迭代器。

以下示例中的代码会列出名称以 my 开头的用户：

users = root.users.iter(like="my%")
for user in users:
  print(user.name)

删除用户

您可以使用 UserResource.drop 方法删除用户。

以下示例中的代码会删除 my_user 用户：

my_user_res = root.users["my_user"]
my_user_res.drop()

管理角色

您可以在 Snowflake 中管理角色。角色是账户级对象。Snowflake Python APIs 使用两种不同类型来表示角色：

  • Role：显示角色的属性，例如其名称。

  • RoleResource：显示可用于授予和管理相应 Role 对象的权限以及删除角色的方法。

创建角色

要创建角色，请先创建一个指定角色名称的 Role 对象。

以下示例中的代码会创建一个 Role 对象，用于表示一个名为 my_role 角色：

from snowflake.core.role import Role

my_role = Role(name="my_role")
root.roles.create(my_role)

然后，代码通过将 Role 对象传递给 RoleCollection.create 方法创建角色。

在会话中使用角色

以下示例中的代码会在当前会话中应用角色 my_role。

root.session.use_role("my_role")

列出角色

您可以使用 iter 方法列出账户中的角色。该方法会返回 Role 对象的 PagedIter 迭代器。

以下示例中的代码会列出账户中的所有角色名称：

role_list = root.roles.iter()
for role_obj in role_list:
  print(role_obj.name)

删除角色

您可以使用 RoleResource.drop 方法删除角色。

以下示例中的代码会删除 my_role 角色：

my_role_res = root.roles["my_role"]
my_role_res.drop()

管理数据库角色

您可以在 Snowflake 中管理 数据库角色。数据库角色是数据库级对象。Snowflake Python APIs 使用两种不同类型来表示数据库角色：

  • DatabaseRole：显示数据库角色的属性，例如其名称和注释。

  • DatabaseRoleResource：显示可用于授予和管理相应 DatabaseRole 对象的权限以及删除数据库角色的方法。

创建数据库角色

要创建数据库角色，请先创建一个指定角色名称的 DatabaseRole 对象。

以下示例中的代码会创建一个 DatabaseRole 对象，用于表示一个名为 my_db_role 数据库角色：

from snowflake.core.database_role import DatabaseRole

my_db_role = DatabaseRole(
  name="my_db_role",
  comment="sample comment"
)

my_db_role_ref = root.databases['my_db'].database_roles.create(my_db_role)

然后，代码通过将 DatabaseRole 对象传递给 DatabaseRoleCollection.create 方法创建数据库角色。

克隆数据库角色

以下示例中的代码会在 my_db_2 目标数据库中创建一个名为 dr2 的数据库角色，作为 my_db 数据库中现有 dr1 数据库角色的副本。

database_role_ref = root.databases['my_db'].database_roles['dr1'].clone(target_database_role='dr2', target_database='my_db_2')

列出数据库角色

您可以使用 iter 方法列出账户中的数据库角色。该方法会返回 DatabaseRole 对象的 PagedIter 迭代器。

以下示例中的代码会列出 my_db 数据库中名为 my_db_role 的数据库角色，将结果数限制为 1：

db_role_list = root.databases['my_db'].database_roles.iter(limit=1, from_name='my_db_role')
for db_role_obj in db_role_list:
  print(db_role_obj.name)

删除数据库角色

您可以使用 DatabaseRoleResource.drop 方法删除数据库角色。

以下示例中的代码会删除 my_db_role 数据库角色：

root.databases['my_db'].database_roles['my_db_role'].drop()

管理访问权限

您可以使用 API 管理账户角色、数据库角色或用户对安全 Snowflake 对象的访问权限。有关 Snowflake 中的角色、安全对象和访问控制框架的更多信息，请参阅 访问控制概述。

对于账户角色

以下代码示例演示了为 账户角色 授予权限、撤销权限和列出授权的 API 操作。

授予权限

from snowflake.core.role import Securable

root.roles['my_role'].grant_privileges(
    privileges=["OPERATE"], securable_type="WAREHOUSE", securable=Securable(name='my_wh')
)

授予角色

from snowflake.core.role import Securable

root.roles['my_role'].grant_role(role_type="ROLE", role=Securable(name='my_role_1'))

授予所有权限

from snowflake.core.role import ContainingScope

root.roles['my_role'].grant_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

授予未来权限

from snowflake.core.role import ContainingScope

root.roles['my_role'].grant_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

撤销权限

from snowflake.core.role import Securable

root.roles['my_role'].revoke_privileges(
    privileges=["OPERATE"], securable_type="WAREHOUSE", securable=Securable(name='my_wh')
)

撤销角色

from snowflake.core.role import Securable

root.roles['my_role'].revoke_role(role_type="ROLE", role=Securable(name='my_role_1'))

撤销所有权限

from snowflake.core.role import ContainingScope

root.roles['my_role'].revoke_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

撤销未来权限

from snowflake.core.role import ContainingScope

root.roles['my_role'].revoke_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

权限的撤销授权选项

from snowflake.core.role import Securable

 root.roles['my_role'].revoke_grant_option_for_privileges(
    privileges=["OPERATE"], securable_type="WAREHOUSE", securable=Securable(name='my_wh')
)

所有权限的撤销授权选项

from snowflake.core.role import ContainingScope

root.roles['my_role'].revoke_grant_option_for_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

未来权限的撤销授权选项

from snowflake.core.role import ContainingScope

root.roles['my_role'].revoke_grant_option_for_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

列出对角色的授权

root.roles['my_role'].iter_grants_to()

列出有关角色的授权

root.roles['my_role'].iter_grants_on()

列出角色的授权

root.roles['my_role'].iter_grants_of()

列出对角色的未来授权

root.roles['my_role'].iter_future_grants_to()

对于用户

以下代码示例演示了为用户授予角色、撤销角色和列出角色的 API 操作。

向用户授予角色

from snowflake.core.user import Securable

root.users['my_user'].grant_role(role_type="ROLE", role=Securable(name='my_role'))

撤销用户的角色

from snowflake.core.user import Securable

root.users['my_user'].revoke_role(role_type="ROLE", role=Securable(name='my_role'))

列出授予用户的角色

root.users['my_user'].iter_grants_to()

对于数据库角色

以下代码示例演示了为 数据库角色 授予权限、撤销权限和列出授权的 API 操作。

授予权限

from snowflake.core.database_role import Securable

root.databases['my_db'].database_roles['my_db_role'].grant_privileges(
    privileges=["MODIFY"], securable_type="DATABASE", securable=Securable(name='my_db')
)

授予角色

from snowflake.core.database_role import Securable

root.databases['my_db'].database_roles['my_db_role'].grant_role(role_type="DATABASE ROLE", role=Securable(name='my_db_role_1'))

授予所有权限

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].grant_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

授予未来权限

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].grant_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

撤销权限

from snowflake.core.database_role import Securable

root.databases['my_db'].database_roles['my_db_role'].revoke_privileges(
    privileges=["MODIFY"], securable_type="DATABASE", securable=Securable(name='my_db')
)

撤销角色

from snowflake.core.database_role import Securable

root.databases['my_db'].database_roles['my_db_role'].revoke_role(role_type="DATABASE ROLE", role=Securable(name='my_db_role_1'))

撤销所有权限

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].revoke_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

撤销未来权限

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].revoke_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

权限的撤销授权选项

from snowflake.core.database_role import Securable

root.databases['my_db'].database_roles['my_db_role'].revoke_grant_option_for_privileges(
    privileges=["MODIFY"], securable_type="DATABASE", securable=Securable(name='my_db')
)

所有权限的撤销授权选项

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].revoke_grant_option_for_privileges_on_all(
    privileges=["SELECT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

未来权限的撤销授权选项

from snowflake.core.database_role import ContainingScope

root.databases['my_db'].database_roles['my_db_role'].revoke_grant_option_for_future_privileges(
    privileges=["SELECT", "INSERT"],
    securable_type="TABLE",
    containing_scope=ContainingScope(database='my_db', schema='my_schema'),
)

列出对角色的授权

root.databases['my_db'].database_roles['my_db_role'].iter_grants_to()

列出对角色的未来授权

root.databases['my_db'].database_roles['my_db_role'].iter_future_grants_to()

使用 Grant 资源管理授权 – 已弃用

您可以执行 GRANT <privileges> ... TO ROLE 操作以向角色授予对安全 Snowflake 对象的访问权限。

授予权限

要授予 Snowflake 对象的权限，请首先创建一个指定以下属性的 Grant 对象：

  • grantee：要授予权限的角色或用户。

  • securable：受权限保护的 Snowflake 对象。

  • privileges：要授予某个角色的权限。

向角色授予账户的 CREATE 权限

以下示例中的代码会创建一个表示授予操作的 Grant 对象，该操作会向当前 Snowflake 账户中的角色 my_role 授予权限 create_database 和 create_warehouse。代码使用 root.grants.grant 方法执行操作。

from snowflake.core.grant import Grant
from snowflake.core.grant._grantee import Grantees
from snowflake.core.grant._privileges import Privileges
from snowflake.core.grant._securables import Securables

root.grants.grant(
  Grant(
    grantee=Grantees.role(name='my_role'),
    securable=Securables.current_account,
    privileges=[Privileges.create_database,
                Privileges.create_warehouse],
  )
)

向角色授予数据库权限

以下示例中的代码会向角色 my_role 授予对数据库 my_db 的 导入权限：

from snowflake.core.grant import Grant
from snowflake.core.grant._grantee import Grantees
from snowflake.core.grant._privileges import Privileges
from snowflake.core.grant._securables import Securables

root.grants.grant(
  Grant(
    grantee=Grantees.role('my_role'),
    securable=Securables.database('my_db'),
    privileges=[Privileges.imported_privileges],
  )
)

将一个角色授予另一个角色

您可以将角色分配给另一个角色，以在角色之间建立“父子”关系（也称为 角色层次结构）。

以下示例中的代码会向 ACCOUNTADMIN 系统角色授予 my_role 用户角色：

from snowflake.core.grant import Grant
from snowflake.core.grant._grantee import Grantees
from snowflake.core.grant._securables import Securables

root.grants.grant(
  Grant(
    grantee=Grantees.role('ACCOUNTADMIN'),
    securable=Securables.role('my_role'),
  )
)