如何用Python递归遍历Snowflake角色授权层级并生成无重复列表?
遍历Snowflake角色层级的Python实现
获取子角色的SQL语句
要筛选指定角色的直接子角色(仅ROLE类型的授权),可以用Snowflake的INFORMATION_SCHEMA.ROLE_GRANTS视图,这个视图结构化存储了所有角色授权关系,比SHOW GRANTS更适合程序化查询:
SELECT GRANTEE_NAME FROM INFORMATION_SCHEMA.ROLE_GRANTS WHERE GRANTED_ROLE_NAME = %s AND GRANTEE_TYPE = 'ROLE';
用参数化查询避免SQL注入,替换%s为目标角色名即可。
Python递归遍历实现
以下代码使用snowflake-connector-python库,递归遍历角色层级,通过集合自动去重,最终返回无重复的角色列表:
import snowflake.connector def get_snowflake_connection(): # 替换为你的Snowflake连接参数 return snowflake.connector.connect( user='YOUR_USER', password='YOUR_PASSWORD', account='YOUR_ACCOUNT', warehouse='YOUR_WAREHOUSE', database='YOUR_DATABASE', schema='INFORMATION_SCHEMA' ) def traverse_role_hierarchy(start_role, conn, visited=None): if visited is None: visited = set() # 避免重复遍历已访问的角色 if start_role in visited: return visited visited.add(start_role) # 查询当前角色的直接子角色 cursor = conn.cursor() try: cursor.execute( """ SELECT GRANTEE_NAME FROM INFORMATION_SCHEMA.ROLE_GRANTS WHERE GRANTED_ROLE_NAME = %s AND GRANTEE_TYPE = 'ROLE'; """, (start_role,) ) child_roles = [row[0] for row in cursor.fetchall()] finally: cursor.close() # 递归遍历每个子角色 for role in child_roles: traverse_role_hierarchy(role, conn, visited) return visited if __name__ == '__main__': target_role = 'YOUR_START_ROLE' with get_snowflake_connection() as conn: role_set = traverse_role_hierarchy(target_role, conn) # 转换为排序后的列表(可选) role_list = sorted(role_set) print("遍历得到的角色列表:") for role in role_list: print(role)
关键说明
- 去重机制:使用
set存储已访问角色,自动跳过重复角色,避免循环遍历(比如角色A授予B,B又授予A的情况) - 递归终止:当角色已在
visited集合中时,直接返回,停止递归 - 连接管理:使用上下文管理器(
with)确保连接自动关闭,避免资源泄漏 - 权限要求:执行代码的Snowflake用户需要有访问
INFORMATION_SCHEMA.ROLE_GRANTS的权限(通常需要ACCOUNTADMIN或类似权限)
内容的提问来源于stack exchange,提问作者SnowflakeDev
相关产品推荐
相关产品推荐

