如何用Python将SQL查询中的*扩展为指定列名？

一、可用的第三方库 #

没有完全一键实现需求的库，但可以借助SQL解析库简化开发：

• sqlparse：轻量SQL解析工具，能识别SELECT子句、表别名、通配符等结构，帮你定位需要替换的*或alias.*
• sqlglot：功能更强大的SQL解析/转译库，支持遍历AST（抽象语法树），精准定位所有通配符节点，方便替换

用sqlglot的核心思路：

1. 将SQL解析为AST结构
2. 遍历AST找到所有Wildcard节点（即*或alias.*）
3. 根据传入的表-列映射，替换节点为对应的列名列表（如my_table.id, my_table.column1）
4. 重新生成完整SQL

二、手动实现的要点 #

如果不想依赖第三方库，直接拼接表和列名是可行的，但必须处理几个关键场景：

• 表别名映射：比如my_table as m，要识别m.*对应my_table的列，替换为m.id, m.column1...
• 子查询递归处理：先展开内层子查询的*，再将子查询的列映射为subquery.column格式
• 多表列区分：多表关联时，不同表的同名列要分别标注（如my_table.id和other_table.id）

手动实现的核心步骤：

1. 解析SQL，提取表与别名的映射关系（比如从my_table as m中得到m → my_table）
2. 定位所有SELECT子句中的通配符
3. 根据别名找到对应表，拼接成带前缀的列名列表
4. 替换原SQL中的通配符，注意递归处理嵌套子查询

三、简化版实现代码 #

import sqlparse
from sqlparse.sql import IdentifierList
from sqlparse.tokens import Keyword, Wildcard

def expand_select_wildcards(sql, table_columns, alias_map):
    parsed = sqlparse.parse(sql)[0]
    
    # 遍历处理所有SELECT子句中的通配符
    for token in parsed.flatten():
        if token.ttype == Keyword and token.value.upper() == 'SELECT':
            current = token.next
            while current and (current.ttype != Keyword or current.value.upper() not in ('FROM', 'WHERE')):
                if isinstance(current, IdentifierList):
                    for item in current.get_identifiers():
                        if item.ttype == Wildcard:
                            # 获取通配符的别名（如m.*中的m）
                            alias = item.get_real_name()
                            # 找到别名对应的表名
                            table = alias_map.get(alias, alias) if alias else None
                            
                            # 生成需要替换的列列表
                            if table and table in table_columns:
                                columns = [f"{alias}.{col}" for col in table_columns[table]]
                            else:
                                # 无别名时拼接所有表的列
                                columns = []
                                for tbl, cols in table_columns.items():
                                    tbl_alias = alias_map.get(tbl, tbl)
                                    columns.extend([f"{tbl_alias}.{col}" for col in cols])
                            
                            # 替换通配符为列列表
                            item.replace(', '.join(columns))
                current = current.next
    
    return str(parsed)

# 测试示例
table_columns = {
    "my_table": ["id", "column1", "column2"],
    "other_table": ["id", "column3", "column4"]
}
# 手动指定别名映射，实际可从SQL中自动解析
alias_map = {"m": "my_table", "o": "other_table", "subquery": "subquery"}

# 先处理内层子查询
subquery = "SELECT * FROM my_table as m join other_table as o on o.id=m.id"
expanded_subquery = expand_select_wildcards(subquery, table_columns, alias_map)

# 处理外层查询，需要先提取子查询的列名作为subquery的列
sub_cols = [col.split('.')[1] for col in expanded_subquery.split('SELECT ')[1].split(' FROM ')[0].split(', ')]
outer_sql = f"SELECT * FROM ({expanded_subquery}) AS subquery"
expanded_result = expand_select_wildcards(outer_sql, {"subquery": sub_cols}, alias_map)

print(expanded_result)

注意：这个简化版仅处理基础场景，复杂SQL（如嵌套子查询、带函数的SELECT、多表JOIN的复杂条件等）需要更完善的AST遍历逻辑，推荐使用sqlglot来处理这类复杂情况。

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