如何使用SQLAlchemy向带serial自增ID的PostgreSQL表插入数据

问题原因 #

你当前代码报错的核心原因是：INSERT INTO {table_name} (SELECT * FROM temporary_table) 语句默认要求临时表的列数、列顺序和目标表完全一致。如果目标表包含serial类型的自增主键，而你的DataFrame中没有对应主键列，临时表就会比目标表少一列，触发缺省键值的报错。

优化后代码 #

def export_to_sql(df, table_name):
    from sqlalchemy import create_engine, text
    # 建议将数据库连接信息提取为全局配置或函数入参，不要硬编码在函数内部
    engine = create_engine(f'postgresql://{user}:{password}@{host}:5432/{user}')
    
    # 临时表写入策略改为replace，避免上次执行异常未删表导致的旧数据残留
    df.to_sql(con=engine,
              name='temporary_table',
              if_exists='replace',
              index=False,
              method='multi')
    
    with engine.begin() as cnx:
        # 查询目标表所有非自增列，过滤默认值为序列nextval的serial类型主键
        col_query = text("""
            SELECT string_agg(column_name, ',') 
            FROM information_schema.columns 
            WHERE table_name = :table_name
              AND column_default NOT LIKE 'nextval%%'
            ORDER BY ordinal_position
        """)
        insert_cols = cnx.execute(col_query, {"table_name": table_name}).scalar()
        
        # 插入时仅指定非自增列，serial主键由PostgreSQL自动分配值
        insert_sql = text(f"""
            INSERT INTO {table_name} ({insert_cols}) 
            SELECT {insert_cols} FROM temporary_table 
            ON CONFLICT DO NOTHING;
            DROP TABLE temporary_table;
        """)
        cnx.execute(insert_sql)

注：如果使用的sqlalchemy版本低于1.4，可删除代码中text包装，直接传入SQL字符串即可。

功能说明 #

• 完全兼容你提供的所有表结构：带serial主键的symbols、hashtags表插入时无需传入id列，数据库自动生成主键值；无自增列的关联表tweet_symols、tweet_hashtags的写入逻辑和原函数完全一致，不会影响原有业务
• 保留了原函数临时表批量写入、冲突忽略的特性，性能和防重逻辑不变

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