CloudRun上FastAPI应用连接BigQuery的正确方法及客户端初始化位置

直接给结论：在模块级（外部作用域）初始化BigQuery客户端，然后在各个函数里复用它，这是GCP官方推荐的做法，理由如下：

• 减少资源开销：BigQuery客户端内部维护了连接池、认证缓存和会话状态，每次初始化客户端都会重新建立这些资源，复用能避免重复创建的性能损耗，尤其在CloudRun这种可能处理大量请求的环境下更明显。
• 线程安全保障：google.cloud.bigquery.Client本身是线程安全的，FastAPI的请求处理（即使是异步模式）可以安全地共享同一个客户端实例，不用担心并发问题。
• 适配CloudRun实例模型：CloudRun的每个容器实例会持续处理多个请求直到被缩容，复用客户端能让整个实例生命周期内都高效利用资源，而不是每个请求都重新初始化。

另外，你的代码里有个容易踩坑的问题：SQL语句中的WHERE id = id不会正确使用参数，应该改成WHERE id = @id，不然参数化查询不起作用，会返回全表数据。

修正后的代码示例：

from google.cloud import bigquery

# 模块级初始化，全局复用
client = bigquery.Client()

async def get_data(id: int) -> list[dict]:
    query = """
        SELECT * 
        FROM `my-proj.my-dataset.my-table`
        WHERE id = @id
    """
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("id", "INT64", id)
       ]
    )
    # 注意：这里的client.query是同步方法，在async函数里调用会阻塞事件循环
    # 如果要优化异步性能，可以用asyncio.to_thread包装，或者使用BigQuery异步客户端
    query_job = client.query(query, job_config=job_config)

    results = []
    for row in query_job:
        results.append(dict(row.items()))
    return results

如果想进一步适配FastAPI的异步特性，避免阻塞事件循环，可以用asyncio.to_thread把同步的查询调用包装起来：

import asyncio
from google.cloud import bigquery

client = bigquery.Client()

async def get_data(id: int) -> list[dict]:
    query = """
        SELECT * 
        FROM `my-proj.my-dataset.my-table`
        WHERE id = @id
    """
    job_config = bigquery.QueryJobConfig(
        query_parameters=[
            bigquery.ScalarQueryParameter("id", "INT64", id)
       ]
    )

    # 用to_thread把同步调用放到线程池，不阻塞异步事件循环
    query_job = await asyncio.to_thread(client.query, query, job_config=job_config)
    
    results = []
    # 遍历结果也是同步操作，同样可以包装
    rows = await asyncio.to_thread(list, query_job)
    for row in rows:
        results.append(dict(row.items()))
    return results

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