求基于Python API+Swagger/Postman的BigQuery数据插入方案及代码

方案说明 #

采用FastAPI构建API，它自带Swagger UI（开箱即用的接口测试页面），同时兼容Postman调用；使用Google Cloud BigQuery官方SDK完成数据插入，修正原有代码中的语法错误与逻辑冗余。

依赖安装 #

执行以下命令安装所需依赖：

pip install fastapi uvicorn google-cloud-bigquery pydantic

完整实现代码 #

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

# 初始化FastAPI应用
app = FastAPI(title="BigQuery Data Insert API", version="1.0")

# 配置BigQuery认证
# 替换为你的服务账号密钥文件路径
credentials = service_account.Credentials.from_service_account_file('path/to/your-service-account-key.json')
project_id = 'your-project-id'
client = bigquery.Client(credentials=credentials, project=project_id)

# 定义请求数据模型（用于Swagger自动生成参数表单）
class DataInput(BaseModel):
    group_name: str
    user_name: str

@app.post("/add-data", summary="向BigQuery表插入数据")
def add_data_to_bigquery(data: DataInput):
    try:
        # BigQuery参数化查询（避免SQL注入，使用@占位符）
        query = """
            INSERT INTO `your-project-id.your-dataset.your-table` (group, user)
            VALUES (@group_name, @user_name)
        """
        # 配置查询参数
        job_config = bigquery.QueryJobConfig(
            query_parameters=[
                bigquery.ScalarQueryParameter("group_name", "STRING", data.group_name),
                bigquery.ScalarQueryParameter("user_name", "STRING", data.user_name)
            ]
        )
        # 执行查询并等待完成
        query_job = client.query(query, job_config=job_config)
        query_job.result()
        
        return {"status": "success", "message": "数据已成功插入BigQuery"}
    except Exception as e:
        # 抛出具体异常信息，便于调试
        raise HTTPException(status_code=500, detail=f"插入失败: {str(e)}")

# 运行方式：在终端执行 uvicorn main:app --reload

关键修正点（对比原有代码） #

• 移除冗余的SQLite连接代码：原有代码中SQLite逻辑未实际使用，直接删除
• 修正BigQuery参数化查询：使用@占位符+ScalarQueryParameter传递参数，避免SQL注入，符合BigQuery语法规范
• 初始化job_config实例：原有代码直接调用未定义的job_config，现在正确创建配置对象
• 引入FastAPI框架：暴露POST接口，自动生成Swagger UI，支持Postman调用
• 优化异常处理：捕获具体异常并返回HTTP状态码与详细信息，替代模糊的print输出

使用步骤 #

1. 替换代码中的服务账号密钥路径、项目ID、BigQuery表路径（格式：项目ID.数据集ID.表名）
2. 运行API：在终端执行 uvicorn main:app --reload
3. Swagger测试：访问 http://localhost:8000/docs，找到/add-data接口，点击"Try it out"输入参数后执行
4. Postman调用：发送POST请求到 http://localhost:8000/add-data，Body选择raw->JSON，传入如下格式数据：

   {
       "group_name": "测试组",
       "user_name": "测试用户"
   }
   

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