如何用Python高效遍历SQL数据库表查找指定列

需求目标 #

访问包含数百张表的SQL Server数据库，遍历所有表的列，找出包含指定列名的表，并将这些表名存入列表。

初始准备：数据库连接与表名获取 #

已完成数据库访问配置，成功连接数据库并获取所有表名：

import pandas
import pyodbc

main_path = 'my_database_name'

with pyodbc.connect("DRIVER={SQL Server};" +
    "SERVER=MyServerName;" +
    f"DATABASE={main_path};" +
    "Trusted_Connection=yes;") as main_conn:

db_table_names = [_.table_name for _ in main_conn.cursor().tables()]

低效的初始实现 #

最初尝试遍历表名列表，读取整张表后检查目标列是否存在：

col_name ='whatever_col_i_need'
table_matches = [table_name if col_name in pandas.read_sql(f"SELECT * FROM {table_name }",
                con=main_conn).columns else None for table in db_tables]

问题 #

这种方法资源占用高、速度极慢——因为SELECT * FROM {table_name}会读取整张表的所有数据，面对数百张表时，不仅耗时久，还会占用大量内存和数据库连接资源。

错误的优化尝试 #

想用with语句优化资源管理，但存在格式错误：

found_tables = []

for table_name in db_table_names:
    with pandas.read_sql(f"SELECT * FROM {table_name}", con=main_conn) as temp_tbl:

    if 'LOCATION' in temp_tbl.columns:
        found_tables.append(table_name)

错误原因 #

1. with块后的代码没有缩进，导致语法错误；
2. pandas的read_sql返回的DataFrame对象不支持上下文管理（不是上下文管理器），无法用with语句包裹。

高效解决方案：仅查询指定列 #

无需读取全表数据，通过try/except尝试查询指定列，若查询成功则说明列存在，记录对应表名：

found_tables = []

for table_name in db_tables:
    try:
        # 打印进度，方便跟踪
        print(f'正在检查: {table_name} 中是否存在 {col_name}...')
        pandas.read_sql_query(f"SELECT {col_name} FROM {table_name}", con=main_conn)
        found_tables.append(table_name)
    except Exception as e:
        print(e)
        pass

原理 #

• 若表中存在目标列，SELECT {col_name} FROM {table_name}会执行成功（还可以加TOP 1进一步减少数据传输，比如SELECT TOP 1 {col_name} FROM {table_name}）；
• 若列不存在，SQL会抛出异常，被try/except捕获后直接跳过该表，无需处理无效数据。

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