使用copy_from()将DataFrame导入PostgreSQL时遭遇WinError 32文件占用问题求助

这个错误在Windows环境下特别常见，核心原因是你的临时文件在读写过程中没有被正确释放，或者文件句柄没有及时关闭导致系统锁定了文件。下面给你两种解决方案，强烈推荐第二种，既高效又能彻底规避文件占用问题：

方案1：修复文件句柄管理（确保文件被正确关闭） #

你的代码里打开文件后没有显式关闭句柄，而且在删除文件时，copy_from可能还在占用文件资源。用with语句可以自动管理文件句柄，不管代码执行成功还是抛出异常，都会自动关闭文件：

import os
import psycopg2

def copy_from_datafile(conn, df_s, table):
    """将给定的DataFrame保存为磁盘上的CSV文件，将CSV文件加载到缓冲区并使用copy_from()将其复制到PostgreSQL数据库表"""
    if table == "symbols_option":
        command= '''CREATE TEMPORARY TABLE IF NOT EXISTS _symbol (
        instrument_token INTEGER NOT NULL,
        exchange_token INTEGER,
        tradingsymbol TEXT NOT NULL,
        name TEXT,
        expiry TIMESTAMP,
        strike DOUBLE PRECISION,
        lot_size INTEGER,instrument_type TEXT,
        segment TEXT NOT NULL,
        exchange TEXT,
        PRIMARY KEY (instrument_token, tradingsymbol));'''
        query_main = '''INSERT INTO {table}(instrument_token, exchange_token, tradingsymbol, name, expiry, strike, lot_size,instrument_type, segment, exchange )
        SELECT * FROM _symbol ON CONFLICT (instrument_token, tradingsymbol) DO NOTHING '''.format(table=table)
        columns = ["instrument_token", "exchange_token", "tradingsymbol", "name", "expiry", "strike", "lot_size","instrument_type", "segment", "exchange" ]
    elif table == "symbols_future":
        command= '''CREATE TEMPORARY TABLE IF NOT EXISTS _symbol (
        instrument_token INTEGER NOT NULL,exchange_token INTEGER,
        tradingsymbol TEXT NOT NULL,name TEXT,
        expiry TIMESTAMP,lot_size INTEGER,instrument_type TEXT,
        segment TEXT NOT NULL,
        exchange TEXT,
        PRIMARY KEY (instrument_token, tradingsymbol));'''
        query_main = '''INSERT INTO {table}(instrument_token, exchange_token, tradingsymbol,name, expiry,lot_size,instrument_type, segment, exchange )
        SELECT * FROM _symbol ON CONFLICT (instrument_token, tradingsymbol) DO NOTHING '''.format(table=table)
        columns = [ "instrument_token", "exchange_token", "tradingsymbol", "name", "expiry", "lot_size", "instrument_type", "segment", "exchange"]
    else:
        print(f"Unknown table: {table}")
        return

    tmp_df = 'temp.csv'
    df_s.to_csv(tmp_df, header=False, index=False)
    
    cursor = conn.cursor()
    try:
        # 用with语句自动管理文件句柄
        with open(tmp_df, 'r') as f:
            cursor.execute(command)
            cursor.copy_from(f, '_symbol', sep=",", columns=columns)
            cursor.execute(query_main)
        print("Data inserted using copy_from_datafile() successfully....")
        conn.commit()
    except (Exception, psycopg2.DatabaseError) as err:
        print(err)
        conn.rollback()
    finally:
        # 不管成功失败，都尝试清理文件并关闭游标
        try:
            os.remove(tmp_df)
        except Exception as e:
            print(f"Failed to delete temp file: {e}")
        cursor.close()

方案2：完全避免临时文件（推荐） #

其实根本不需要生成磁盘上的临时文件，我们可以用io.StringIO把DataFrame转换成内存中的CSV缓冲区，直接传递给copy_from，彻底绕开文件IO和锁定问题：

import io
import psycopg2

def copy_from_datafile(conn, df_s, table):
    """将DataFrame通过内存缓冲区直接复制到PostgreSQL数据库表"""
    if table == "symbols_option":
        command= '''CREATE TEMPORARY TABLE IF NOT EXISTS _symbol (
        instrument_token INTEGER NOT NULL,
        exchange_token INTEGER,
        tradingsymbol TEXT NOT NULL,
        name TEXT,
        expiry TIMESTAMP,
        strike DOUBLE PRECISION,
        lot_size INTEGER,instrument_type TEXT,
        segment TEXT NOT NULL,
        exchange TEXT,
        PRIMARY KEY (instrument_token, tradingsymbol));'''
        query_main = '''INSERT INTO {table}(instrument_token, exchange_token, tradingsymbol, name, expiry, strike, lot_size,instrument_type, segment, exchange )
        SELECT * FROM _symbol ON CONFLICT (instrument_token, tradingsymbol) DO NOTHING '''.format(table=table)
        columns = ["instrument_token", "exchange_token", "tradingsymbol", "name", "expiry", "strike", "lot_size","instrument_type", "segment", "exchange" ]
    elif table == "symbols_future":
        command= '''CREATE TEMPORARY TABLE IF NOT EXISTS _symbol (
        instrument_token INTEGER NOT NULL,exchange_token INTEGER,
        tradingsymbol TEXT NOT NULL,name TEXT,
        expiry TIMESTAMP,lot_size INTEGER,instrument_type TEXT,
        segment TEXT NOT NULL,
        exchange TEXT,
        PRIMARY KEY (instrument_token, tradingsymbol));'''
        query_main = '''INSERT INTO {table}(instrument_token, exchange_token, tradingsymbol,name, expiry,lot_size,instrument_type, segment, exchange )
        SELECT * FROM _symbol ON CONFLICT (instrument_token, tradingsymbol) DO NOTHING '''.format(table=table)
        columns = [ "instrument_token", "exchange_token", "tradingsymbol", "name", "expiry", "lot_size", "instrument_type", "segment", "exchange"]
    else:
        print(f"Unknown table: {table}")
        return

    cursor = conn.cursor()
    try:
        # 创建内存缓冲区，写入DataFrame内容
        buffer = io.StringIO()
        df_s.to_csv(buffer, header=False, index=False)
        buffer.seek(0)  # 将指针移到缓冲区开头，方便读取
        
        cursor.execute(command)
        cursor.copy_from(buffer, '_symbol', sep=",", columns=columns)
        cursor.execute(query_main)
        
        print("Data inserted using copy_from_datafile() successfully....")
        conn.commit()
    except (Exception, psycopg2.DatabaseError) as err:
        print(err)
        conn.rollback()
    finally:
        cursor.close()
        buffer.close()

额外优化说明 #

我还在代码里做了这些细节调整：

• 用elif替代重复的if判断，减少逻辑冗余
• 提取了公共的columns变量，避免重复代码
• 添加事务回滚和提交逻辑，确保数据库状态一致
• 增加了未知表的处理分支，避免后续无意义报错

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