无需服务账号实现500+用户通过Python访问BigQuery表

1. 采用用户身份OAuth 2.0认证 #

让每个用户用自身Google账号通过OAuth 2.0完成认证，直接访问BigQuery：

• 无需分发服务账号密钥，用户以个人身份登录后，BigQuery会自动在审计日志、Cloud Logging中记录单个用户的所有访问活动，完全满足追踪需求。
• Python端可借助google-cloud-bigquery库配合OAuth流程，用户首次运行代码时会跳转至Google授权页面，后续凭据会自动缓存，无需重复授权。
• 只需在Google Cloud Console创建一个「桌面应用」或「其他」类型的OAuth客户端ID，将ID信息配置到代码中即可。

示例代码片段：

from google.cloud import bigquery
from google.oauth2 import user_credentials

# 替换为你的OAuth客户端配置
client_config = {
    "installed": {
        "client_id": "你的客户端ID",
        "client_secret": "你的客户端密钥",
        "auth_uri": "https://accounts.google.com/o/oauth2/auth",
        "token_uri": "https://oauth2.googleapis.com/token"
    }
}

# BigQuery访问权限范围
scopes = ["https://www.googleapis.com/auth/bigquery"]

# 获取用户凭据
credentials = user_credentials.UserCredentials.from_client_config(
    client_config, scopes
)

# 初始化BigQuery客户端
client = bigquery.Client(credentials=credentials, project="你的项目ID")

# 执行查询
query_job = client.query("SELECT * FROM `你的项目.数据集.表名` LIMIT 100")
results = query_job.result()

2. 用用户组管理IAM权限（组织内部用户适用） #

如果500名用户均属于你的企业/组织，可通过Google Workspace或Cloud Identity创建用户组：

• 创建一个BigQuery专属用户组（如bq-users@你的域名.com），将所有用户批量加入组中。
• 在BigQuery数据集或表的IAM设置里，给该组分配对应权限（比如roles/bigquery.dataViewer用于读数据，roles/bigquery.jobUser用于提交查询任务，按需调整）。
• 结合OAuth 2.0认证，用户用个人账号登录后自动继承组权限，同时BigQuery会记录每个用户的操作日志，既简化权限管理，又满足追踪需求。

3. 优化自建API方案（若坚持此路径） #

如果已经在搭建API，需重点解决用户身份验证和活动溯源问题：

• API后端使用单个服务账号（需具备BigQuery访问权限），但在API层先做用户身份校验（比如用JWT、内部系统凭证或OAuth）。
• 执行BigQuery查询时，通过job_user参数指定当前用户的邮箱（服务账号需拥有roles/bigquery.admin或roles/bigquery.jobUser权限），这样BigQuery审计日志会显示该用户为查询发起者，实现活动追踪。

示例Flask后端代码：

from flask import Flask, request, jsonify
from google.cloud import bigquery
import jwt

app = Flask(__name__)
client = bigquery.Client(project="你的项目ID")

# 用于验证JWT的密钥（需与前端保持一致）
SECRET_KEY = "你的密钥"

@app.route('/query', methods=['POST'])
def run_query():
    # 解析用户JWT令牌
    auth_header = request.headers.get('Authorization')
    if not auth_header or not auth_header.startswith('Bearer '):
        return jsonify({"error": "身份验证失败"}), 401
    
    token = auth_header.split(' ')[1]
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=['HS256'])
        user_email = payload['email']
    except:
        return jsonify({"error": "无效令牌"}), 401
    
    # 获取查询语句
    query = request.json.get('query')
    if not query:
        return jsonify({"error": "未提供查询语句"}), 400
    
    # 执行查询并指定发起用户
    job_config = bigquery.QueryJobConfig(job_user=user_email)
    query_job = client.query(query, job_config=job_config)
    results = [dict(row) for row in query_job.result()]
    
    return jsonify(results)

if __name__ == '__main__':
    app.run()

避坑提示 #

• 不要用API密钥直接开放BigQuery访问：API密钥仅验证请求来源，无法识别用户身份，既无法追踪单个用户活动，也容易因密钥泄露导致权限滥用。
• 不要尝试突破服务账号上限：Google Cloud的服务账号数量限制是硬规则，强行超额创建会触发风控，且不符合安全最佳实践。

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