如何用Python精准提取SQLite查询中的表与列（含SELECT*场景）

针对sql_metadata精度不足的问题，结合SQLite数据库元数据+SQL语法解析的方案可以满足需求，既能处理常规列提取，也能支持SELECT*时获取对应表的所有字段，最终输出table.column格式。

核心思路 #

1. 从SQLite系统表获取表结构：通过查询sqlite_master和PRAGMA table_info获取数据库中所有表的字段信息，为SELECT*场景提供字段映射。
2. 用SQL语法解析器提取SQL元素：借助sqlglot（比sql_metadata更精准的解析库）解析SQL，识别出查询中涉及的表（含别名）、列，以及SELECT*的位置。
3. 关联表与列：将解析得到的列与对应的表关联，处理表别名、列别名，最终输出规范格式的结果。

实现代码 #

1. 先安装依赖 #

pip install sqlglot

2. 获取SQLite表结构 #

这个函数会读取数据库文件，返回所有表及其字段的字典映射：

import sqlite3

def get_table_columns(db_path):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    
    # 获取所有用户表名（排除SQLite系统表）
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%';")
    tables = [row[0] for row in cursor.fetchall()]
    
    table_columns = {}
    for table in tables:
        # 获取表的字段信息
        cursor.execute(f"PRAGMA table_info({table});")
        columns = [row[1] for row in cursor.fetchall()]
        table_columns[table.lower()] = [col.lower() for col in columns]
    
    conn.close()
    return table_columns

3. 解析SQL并提取表列 #

这个函数负责解析SQL，结合表结构数据，输出最终的table.column结果：

import sqlglot

def extract_table_columns(sql, table_columns):
    try:
        parsed = sqlglot.parse_one(sql)
    except Exception as e:
        print(f"SQL解析失败：{e}")
        return []
    
    # 建立表别名与真实表名的映射（统一小写）
    table_aliases = {}
    for table_expr in parsed.find_all(sqlglot.exp.Table):
        table_name = table_expr.name.lower()
        # 处理表别名，比如 `FROM student_course_attendance sca` 或 `FROM student_course_attendance AS sca`
        alias = table_expr.alias.lower() if table_expr.alias else table_name
        table_aliases[alias] = table_name
    
    used_table_columns = set()
    
    # 遍历所有涉及列的SQL节点：SELECT、WHERE、ORDER BY等
    for col_expr in parsed.find_all(sqlglot.exp.Column):
        col_name = col_expr.name.lower()
        # 获取列所属的表别名（如果有）
        table_alias = col_expr.args.get("this").name.lower() if col_expr.args.get("this") else None
        
        if table_alias:
            # 有表别名，直接映射到真实表名
            real_table = table_aliases.get(table_alias)
            if real_table and col_name in table_columns.get(real_table, []):
                used_table_columns.add(f"{real_table}.{col_name}")
        else:
            # 无表别名，匹配所有包含该列的表（假设列名唯一，可根据需求调整）
            for table, cols in table_columns.items():
                if col_name in cols:
                    used_table_columns.add(f"{table}.{col_name}")
                    break
    
    # 处理SELECT*的情况
    for star_expr in parsed.find_all(sqlglot.exp.Star):
        target_alias = star_expr.args.get("this").name.lower() if star_expr.args.get("this") else None
        
        if target_alias:
            # 比如 `sca.*`，只取对应表的所有列
            real_table = table_aliases.get(target_alias)
            if real_table:
                for col in table_columns.get(real_table, []):
                    used_table_columns.add(f"{real_table}.{col}")
        else:
            # 比如 `SELECT*`，取所有查询表的列
            for alias, real_table in table_aliases.items():
                for col in table_columns.get(real_table, []):
                    used_table_columns.add(f"{real_table}.{col}")
    
    # 去重并排序输出
    return sorted(used_table_columns)

4. 测试示例 #

if __name__ == "__main__":
    # 替换为你的SQLite数据库路径
    DB_PATH = "your_database.db"
    table_cols = get_table_columns(DB_PATH)
    
    test_sql = """
    SELECT student_id FROM student_course_attendance 
    WHERE course_id = 301 
    ORDER BY date_of_attendance DESC LIMIT 1
    """
    
    result = extract_table_columns(test_sql, table_cols)
    print(result)
    # 输出：
    # ['student_course_attendance.course_id', 'student_course_attendance.date_of_attendance', 'student_course_attendance.student_id']

注意事项 #

• 列名重复处理：如果多个表存在同名列，当前代码会取第一个匹配的表。若需更精准匹配，可以结合SQL中FROM子句的表列表来筛选。
• 复杂SQL支持：代码已覆盖SELECT、WHERE、ORDER BY场景，如需支持JOIN、子查询等复杂结构，可以扩展find_all的节点类型（如sqlglot.exp.Join）。
• 大小写兼容：代码统一将表名、列名转为小写，适配SQLite默认的大小写不敏感特性，若数据库为大小写敏感，可移除转小写逻辑。

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