使用来自 CSV 的输入从 Python 运行动态查询

Question

我有一个 CSV 文件，其中包含这些表的表名和主键，格式如下：

|表名 |主键 | |表 1 | Col1 | |表 1 | Col2 | |表 1 | Col3 | |表 2 | Col11 | |表 2 | Col12 |

我想运行一个 sql 查询来验证每个表的 PK 约束。执行此操作的查询如下所示：

select Col1, Col2, Col3 from Table1
group by Col1, Col2, Col3
having count(*)>1 

但我在这个文件中有数千个表。我将如何动态编写和执行此查询并将结果写入平面文件？ 我想使用 Python 3 执行此操作。

尝试：

CSV：

我的 PKTest.py

def getColumns(filename):
    tables = {}

    with open(filename) as f:
        for line in f:
            line = line.strip()
            if 'Primary Key' in line:
                continue

            cols = line.split('|')
            table = cols[1].strip()
            col = cols[2].strip()

            if table in tables:
                tables[table].append(col)
            else:
                tables[table] = [col]
    return tables

def runSQL(table, columns):
    statement = 'select {0} from {1} group by {0} having count(*) > 1'.format(', '.join(columns), table.replace(' ',''))
    return statement

if __name__ == '__main__':
    tables = getColumuns('PKTest.csv')
    try:
        #cursor to connect

        for table in tables:
            sql = runSQL(table,tables[table])
            print(sql)
            cursor.execute(sql)
            for result in cursor:
                print(result)

    finally:
        cursor.close()
    ctx.close()

Answer 1

由于我无法访问 Oracle，因此您将不得不对这个答案进行一些即兴创作。

假设有一个名为 so.csv 的文件包含您的问题中显示的数据。

像这样创建一个名为so.py 的文件。我将添加一些代码和一些解释。您可以将文件拼凑在一起或从此处复制/粘贴：https://rextester.com/JLQ73751。

在文件顶部，导入您的 Oracle 依赖项：

# import cx_Oracle
# https://www.oracle.com/technetwork/articles/dsl/python-091105.html

然后，创建一个函数来解析您的 so.csv 并将表和列放入字典中，如下所示：{'Table 1': ['Col1', 'Col2', 'Col3'], 'Table 2': ['Col11', 'Col12']}

def get_tables_columns(filename):

    tables = {}

    with open(filename) as f:
        for line in f:
            line = line.strip()
            if 'Primary Key' in line:
                continue

            cols = line.split('|')

            table = cols[1].strip()
            col = cols[2].strip()

            if table in tables:
                tables[table].append(col)
            else:
                tables[table] = [col]

    return tables

然后，如果它知道表和列列表，则创建一个生成sql的函数：

def get_sql(table, columns):

    statement = 'select {0} from {1} group by {0} having count(*) > 1'.format(
            ', '.join(columns),
            table.replace(' ', '')
        )

    return statement

是时候执行函数了：

if __name__ == '__main__':
    tables = get_tables_columns('so.csv')

    # here goes your code to connect with Oracle
    # con = cx_Oracle.connect('pythonhol/welcome@127.0.0.1/orcl')
    # cur = con.cursor()

    for table in tables:
        sql = get_sql(table, tables[table])
        print(sql)

        # here goes your sql statement execution            
        # cur.execute(sql)
        # for result in cur:
        #    print result

    # close your Oracle connection
    # con.close()

您可以包含您的 Oracle 相关语句并运行 python 文件。