如何在Python函数中动态添加BigQuery查询参数实现多条件过滤

完整修改后的函数代码 #

from google.cloud import bigquery

def query_bigquery_with_filters(client: bigquery.Client, sql_template: str, **kwargs):
    # 动态生成BigQuery查询参数
    query_params = []
    for param_name, param_value in kwargs.items():
        # 映射Python类型到BigQuery标准数据类型
        if isinstance(param_value, str):
            bq_type = "STRING"
        elif isinstance(param_value, int):
            bq_type = "INT64"
        elif isinstance(param_value, float):
            bq_type = "FLOAT64"
        elif isinstance(param_value, bool):
            bq_type = "BOOL"
        elif isinstance(param_value, bigquery.Date):
            bq_type = "DATE"
        elif isinstance(param_value, bigquery.Timestamp):
            bq_type = "TIMESTAMP"
        else:
            raise ValueError(f"参数 {param_name} 的类型 {type(param_value)} 暂不支持")
        
        query_params.append(bigquery.ScalarQueryParameter(param_name, bq_type, param_value))
    
    # 配置查询作业并执行
    job_config = bigquery.QueryJobConfig(query_parameters=query_params)
    query_job = client.query(sql_template, job_config=job_config)
    
    # 返回结果转为DataFrame（可根据需求调整返回格式）
    return query_job.result().to_dataframe()

用法示例 #

1. 定义带动态占位符的SQL模板

SELECT 
    country_cd, fcast_item, fcast_value, fcast_date
FROM `your-project.your-dataset.your-table`
WHERE 
    country_cd = @country_cd
    AND fcast_item = @fcast_item
    AND fcast_date >= @start_date

2. 调用函数传递可变参数

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

# 调用函数，传入任意数量的过滤参数
result_df = query_bigquery_with_filters(
    client=client,
    sql_template=sql,
    country_cd="CN",
    fcast_item="REVENUE",
    start_date=bigquery.Date.fromisoformat("2024-01-01")
)

print(result_df.head())

关键逻辑说明 #

• **kwargs 的作用：用来接收任意数量的键值对参数，参数的键名必须和SQL模板中的@占位符完全一致（大小写敏感）。比如SQL里的@country_cd对应调用时的country_cd="CN"。
• 动态生成参数：遍历kwargs中的每一组键值，根据Python值的类型自动映射到BigQuery支持的数据类型，无需手动逐个创建ScalarQueryParameter，新增过滤条件时只需在SQL里加对应占位符，调用时传参数即可，完全不用修改函数。
• 类型扩展：如果需要支持更多BigQuery数据类型（比如ARRAY、STRUCT），可以在类型判断分支里添加对应的映射规则。

可选优化：支持可选过滤条件 #

如果需要部分参数可选（比如用户可能不传某个过滤条件），可以修改SQL模板和函数逻辑：

SELECT * FROM `your-project.your-dataset.your-table`
WHERE 
    (@country_cd IS NULL OR country_cd = @country_cd)
    AND (@fcast_item IS NULL OR fcast_item = @fcast_item)

调用时如果不需要某个过滤条件，传None即可：

result_df = query_bigquery_with_filters(
    client=client,
    sql_template=sql,
    country_cd="US",
    fcast_item=None
)

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