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

至少有两种方法：

• 数据库方法：创建相关主键；使用 SQLAlchemy，您可以定义例如基于您的简约示例name = Column('First Name', String(20), primary_key=True)
• 编码方法：检查属性、属性集是否已经存在于表中，否则创建。查看相关代码示例here。

在性能方面，我相信数据库方法更好。这也是更有意义的一个。

Answer 3

您可以创建一个类方法来获取或创建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