如何通过查询/Python方法获取BigQuery表中用户可访问列列表？

可以通过BigQuery内置的信息架构视图或Python结合BigQuery API两种方式实现需求。

一、使用BigQuery查询语句 #

利用INFORMATION_SCHEMA下的权限视图，可直接查询用户的列级访问权限，包括直接授权和通过角色继承的权限。

1. 查询直接列级权限 #

如果用户的权限是直接授予到列的，用以下语句：

SELECT
  table_name,
  column_name,
  grantee
FROM
  `[项目ID].[数据集ID].INFORMATION_SCHEMA.COLUMN_PRIVILEGES`
WHERE
  grantee IN ('userA@yourdomain.com', 'userB@yourdomain.com')
  AND table_name = 'TableA'
ORDER BY grantee, column_name;

2. 包含角色继承的完整权限查询 #

很多场景下用户权限来自角色，需要关联角色授权视图来统计：

WITH user_assigned_roles AS (
  -- 获取用户直接拥有的角色
  SELECT grantee, role_name
  FROM `[项目ID].[数据集ID].INFORMATION_SCHEMA.ROLE_GRANTS`
  WHERE grantee IN ('userA@yourdomain.com', 'userB@yourdomain.com')
  UNION ALL
  -- 获取角色继承的角色（递归查询，这里只做一层，如需多层可扩展）
  SELECT r.grantee, g.role_name
  FROM `[项目ID].[数据集ID].INFORMATION_SCHEMA.ROLE_GRANTS` r
  JOIN `[项目ID].[数据集ID].INFORMATION_SCHEMA.ROLE_GRANTS` g
  ON r.role_name = g.grantee
  WHERE r.grantee IN ('userA@yourdomain.com', 'userB@yourdomain.com')
),
role_column_permissions AS (
  -- 通过角色获取列权限
  SELECT
    cp.table_name,
    cp.column_name,
    ur.grantee
  FROM `[项目ID].[数据集ID].INFORMATION_SCHEMA.COLUMN_PRIVILEGES` cp
  JOIN user_assigned_roles ur ON cp.grantee = ur.role_name
)
-- 合并直接权限和角色权限，去重
SELECT * FROM role_column_permissions
UNION DISTINCT
SELECT table_name, column_name, grantee
FROM `[项目ID].[数据集ID].INFORMATION_SCHEMA.COLUMN_PRIVILEGES`
WHERE grantee IN ('userA@yourdomain.com', 'userB@yourdomain.com')
  AND table_name = 'TableA'
ORDER BY grantee, column_name;

二、使用Python方法 #

借助google-cloud-bigquery库，通过模拟用户权限执行测试查询的方式，验证用户对每一列的访问权限。

代码示例 #

from google.cloud import bigquery

def get_user_accessible_columns(project_id, dataset_id, table_name, target_users):
    client = bigquery.Client(project=project_id)
    table_ref = f"{dataset_id}.{table_name}"
    table = client.get_table(table_ref)
    user_columns = {}

    for user in target_users:
        accessible = []
        # 模拟用户身份执行查询（需确保当前账号有模拟权限）
        impersonated_client = bigquery.Client(
            project=project_id,
            impersonated_service_account=user
        )
        for col in table.schema:
            test_query = f"SELECT `{col.name}` FROM `{table_ref}` LIMIT 1"
            try:
                # 执行查询，成功则说明有权限
                impersonated_client.query(test_query).result()
                accessible.append(col.name)
            except Exception:
                # 查询失败，无权限
                pass
        user_columns[user] = accessible
    return user_columns

# 调用示例
result = get_user_accessible_columns(
    "your-project-id",
    "your-dataset",
    "TableA",
    ["userA@yourdomain.com", "userB@yourdomain.com"]
)
for user, cols in result.items():
    print(f"{user} 可访问列: {', '.join(cols)}")

注意：使用此方法时，运行代码的账号需要拥有roles/iam.serviceAccountTokenCreator权限，才能模拟目标用户执行查询。

内容的提问来源于stack exchange，提问作者develop