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