SQLAlchemy，防止重复行

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

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中是否有完成此功能的功能？或者我需要确保重复项不存在编程？

来源

2017-06-13 nven

这听起来像你有错误的主键。如果这是你想要的，你应该把名字作为主键。尽管如此，人们无法真正改变名称而不会引起巨大的多表数据库更新，因为所有外键都必须改变。另外，如果您尝试将名称更改为现有名称，会发生什么情况？你只想消灭已经存在的那一排？ – Omnifarious

按名称创建唯一索引（即使它不是主键）看起来像强制名称唯一性的正确方法。 – 9000

为了防止提交重复文件，请执行@ 9000所说的操作。要让会话在第二个例子中只创建一次bob，请参阅[unique object recipes]（https://bitbucket.org/zzzeek/sqlalchemy/wiki/UsageRecipes/UniqueObject）。 –

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

来源

2017-06-13 03:13:16

用于快速访问https://stackoverflow.com/a/32952421/4386191存在'表达式。 –

这很有趣。我希望有一种方法来解决这个提交。我认为在'get_or_create'将'bob1'对象返回给'bob2 = Employee.get_or_create（name ='bob'）''调用之前，需要首先将bob1添加到会话中？在我的包的当前工作流程中，我首先生成这些类型的对象的大型列表，并且根本没有与数据库进行交互，直到在@nven oops提交 – nven

之前，您是对的，它必须被添加到会话中。我将此添加到示例中，并显示了可能适用于您的模式。不需要提交，只需在检查唯一性后将对象添加到会话中即可。 –

另一种get_or_cerate（）解决方案。

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

来源

2017-06-14 08:46:41 farax

SQLAlchemy，防止重复行

回答

相关问题