PostgreSQL中SQLAlchemy批量insert抛出AttributeError问题咨询

你的Statement接口使用方式错误：

1. INSERT语句默认执行后不返回结果行，你却用select(Test).from_statement()把插入语句包装成了查询ORM实体的SELECT语句，ORM执行时会尝试从返回结果中解析Test模型实例，但插入操作无返回结果集，对应的结果元数据是_NoResultMetaData类型，不存在_indexes_for_keys属性，直接触发报错。
2. 示例代码存在笔误：冲突判断传入的Company.local_id未定义，你当前操作的是Test表，没有Company模型。

不需要用select()包装插入语句，直接执行带冲突处理的插入语句即可。如果需要获取插入成功的记录主键，追加RETURNING子句就行。
修正后的代码：

import os

import sqlalchemy
from sqlalchemy import Column, INTEGER, TEXT
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class Test(Base):
    __tablename__ = 'test'
    id = Column(INTEGER, primary_key=True)
    data = Column(TEXT)


engine = sqlalchemy.create_engine(os.environ['DATABASE_CONNECTION'])
Session = sessionmaker(engine)
Base.metadata.create_all(engine, checkfirst=True)

buffer = [
    {'data': "First test"},
    {'data': "Second test"}
]

# 构造批量插入语句，指定冲突时忽略，索引字段替换为你实际业务的唯一键/主键即可
insert_stmt = insert(Test).values(buffer)
insert_or_ignore = insert_stmt.on_conflict_do_nothing(index_elements=[Test.id])
# 需要返回插入记录主键的话加returning，不需要可省略
insert_or_ignore = insert_or_ignore.returning(Test.id)

with Session() as session, session.begin():
    result = session.execute(insert_or_ignore)
    # 需要获取返回主键时调用，不需要可省略
    inserted_ids = result.scalars().all()

• 你当前使用的PostgreSQL 14.4、psycopg2-binary 2.9.3、SQLAlchemy 1.4.39版本组合完全支持上述写法，无需升级依赖。
• 这种直接执行insert构造语句的批量插入性能和bulk_insert_mappings相当，同时支持灵活的冲突处理逻辑，适配静默忽略重复条目的需求。

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