BigQuery使用配置中查询模板时未动态填充查询结果问题

读取配置文件中定义的动态查询字符串模板，结合BigQuery查询结果完成SQL格式化时，模板始终以静态文本形式输出，无法动态填充数据库返回的实际字段值。

def bq_exec_sql(sql):
    client = bigquery.Client(project='development')
    return client.query(sql)

def generate_select(sql, templete_select):
    job = bq_exec_sql(sql)
    result = ''
    print(templete_select)
    try:
        for row in job:
            result += templete_select
            
        print(result)
                
if __name__ == '__main__':
    for source in dashboard_activity_list:
       sql = config.get(source).source_select  # 读取自配置文件
       templete_select = config.get(source).select_template   # 读取自配置文件
       generate_select(sql, templete_select) 

select '"+row['table_id']+"' as table_id, '"+row['frequency']+"' as frequency from `trend-dev.test.table1`
 select '"+row['table_id']+"' as table_id, '"+row['info']+"' as info from `trend-dev.test.table2`

select table_name1 as table_id, daily from `trend-dev.test.table1`
 select table_name2 as table_id, chart_data from `trend-dev.test.table2`

dashboard_activity_list: [source_partner_trend, source_partner_normal]
source_partner_trend: 
    source_select : select * from `trend-dev.test.trend_partner` 
    source_key : trend_source_partner
    distination_table : test.trend_partner_dashboard_feed
    select_template : select '"+row['table_id']+"' as table_id, '"+row['frequency']+"' as frequency from `trend-dev.test.table1`
source_partner_normal: 
    source_select : select * from `trend-dev.test.normal_partner` 
    source_key : normal_source_partner
    distination_table : test.normal_partner_dashboard_feed
    select_template : select '"+row['table_id']+"' as table_id, '"+row['info']+"' as info from `trend-dev.test.table2`

遍历BigQuery返回的row结果集时，未对配置读取的select_template模板做变量解析替换，直接拼接静态模板字符串，导致输出内容未携带实际查询到的字段值。配置中直接写Python拼接表达式的写法本身也无法被自动识别执行。

• 第一步：修改配置文件中的模板规则，替换原有的Python拼接表达式为Python标准的占位符格式，避免在配置中硬编码代码逻辑，同时方便格式化替换。修改后的模板示例如下：

  # source_partner_trend 对应模板
  select_template : select '{table_id}' as table_id, '{frequency}' as frequency from `trend-dev.test.table1`
  # source_partner_normal 对应模板
  select_template : select '{table_id}' as table_id, '{info}' as info from `trend-dev.test.table2`
  

• 第二步：调整结果遍历逻辑，每遍历一行查询结果，就用当前行的字段值替换模板中的占位符，再拼接到最终结果中。修复后的核心函数代码如下：

  def generate_select(sql, templete_select):
      job = bq_exec_sql(sql)
      result = ''
      try:
          for row in job:
              # 将行对象转为字典，匹配占位符完成替换，拼接时加换行分隔多条SQL
              result += templete_select.format_map(dict(row)) + "\n"
      print(result)
  

注意：如果查询返回的字段值包含单引号，需要额外做转义处理，避免生成的SQL出现语法错误。

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