Python读取点分隔schema配置生成SQL并分文件存储查询结果问题

你现有函数仅返回单条映射的核心原因是构建schema-表字典时，对同一个schema键直接赋值覆盖了之前的表记录，改用列表存储同schema下的所有表即可解决，完整实现逻辑如下：

1. 生成全量schema与表映射集合 #

读取配置文件，按schema分组存储对应表名：

def get_schema_table_map(config_path: str) -> dict:
    schema_map = {}
    with open(config_path, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            # 跳过空行和不符合格式的行
            if not line or '.' not in line:
                continue
            # 仅分割第一个点，兼容表名带点的特殊场景
            schema, table = line.split('.', 1)
            if schema not in schema_map:
                schema_map[schema] = []
            schema_map[schema].append(table)
    return schema_map

示例输入对应的返回结果为：{'sch1': ['ft1', 'ft2'], 'sch2': ['ft3', 'ft4']}

2. 生成syscat.columns查询SQL #

针对单张表生成字段查询语句，可按需调整返回的字段列表：

def gen_column_query_sql(schema: str, table: str) -> str:
    return f"""
SELECT colname, typename, length, nulls, default
FROM syscat.columns
WHERE tabschema = UPPER('{schema}')
  AND tabname = UPPER('{table}')
ORDER BY colno ASC;
    """

3. 执行SQL并导出对应文件 #

以下为DB2数据库的Python实现示例，使用其他数据库可替换为对应驱动：

# 需提前安装ibm_db驱动：pip install ibm_db
import ibm_db

def export_all_table_columns(schema_map: dict, db_conn_info: dict):
    # 建立数据库连接
    conn = ibm_db.connect(
        f"DATABASE={db_conn_info['db_name']};HOSTNAME={db_conn_info['host']};PORT={db_conn_info['port']};"
        f"PROTOCOL=TCPIP;UID={db_conn_info['user']};PWD={db_conn_info['password']};",
        "", ""
    )
    try:
        for schema, tables in schema_map.items():
            for table in tables:
                sql = gen_column_query_sql(schema, table)
                stmt = ibm_db.exec_immediate(conn, sql)
                # 构造输出文件名
                output_path = f"./{schema}_{table}.txt"
                with open(output_path, 'w', encoding='utf-8') as f:
                    # 写入表头
                    f.write("字段名\t字段类型\t长度\t允许空值\t默认值\n")
                    # 写入查询结果
                    row = ibm_db.fetch_tuple(stmt)
                    while row:
                        f.write("\t".join([str(col) for col in row]) + "\n")
                        row = ibm_db.fetch_tuple(stmt)
    finally:
        ibm_db.close(conn)

# 调用示例
if __name__ == "__main__":
    # 替换为你的配置文件路径
    schema_table_map = get_schema_table_map("./table_config.conf")
    # 替换为你的数据库连接信息
    db_info = {
        "db_name": "你的数据库名",
        "host": "数据库地址",
        "port": 50000,
        "user": "用户名",
        "password": "密码"
    }
    export_all_table_columns(schema_table_map, db_info)

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