使用 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)

Copy

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

Copy

获取用户详细信息¶

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

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

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

Copy

列出用户¶

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

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

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

Copy

删除用户¶

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

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

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

Copy

管理角色¶

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

Copy

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

在会话中使用角色¶

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

root.session.use_role("my_role")

Copy

列出角色¶

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

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

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

Copy

删除角色¶

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

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

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

Copy

管理数据库角色¶

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

Copy

然后，代码通过将 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')

Copy

列出数据库角色¶

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

Copy

删除数据库角色¶

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

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

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

Copy

管理访问权限¶

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

Copy

授予角色¶

from snowflake.core.role import Securable

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

Copy

授予所有权限¶

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'),
)

Copy

授予未来权限¶

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'),
)

Copy

撤消权限¶

from snowflake.core.role import Securable

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

Copy

撤销角色¶

from snowflake.core.role import Securable

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

Copy

撤消所有权限¶

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'),
)

Copy

撤销未来权限¶

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'),
)

Copy

权限的撤消授权选项¶

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')
)

Copy

所有权限的撤消授权选项¶

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'),
)

Copy

未来权限的撤消授权选项¶

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'),
)

Copy

列出对角色的授权¶

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

Copy

列出有关角色的授权¶

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

Copy

列出角色的授权¶

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

Copy

列出对角色的未来授权¶

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

Copy

对于用户¶

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

向用户授予角色¶

from snowflake.core.user import Securable

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

Copy

撤消用户的角色¶

from snowflake.core.user import Securable

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

Copy

列出授予用户的角色¶

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

Copy

对于数据库角色¶

以下代码示例演示了为数据库角色授予权限、撤消权限和列出授权的 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')
)

Copy

授予角色¶

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'))

Copy

授予所有权限¶

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'),
)

Copy

授予未来权限¶

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'),
)

Copy

撤消权限¶

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')
)

Copy

撤销角色¶

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'))

Copy

撤消所有权限¶

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'),
)

Copy

撤销未来权限¶

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'),
)

Copy

权限的撤消授权选项¶

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')
)

Copy

所有权限的撤消授权选项¶

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'),
)

Copy

未来权限的撤消授权选项¶

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'),
)

Copy

列出对角色的授权¶

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

Copy

列出对角色的未来授权¶

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

Copy

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

您可以执行 GRANT <privileges> 操作以向角色授予对安全 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],
  )
)

Copy

向角色授予数据库权限¶

以下示例中的代码会向角色 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],
  )
)

Copy

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

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

以下示例中的代码会向 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'),
  )
)

Copy

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

先决条件¶

管理用户¶

创建用户¶

获取用户详细信息¶

列出用户¶

删除用户¶

管理角色¶

创建角色¶

在会话中使用角色¶

列出角色¶

删除角色¶

管理数据库角色¶

创建数据库角色¶

克隆数据库角色¶

列出数据库角色¶

删除数据库角色¶

管理访问权限¶

对于账户角色¶

授予权限¶

授予角色¶

授予所有权限¶

授予未来权限¶

撤消权限¶

撤销角色¶

撤消所有权限¶

撤销未来权限¶

权限的撤消授权选项¶

所有权限的撤消授权选项¶

未来权限的撤消授权选项¶

列出对角色的授权¶

列出有关角色的授权¶

列出角色的授权¶

列出对角色的未来授权¶

对于用户¶

向用户授予角色¶

撤消用户的角色¶

列出授予用户的角色¶

对于数据库角色¶

授予权限¶

授予角色¶

授予所有权限¶

授予未来权限¶

撤消权限¶

撤销角色¶

撤消所有权限¶

撤销未来权限¶

权限的撤消授权选项¶

所有权限的撤消授权选项¶

未来权限的撤消授权选项¶

列出对角色的授权¶

列出对角色的未来授权¶

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

授予权限¶

向角色授予账户的 CREATE 权限¶

向角色授予数据库权限¶

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

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