Получить ключ из исключения IntegrityError

Question

Я пытаюсь сохранить список пользовательских объектов с помощью sqlalchemy:

session.bulk_save_objects(listOfObjects)

И получение IntegrityError:

psycopg2.IntegrityError: duplicate key value violates unique constraint "ppoi_ukey"

А: Key ("id")=(42555) already exists.

Есть ли способ получить ключ Id (42555) в виде целого числа, чтобы откатиться, извлечь этот ключ из списка и снова вставить список без него?

Answer 1

Я попробовал кое-что, что, я думаю, могло бы вам помочь.

from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker 
from sqlalchemy.exc import IntegrityError

Base = declarative_base()

#Create a dummy model 
class Person(Base):
    __tablename__ = 'person'
    id = Column(Integer, primary_key=True)
    name = Column(String(250), nullable=False)

    def __repr__(self):
        return "{ id="+str(self.id)+", name="+self.name+ " }"

 # Create an engine that stores data in the local directory's
# sqlalchemy_example.db file.
engine = create_engine('sqlite:///sqlalchemy_example.db')

# Create all tables 
Base.metadata.create_all(engine)
DBSession = sessionmaker(bind=engine)
session = DBSession()
list_of_objects = []
new_person = Person(id=1,name='new person')
list_of_objects.append(new_person)
new_person2 = Person(id=2,name='new person2')
list_of_objects.append(new_person2)
# Violating unique constraint purposely 
new_person3 = Person(id=1,name='new person')
list_of_objects.append(new_person3)
print(list_of_objects)

for person in list_of_objects:
# with session.no_autoflush:
  try:
    session.add(person)
    session.commit()
  except IntegrityError as e:
    session.rollback()
    #print(e) #prints whole exception 
    print(e.params) #print lists of param to the query
    print(e.params[0]) #assuming the first param is unique id which is violating constraint you can get it from here and use it as you wanted ..
  except Exception as ex:
      pass
person = session.query(Person).all()
print(len(person)) #returns 2