如何使用PyPika从JSON API负载动态构建支持函数的SELECT查询

你当前的实现会把所有列字符串直接和表名拼接，导致函数表达式被错误解析成表的列（比如fn.Sum(revenue)变成customers.fn.Sum(revenue)）。要解决这个问题，需要对每个列字符串做解析，区分普通列和函数表达式，再调用PyPika对应的API生成正确的SQL片段。

解决方案代码 #

from pypika import Table, Query, functions as fn
import re

def parse_column(table, col_str):
    # 匹配 fn.XXX(column) 格式的函数表达式
    func_pattern = re.compile(r'fn\.(\w+)\((\w+)\)')
    match_result = func_pattern.match(col_str)
    
    if match_result:
        func_name = match_result.group(1)
        target_col = match_result.group(2)
        # 从PyPika的functions模块获取对应函数
        pypika_function = getattr(fn, func_name, None)
        if pypika_function:
            return pypika_function(getattr(table, target_col))
    
    # 处理普通列名
    return getattr(table, col_str)

def generate_sql(tablename, collist):
    table = Table(tablename)
    # 批量解析所有列
    parsed_columns = [parse_column(table, col) for col in collist]
    query = Query.from_(table).select(*parsed_columns)
    return query.get_sql(quote_char=None)

# 测试用例
tablename = 'customers'
collist = ['id', 'fname', 'fn.Sum(revenue)']
print(generate_sql(tablename, collist))

输出结果 #

SELECT id,fname,SUM(revenue) FROM customers

扩展说明 #

如果需要支持更复杂的表达式（比如带别名的函数fn.Sum(revenue) as total_revenue、分析函数fn.RowNumber().over(partition_by=id)），可以进一步扩展parse_column函数：

• 增加别名解析逻辑，用AS分割字符串提取别名，再通过PyPika的alias()方法设置
• 针对分析函数，编写更复杂的正则或字符串处理逻辑，解析over()内的参数并调用PyPika对应的over()方法

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