如何用Python一次性提取MS Access多个数据表的数据？

场景1：多表结构完全一致 #

如果你的4个数据表结构相同（包含要查询的统计列和条件列），可以用SQL的UNION ALL语法合并多个表的查询结果，一次获取所有数据。

修改代码片段： #

替换原代码中num1 == 1分支的内容：

if num1 == 1:
    # 输入多个表名，用逗号分隔（例如：rawData1,rawData2,rawData3,rawData4）
    table_names = input("Enter the names of the tables (separated by commas): ").split(',')
    # 去除表名前后的空格
    table_names = [tbl.strip() for tbl in table_names]
    column_to_count = input("Enter the name of the column to count the variables in: ")
    condition_column = input("Enter the name of the column to use as a condition: ")
    condition_value = input("Enter the value to use as a condition: ")
    pd.options.display.max_rows = None

    # 生成每个表的查询语句，用UNION ALL连接
    table_queries = []
    for tbl in table_names:
        single_query = f"SELECT {column_to_count}, {condition_column} FROM {tbl} WHERE {condition_column} = '{condition_value}'"
        table_queries.append(single_query)
    # 合并所有查询语句
    query = " UNION ALL ".join(table_queries)
    df = pd.read_sql(query, conn)

场景2：多表结构不同 #

如果4个表的列结构不一致，无法直接用UNION ALL合并，可以分别查询每个表，再把结果合并成一个DataFrame。

修改代码片段： #

同样替换num1 == 1分支的内容：

if num1 == 1:
    table_names = input("Enter the names of the tables (separated by commas): ").split(',')
    table_names = [tbl.strip() for tbl in table_names]
    column_to_count = input("Enter the name of the column to count the variables in: ")
    condition_column = input("Enter the name of the column to use as a condition: ")
    condition_value = input("Enter the value to use as a condition: ")
    pd.options.display.max_rows = None

    # 存储每个表的查询结果
    df_list = []
    for tbl in table_names:
        try:
            query = f"SELECT {column_to_count}, {condition_column} FROM {tbl} WHERE {condition_column} = '{condition_value}'"
            single_df = pd.read_sql(query, conn)
            # 可选：添加一列标记数据来源表，方便后续区分
            single_df['source_table'] = tbl
            df_list.append(single_df)
        except Exception as e:
            print(f"Query failed for table {tbl}: {str(e)}")
    # 合并所有表的结果
    df = pd.concat(df_list, ignore_index=True)

额外优化建议 #

• 避免SQL注入：当前用字符串拼接SQL存在注入风险，生产环境建议对条件值使用参数化查询：

  # 条件值参数化写法
  query = f"SELECT {column_to_count}, {condition_column} FROM {tbl} WHERE {condition_column} = ?"
  single_df = pd.read_sql(query, conn, params=(condition_value,))
  

• 验证表名合法性：可以先获取数据库中所有表名，验证输入的表是否存在：

  cursor = conn.cursor()
  # 获取数据库内所有数据表名
  existing_tables = [table.table_name for table in cursor.tables(tableType='TABLE')]
  # 检查输入的表名是否存在
  for tbl in table_names:
      if tbl not in existing_tables:
          print(f"Warning: Table {tbl} does not exist!")
  

内容的提问来源于stack exchange，提问作者Азизбек Нуриддинов