在GCP中创建读取BigQuery数据的REST API（Python端到端方案咨询）

整体架构思路 #

推荐两种云原生实现路径：

• Cloud Functions：轻量无服务器，适合快速搭建简单API，无需管理服务器
• Cloud Run：容器化无服务器，适合需要自定义环境、扩缩容更灵活的场景

两者都能直接和BigQuery集成，通过服务账号权限控制数据访问，对外提供HTTP接口供外部应用调用。

步骤分解与示例代码 #

一、前置准备 #

• 安装并初始化gcloud SDK，完成本地认证
• 启用GCP服务：BigQuery API、Cloud Functions API（选前者）或Cloud Run API（选后者）
• 确保你的GCP账号拥有BigQuery数据读取权限，以及对应服务的部署权限

二、方案1：Cloud Functions（快速实现） #

代码实现（main.py）

import functions_framework
from google.cloud import bigquery
import json

# 初始化BigQuery客户端
client = bigquery.Client()

@functions_framework.http
def bigquery_api(request):
    # 设置CORS，允许外部域名调用（可选，根据实际需求调整）
    headers = {
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Methods': 'GET, POST'
    }

    # 处理OPTIONS请求（CORS预检）
    if request.method == 'OPTIONS':
        return ('', 204, headers)

    # 从请求参数获取查询条件（示例：获取指定用户的数据）
    request_args = request.args
    user_id = request_args.get('user_id', default=None, type=str)
    
    if not user_id:
        return (json.dumps({"error": "Missing user_id parameter"}), 400, headers)

    # BigQuery查询语句
    query = f"""
        SELECT user_name, email, created_at
        FROM `your-project-id.your-dataset.your-table`
        WHERE user_id = @user_id
    """

    # 使用参数化查询防止SQL注入
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("user_id", "STRING", user_id)
        ]
    )

    try:
        # 执行查询
        query_job = client.query(query, job_config=job_config)
        results = query_job.result()

        # 将查询结果转为JSON格式
        data = [dict(row) for row in results]
        return (json.dumps({"data": data}), 200, headers)
    except Exception as e:
        return (json.dumps({"error": str(e)}), 500, headers)

部署命令

gcloud functions deploy bigquery-api \
    --runtime python311 \
    --trigger-http \
    --allow-unauthenticated \
    --service-account your-service-account@your-project-id.iam.gserviceaccount.com

注意：--allow-unauthenticated会允许公开访问，生产环境建议改用IAM认证或API Key限制访问

权限配置

给部署时指定的服务账号添加BigQuery Data Viewer角色，确保它能读取目标BigQuery数据集/表。

三、方案2：Cloud Run（容器化灵活部署） #

1. 代码实现（main.py，使用FastAPI）

from fastapi import FastAPI, HTTPException
from google.cloud import bigquery
from pydantic import BaseModel
import uvicorn

app = FastAPI()
client = bigquery.Client()

# 定义请求参数模型
class QueryRequest(BaseModel):
    user_id: str

@app.get("/api/get-user-data")
async def get_user_data(user_id: str):
    if not user_id:
        raise HTTPException(status_code=400, detail="Missing user_id parameter")
    
    query = f"""
        SELECT user_name, email, created_at
        FROM `your-project-id.your-dataset.your-table`
        WHERE user_id = @user_id
    """
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("user_id", "STRING", user_id)
        ]
    )
    
    try:
        query_job = client.query(query, job_config=job_config)
        results = query_job.result()
        data = [dict(row) for row in results]
        return {"data": data}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/api/get-user-data")
async def post_user_data(request: QueryRequest):
    return await get_user_data(request.user_id)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8080)

2. Dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY main.py .

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]

3. requirements.txt

fastapi==0.104.1
uvicorn==0.24.0.post1
google-cloud-bigquery==3.12.0
pydantic==2.5.0

4. 部署命令

# 构建镜像
gcloud builds submit --tag gcr.io/your-project-id/bigquery-api-service

# 部署到Cloud Run
gcloud run deploy bigquery-api-service \
    --image gcr.io/your-project-id/bigquery-api-service \
    --platform managed \
    --allow-unauthenticated \
    --service-account your-service-account@your-project-id.iam.gserviceaccount.com

权限配置

同样给服务账号添加BigQuery Data Viewer角色，生产环境建议关闭--allow-unauthenticated，使用Cloud IAP或API网关控制访问。

四、测试API #

部署完成后，你会得到一个API端点，比如：

• Cloud Functions：https://REGION-PROJECT_ID.cloudfunctions.net/bigquery-api?user_id=123
• Cloud Run：https://bigquery-api-service-xxxxxx-REGION.run.app/api/get-user-data?user_id=123

用curl或Postman发送请求，验证是否能正常返回BigQuery中的数据。

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