如何从sql查询中提取表名和列名？

Posted 2023-02-23

【中文标题】如何从sql查询中提取表名和列名？

【英文标题】：How to extract table names and column names from sql query?

【发布时间】：2016-06-08 01:43:04

【问题描述】：

假设我们有这样一个简单的查询：

Select a.col1, b.col2 from tb1 as a inner join tb2 as b on tb1.col7 = tb2.col8;

结果应该是这样的：

tb1 col1
tb1 col7
tb2 col2
tb2 col8

我尝试使用一些 python 库来解决这个问题：

1) 即使只使用sqlparse 提取表也可能是个大问题。比如this官方书根本就不能正常使用。

2) 使用正则表达式似乎真的很难实现。

3) 但后来我找到了 this ，这可能会有所帮助。但是问题是我无法连接到任何数据库并执行该查询。

有什么想法吗？

【问题讨论】：

谷歌搜索“python sql parser”会出现这个问题：***.com/questions/1394998/parsing-sql-with-python

【参考方案1】：

真的，这不是一件容易的事。您可以使用词法分析器（在此示例中为ply）并定义多个规则以从字符串中获取多个标记。以下代码为 SQL 字符串的不同部分定义了这些规则，并将它们重新组合在一起，因为输入字符串中可能存在别名。结果，您会得到一个字典 (result)，其中不同的表名作为键。

import ply.lex as lex, re

tokens = (
    "TABLE",
    "JOIN",
    "COLUMN",
    "TRASH"
)

tables = "tables": , "alias": 
columns = []

t_TRASH = r"Select|on|=|;|\s+|,|\t|\r"

def t_TABLE(t):
    r"from\s(\w+)\sas\s(\w+)"

    regex = re.compile(t_TABLE.__doc__)
    m = regex.search(t.value)
    if m is not None:
        tbl = m.group(1)
        alias = m.group(2)
        tables["tables"][tbl] = ""
        tables["alias"][alias] = tbl

    return t

def t_JOIN(t):
    r"inner\s+join\s+(\w+)\s+as\s+(\w+)"

    regex = re.compile(t_JOIN.__doc__)
    m = regex.search(t.value)
    if m is not None:
        tbl = m.group(1)
        alias = m.group(2)
        tables["tables"][tbl] = ""
        tables["alias"][alias] = tbl
    return t

def t_COLUMN(t):
    r"(\w+\.\w+)"

    regex = re.compile(t_COLUMN.__doc__)
    m = regex.search(t.value)
    if m is not None:
        t.value = m.group(1)
        columns.append(t.value)
    return t

def t_error(t):
    raise TypeError("Unknown text '%s'" % (t.value,))
    t.lexer.skip(len(t.value))

# here is where the magic starts
def mylex(inp):
    lexer = lex.lex()
    lexer.input(inp)

    for token in lexer:
        pass

    result = 
    for col in columns:
        tbl, c = col.split('.')
        if tbl in tables["alias"].keys():
            key = tables["alias"][tbl]
        else:
            key = tbl

        if key in result:
            result[key].append(c)
        else:
            result[key] = list()
            result[key].append(c)

    print result
    # 'tb1': ['col1', 'col7'], 'tb2': ['col2', 'col8']    

string = "Select a.col1, b.col2 from tb1 as a inner join tb2 as b on tb1.col7 = tb2.col8;"
mylex(string)

【讨论】：

你说得对，这个任务真的很难。但它似乎有点流行，我相信它已经解决了。

我得到这个错误：TypeError: main'> is a built-in module (PS. 我不是 Python 专家)

应该警告该脚本需要作为文件运行，并且除非您有非常基本的 SQL 字符串要解析，否则它也不起作用。

【参考方案2】：

我正在解决一个类似的问题，并找到了一个更简单的解决方案，而且似乎效果很好。

import re

def tables_in_query(sql_str):

    # remove the /* */ comments
    q = re.sub(r"/\*[^*]*\*+(?:[^*/][^*]*\*+)*/", "", sql_str)

    # remove whole line -- and # comments
    lines = [line for line in q.splitlines() if not re.match("^\s*(--|#)", line)]

    # remove trailing -- and # comments
    q = " ".join([re.split("--|#", line)[0] for line in lines])

    # split on blanks, parens and semicolons
    tokens = re.split(r"[\s)(;]+", q)

    # scan the tokens. if we see a FROM or JOIN, we set the get_next
    # flag, and grab the next one (unless it's SELECT).

    tables = set()
    get_next = False
    for tok in tokens:
        if get_next:
            if tok.lower() not in ["", "select"]:
                tables.add(tok)
            get_next = False
        get_next = tok.lower() in ["from", "join"]

    dictTables = dict()
    for table in tables:
        fields = []
        for token in tokens:
            if token.startswith(table):
                if token != table:
                    fields.append(token)
        if len(list(set(fields))) >= 1:
        dictTables[table] = list(set(fields))
    return dictTables

代码改编自https://grisha.org/blog/2016/11/14/table-names-from-sql/

【讨论】：

网址中的代码运行顺利，而您的代码需要一些调试

【参考方案3】：

sql-metadata 是一个 Python 库，它使用由 python-sqlparse 返回的标记化查询并生成查询元数据。

此元数据可以从您提供的 SQL 查询中返回列名和表名。以下是sql-metadata github 自述文件中的几个示例：

>>> sql_metadata.get_query_columns("SELECT test, id FROM foo, bar")
[u'test', u'id']

>>> sql_metadata.get_query_tables("SELECT test, id FROM foo, bar")
[u'foo', u'bar']

>>> sql_metadata.get_query_limit_and_offset('SELECT foo_limit FROM bar_offset LIMIT 50 OFFSET 1000')
(50, 1000)

库的托管版本存在于 sql-app.infocruncher.com 以查看它是否适合您。

【讨论】：

这个包目前还不是很完美，但是很接近。不过，它正在积极研究中，并且是我能找到的最佳解决方案..

在我这边，不幸的是，它不支持我们的“CREATE VIEW AS SELECT”嵌套视图/加入...

@tdebroc 您能否在 sql-metadata github 存储库中为您的案例创建问题？

@tdebroc 太棒了 - 看起来 sql-metadata 作者正在研究它 :) 如果我有时间我也会检查一下。

这应该是选择的答案，sql_metadata 模块很棒。感谢您贡献这个@DylanHogg

【参考方案4】：

不幸的是，为了成功地为“复杂的 SQL”查询执行此操作，您或多或少必须为您正在使用的特定数据库引擎实现一个完整的解析器。

例如，考虑这个非常基本的复杂查询：

WITH a AS (
    SELECT col1 AS c FROM b
)
SELECT c FROM a

在这种情况下，a 不是表，而是公用表表达式 (CTE)，应从输出中排除。没有简单的方法可以使用 regexp:es 来实现 b 是表访问，但 a 不是 - 您的代码真的必须更深入地理解 SQL。

同时考虑

SELECT * FROM tbl

您必须知道特定数据库实例中实际存在的列名（并且特定用户也可以访问）才能正确回答。

如果“使用复杂的 SQL”是指它必须使用任何有效的 SQL 语句，你还需要指定 哪个 SQL 方言 - 或实现方言- 特定的解决方案。一种适用于由未实现 CTE:s 的数据库处理的任何 SQL 的解决方案在实现的数据库中不起作用。

我很抱歉这么说，但我认为您不会找到适用于任意复杂 SQL 查询的完整解决方案。您必须找到一种适用于特定 SQL 方言子集的解决方案。

【参考方案5】：

创建数据库中存在的所有表的列表。然后，您可以在查询中搜索每个表名。 这显然不是万无一失的，如果任何列/别名与表名匹配，代码就会中断。 但它可以作为一种解决方法来完成。

【讨论】：

问题是提取表名和列。请查看 Jan 的回复。

欢迎来到 Stack Overflow。在回答已接受答案的旧问题（寻找绿色✓）以及其他答案之前，请确保您的答案添加了新内容或对它们有帮助。这是How to Answer 的指南。

【参考方案6】：

moz-sql-parser 是一个 python 库，用于将 SQL-92 查询的某些子集转换为 JSON 可解析树。也许它是你想要的。

这是一个例子。

>>> parse("SELECT id,name FROM dual WHERE id>3 and id<10 ORDER BY name")
'select': ['value': 'id', 'value': 'name'], 'from': 'dual', 'where': 'and': ['gt': ['id', 3], 'lt': ['id', 10]], 'orderby': 'value': 'name'

【参考方案7】：

import pandas as pd
#%config PPMagics.autolimit=0


#txt = """<your SQL text here>"""
txt_1 = txt
replace_list = ['\n', '(', ')', '*', '=','-',';','/','.']
count = 0
for i in replace_list:
    txt_1 = txt_1.replace(i, ' ')
txt_1 = txt_1.split()
res = []
for i in range(1, len(txt_1)):
    if txt_1[i-1].lower() in ['from', 'join','table'] and txt_1[i].lower() != 'select': 
        count +=1
        str_count = str(count)
        res.append(txt_1[i] + "." + txt_1[i+1])


#df.head()
res_l = res
f_res_l = []
for i in range(0,len(res_l)):
    if len(res_l[i]) > 15 : # change it to 0 is you want all the caught strings
        f_res_l.append(res_l[i])
    else :
        pass

All_Table_List = f_res_l
print("All the unique tables from the SQL text, in the order of their appearence in the code : \n",100*'*')
df = pd.DataFrame(All_Table_List,columns=['Tables_Names'])
df.reset_index(level=0, inplace=True)
list_=list(df["Tables_Names"].unique())
df_1_Final = pd.DataFrame(list_,columns=['Tables_Names'])
df_1_Final.reset_index(level=0, inplace=True)
df_1_Final

【讨论】：

只有在答案中没有任何解释的代码是没有用的，特别是对于已经接受答案的非常古老的问题并且也是一个赏金问题。解释为什么你的答案比其他人好。

【参考方案8】：

对于我的简单用例（查询中的一个表，没有连接），我使用了以下调整

lst = "select * from table".split(" ")
lst = [item for item in lst if len(item)>0]
table_name = lst[lst.index("from")+1]