SQLAlchemy，防止重复行

Question

我想知道是否有可能防止向数据库提交重复项。例如，假设有一个类如下

class Employee(Base):
   id = Column(Integer, primary_key=True)
   name = Column(String)

如果我要制作一系列这些物品，

employee1 = Employee(name='bob')
employee2 = Employee(name='bob')

session.add_all([employee1, employee2])
session.commit()

我想只添加单个行到数据库中，并employee1与employee2以指向同一个对象在内存中（如果可能）。

SQLAlchemy 中是否有实现此目的的功能？或者我是否需要以编程方式确保重复项不存在？

Answer 1

替代get_or_create()解决方案：

from sqlalchemy.orm.exc import NoResultFound
# ...

def get_or_create(self, model, **kwargs):
    """
    Usage:
    class Employee(Base):
        __tablename__ = 'employee'
        id = Column(Integer, primary_key=True)
        name = Column(String, unique=True)

    get_or_create(Employee, name='bob')
    """
    instance = get_instance(model, **kwargs)
    if instance is None:
        instance = create_instance(model, **kwargs)
    return instance


def create_instance(model, **kwargs):
    """create instance"""
    try:
        instance = model(**kwargs)
        sess.add(instance)
        sess.flush()
    except Exception as msg:
        mtext = 'model:{}, args:{} => msg:{}'
        log.error(mtext.format(model, kwargs, msg))
        sess.rollback()
        raise(msg)
    return instance


def get_instance(self, model, **kwargs):
    """Return first instance found."""
    try:
        return sess.query(model).filter_by(**kwargs).first()
    except NoResultFound:
        return

Answer 2

您可以创建一个类方法来获取或创建一个Employee-- 如果存在则获取它，否则创建：

@classmethod
def get_or_create(cls, name):
    exists = db.session.query(Employee.id).filter_by(name=name).scalar() is not None
    if exists:
        return db.session.query(Employee).filter_by(name=name).first()
    return cls(name=name)


employee1 = Employee(name='bob')
db.session.add(employee1)
employee2 = Employee(name='bob')

employee1 == employee2  # False


bob1 = Employee.get_or_create(name='bob')
if bob1 not in db.session:
    db.session.add(bob1)

len(add_to_session) # 1

bob2 = Employee.get_or_create(name='bob')
if bob2 not in db.session:
    db.session.add(bob2)

len(add_to_session) # 1

bob1 == bob2  # True