使用Python cx_Oracle将CSV/字典数据插入Oracle数据库的问题咨询及优化方案探讨

我来帮你逐个解决这些问题，同时给出更优的实现方案：

问题1：DbConnect实例调用后后续方法未执行 #

你当前代码的核心问题是没有真正调用方法，以及存在一些语法/逻辑错误：

1. 实例化DbConnect时，__init__方法会自动执行，不需要手动调用atp_conn.__init__
2. atp_conn.get_connection和atp_conn.select_from_db只是引用方法对象，没有加括号()，导致方法根本没执行
3. 查询操作不需要connection.commit()，commit仅针对插入/更新/删除等DML操作；另外SQL里的字符串应该用单引号（Oracle中双引号是标识符）

修正后的main代码：

if __name__ == "__main__":
    config_file_parser(config_file)
    atp_conn = DbConnect(db_username, db_password, db_type)
    # 直接调用查询方法（不需要手动调用__init__和get_connection）
    result = atp_conn.select_from_db()
    print(result)

同时修正select_from_db方法，补充连接释放逻辑：

def select_from_db(self):
    # 修正SQL的引号问题，添加连接释放逻辑
    sql = 'SELECT * FROM TT.PMP WHERE TE = \'sall\''
    connection = self.get_connection()
    try:
        with connection.cursor() as cursor:
            cursor.execute(sql)
            # 如果需要返回查询结果，可添加：rows = cursor.fetchall()
        return "executed"
    finally:
        # 将连接释放回连接池，避免资源耗尽
        self.pool.release(connection)

另外，把cx_Oracle.init_oracle_client(lib_dir="./instantclient_19_8")移到DbConnect类定义的外面，因为这个初始化只需要执行一次，放在类内会重复执行导致报错。

问题2：实现CSV数据插入Oracle的insert_csv_data方法 #

可以用csv.reader读取文件，结合executemany批量插入（比循环单条插入效率高很多）：

def insert_csv_data(self, csv_file_path='test.csv'):
    # 请根据你的实际表结构修改列名和表名
    insert_sql = """
        INSERT INTO ATP_DATA (REGION, ATP_NAME, SER, M_START, M_END)
        VALUES (:1, :2, :3, :4, :5)
    """
    connection = self.get_connection()
    try:
        with connection.cursor() as cursor:
            # 读取CSV文件，跳过表头
            with open(csv_file_path, 'r', encoding='utf-8') as f:
                reader = csv.reader(f)
                next(reader)  # 跳过第一行表头
                data_rows = [tuple(row) for row in reader]
                
                # 批量插入数据
                cursor.executemany(insert_sql, data_rows)
        connection.commit()
        return f"插入成功，共插入{len(data_rows)}条数据"
    except Exception as e:
        connection.rollback()
        raise e  # 抛出异常便于调试
    finally:
        self.pool.release(connection)

优化方案：直接使用字典数据插入（无需生成CSV） #

既然已经有atp_dict_data字典列表，完全可以跳过生成CSV的IO步骤，直接批量插入，效率更高。

步骤1：调整数据获取逻辑（避免全局变量） #

修改get_work_request和config_file_parser，让它们返回数据而不是依赖全局变量：

def get_work_request(file=DEFAULT_CONFIG, profile_name=DEFAULT_PROFILE):
    oci_config = oci.config.from_file(file, profile_name=profile_name)
    db_client= oci.database.DatabaseClient(oci_config)
    atp_db_details = db_client.list_autonomous_databases(oci_config["compartment"]).data
    json_response = json.loads(str(atp_db_details))
    
    atp_dict_data = []
    for i in json_response:
        atp_dict_data.append({
            'region': oci_config["region"], 
            'atp_name': i["db_name"], 
            'ser': oci_config["ser"], 
            'm_start': i["m_start"], 
            'm_end': i["m_end"]
        })
    # 可选：如果仍需要生成CSV，保留这段代码，否则可以删除
    keys = atp_dict_data[0].keys()
    with open('test.csv', 'w') as output_file:
        dict_writer = csv.DictWriter(output_file, keys)
        dict_writer.writeheader()
        dict_writer.writerows(atp_dict_data)
    
    return atp_dict_data

def config_file_parser(config_file):
    atp_dict_data = []
    config = configparser.ConfigParser()
    config.read(config_file)
    for config_profile in config.sections():
        data = get_work_request(file=config_file, profile_name=config_profile)
        atp_dict_data.extend(data)  # 合并多个Profile的数据
    return atp_dict_data

步骤2：添加直接插入字典数据的方法 #

在DbConnect类中新增insert_dict_data方法：

def insert_dict_data(self, data_list):
    if not data_list:
        return "没有数据需要插入"
    
    # 自动生成插入SQL（基于字典的key）
    keys = data_list[0].keys()
    column_str = ', '.join(keys)
    bind_str = ', '.join([f':{key}' for key in keys])
    insert_sql = f"INSERT INTO ATP_DATA ({column_str}) VALUES ({bind_str})"
    
    connection = self.get_connection()
    try:
        with connection.cursor() as cursor:
            # 使用命名绑定变量，直接传入字典列表即可
            cursor.executemany(insert_sql, data_list)
        connection.commit()
        return f"插入成功，共插入{len(data_list)}条数据"
    except Exception as e:
        connection.rollback()
        raise e
    finally:
        self.pool.release(connection)

步骤3：在main中调用优化后的方法 #

if __name__ == "__main__":
    # 获取字典数据
    atp_dict_data = config_file_parser(config_file)
    # 初始化数据库连接
    atp_conn = DbConnect(db_username, db_password, db_type)
    # 直接插入字典数据
    insert_result = atp_conn.insert_dict_data(atp_dict_data)
    print(insert_result)

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