使用Python向PostgreSQL插入数据失败，求解决方案及代码优化

尝试编写Python脚本读取Excel文件并导入本地PostgreSQL的player_behaviour表，执行时出现错误（错误截图如下），怀疑是将所有字段值转为字符串导致数据类型不匹配，但不确定，需要问题解决方法及代码优化建议。

原Python代码 #

import psycopg2 as pg
from pandas import read_excel,DataFrame

def execute_query(connection,cursor,query:str):    
    cursor.execute(query)
    connection.commit()

def create_conn():
    try:
        connection = pg.connect("host='localhost' port='5432' dbname='abc' user='abc' password='abc'")
        cursor = connection.cursor()
        return connection,cursor
    except:
        print("Connection failed")

def read_sql_file(filename):
    with open(filename, 'r') as file:
        sql_queries = file.read()
    return sql_queries

def import_csv_data(connection, cursor, csv_file,tablename):
    try:
       df = read_excel(csv_file)

       for i,row in df.iterrows():
           values = ",".join(map(str,row.values))
           query = f"INSERT INTO {tablename} VALUES {values};"
           execute_query(connection,cursor,query)

    except Exception as e:
       print(f"Error: {e}")


def close_connection(connection,cursor):
    connection.close()
    cursor.close()


if __name__ == "__main__":    
    conn, curr = create_conn()
    if conn and curr:        
        
        ## Create table
        # sql_file = "player_behaviour_create.sql"  
        # query = read_sql_file(sql_file)                        
        # execute_query(conn,curr,query)

        ## Insert Data
        import_csv_data(conn,curr,r"Data\data_sample_100_rows.xlsx","player_behaviour")    

        close_connection(conn,curr)

建表SQL语句 #

CREATE TABLE player_behaviour (
    PlayerID INT PRIMARY KEY,
    Age INT,
    Gender VARCHAR(10),
    Location VARCHAR(50),
    GameID INT,
    PlayTime FLOAT,
    FavoriteGame VARCHAR(50),
    SessionID BIGINT,
    CampaignID INT,
    AdsSeen INT,
    PurchasesMade INT,
    EngagementLevel VARCHAR(10)
);

你的怀疑是正确的：错误根源在于直接将所有字段转为字符串拼接SQL语句，导致字符串类型的值未加引号，PostgreSQL无法识别；同时数字类型也可能因格式问题触发类型不匹配错误。正确的解决方式是使用参数化查询，让psycopg2自动处理数据类型转换：

修改import_csv_data函数如下：

def import_csv_data(connection, cursor, excel_file, tablename):
    try:
        df = read_excel(excel_file)
        # 生成与列数匹配的参数占位符
        placeholders = ", ".join(["%s"] * len(df.columns))
        insert_query = f"INSERT INTO {tablename} VALUES ({placeholders});"
        
        # 将DataFrame数据转为元组列表，批量插入
        data_rows = [tuple(row) for _, row in df.iterrows()]
        cursor.executemany(insert_query, data_rows)
        connection.commit()
        print(f"成功导入{len(data_rows)}条数据")
    except Exception as e:
        connection.rollback()  # 出错时回滚事务，避免数据不一致
        print(f"错误信息：{e}")

关键改进点 #

• 使用%s作为PostgreSQL的参数占位符，psycopg2会自动根据数据类型添加引号或格式转换
• 用executemany批量插入，比循环单条插入效率提升数倍
• 增加事务回滚操作，避免部分插入成功导致的数据异常

1. 连接管理优化：用上下文管理器自动释放资源 #

避免手动关闭连接/游标时的遗漏，使用with语句自动管理资源：

def create_conn():
    try:
        return pg.connect("host='localhost' port='5432' dbname='abc' user='abc' password='abc'")
    except pg.Error as e:
        print(f"连接失败：{e}")
        return None

# 主函数简化
if __name__ == "__main__":    
    conn = create_conn()
    if conn:        
        with conn.cursor() as curr:
            import_csv_data(conn, curr, r"Data\data_sample_100_rows.xlsx", "player_behaviour")    
        conn.close()

2. 数据校验：提前匹配表结构与数据列 #

在插入前校验DataFrame列数与表列数是否一致，避免列不匹配错误：

def import_csv_data(connection, cursor, excel_file, tablename):
    try:
        df = read_excel(excel_file)
        # 查询目标表的字段数量
        cursor.execute(f"SELECT COUNT(*) FROM information_schema.columns WHERE table_name = '{tablename}';")
        table_col_count = cursor.fetchone()[0]
        if len(df.columns) != table_col_count:
            raise ValueError(f"数据列数({len(df.columns)})与表列数({table_col_count})不匹配")
        
        placeholders = ", ".join(["%s"] * len(df.columns))
        insert_query = f"INSERT INTO {tablename} VALUES ({placeholders});"
        data_rows = [tuple(row) for _, row in df.iterrows()]
        cursor.executemany(insert_query, data_rows)
        connection.commit()
        print(f"成功导入{len(data_rows)}条数据")
    except pg.Error as e:
        connection.rollback()
        print(f"PostgreSQL错误：{e}")
    except ValueError as e:
        print(f"数据校验错误：{e}")
    except Exception as e:
        print(f"其他错误：{e}")

3. 极简方案：使用pandas内置to_sql #

直接用pandas的to_sql方法，无需手动编写插入逻辑，自动处理类型映射：

from pandas import read_excel
from sqlalchemy import create_engine

def import_excel_to_postgres(excel_file, tablename):
    try:
        # 创建SQLAlchemy引擎
        engine = create_engine('postgresql://abc:abc@localhost:5432/abc')
        df = read_excel(excel_file)
        # if_exists可选值：append(追加)、replace(替换表)、fail(表存在则报错)
        df.to_sql(tablename, engine, if_exists='append', index=False)
        print(f"成功导入{len(df)}条数据")
    except Exception as e:
        print(f"错误信息：{e}")

# 主函数调用
if __name__ == "__main__":    
    import_excel_to_postgres(r"Data\data_sample_100_rows.xlsx", "player_behaviour")

4. 错误处理细化 #

捕获具体异常类型，方便快速定位问题：

• pg.Error：捕获PostgreSQL相关错误
• ValueError：捕获数据校验类错误
• 最后用Exception兜底其他未知错误

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