插入数据库时​​如何避免（或解决）唯一约束冲突？

Question

我有一个带有部分唯一约束的表，其中一个 cosuser_id 只能将一个 is_primary 标志设置为 True。

我想要实现的行为是：当我插入一行 is_primary=True 和一个在数据库中已经将 is_primary 标志设置为 True 的 customer_id 时，我想将该旧条目设置为 is_primary=False ，并且新条目将以 is_primary=True 插入，而不是引发 IntegrityError。

我正在使用带有 postgresql 数据库的 flask-sqlalchemy，是否有任何机制可以轻松执行此操作，或者我需要手动执行此操作，首先更新旧条目，然后插入新行？

class Customer_address(db.Model):
    customer_id = db.Column(None, db.ForeignKey('customer.id'), nullable=False)
    is_primary = db.Column(db.Boolean, nullable=False)
    __table_args__ = (
                Index(
                    'only_one_primary_address',  # Index name
                    'customer_id',  # Columns which are part of the index
                    unique=True,
                    postgresql_where=(is_primary)  # The condition
                ),)

Answer 1

虽然插入新记录并更新现有记录不是单个数据库操作，但我们可以在 SQLAlchemy 中创建一个event listener 来监听新地址插入并执行更新。这在概念上类似于使用 cmets 中讨论的数据库触发器，只是它发生在 SQLALchemy ORM 会话中。

import sqlalchemy as sa
from sqlalchemy import orm

@sa.event.listens_for(Customer_address, 'before_insert')
def receive_init(mapper, connection, target):
    """Listen for the insertion of new customer addresses."""
    # Get the active session 
    session = orm.session.object_session(target)
    cls = target.__class__
    # Update the existing is_primary == True record for this customer
    session.query(cls).filter(sa.and_(cls.customer_id == target.customer_id,
                                      cls.is_primary)).update({'is_primary': False}, synchronize_session='fetch')

在侦听器中执行的Query.update 绕过了 ORM 机制，因此值得阅读文档中与之相关的警告。在这种情况下，将synchronize_session 设置为'fetch' 会使数据库和会话保持同步，但代价是发出额外的SELECT，因此我们会得到如下的一系列事件：

2020-11-22 09:04:26,565 INFO sqlalchemy.engine.base.Engine SELECT customer_address.id AS customer_address_id                 
FROM customer_address                                                                                                        
WHERE customer_address.customer_id = %(customer_id_1)s AND customer_address.is_primary                                       
2020-11-22 09:04:26,565 INFO sqlalchemy.engine.base.Engine {'customer_id_1': 1}                                              
2020-11-22 09:04:26,566 INFO sqlalchemy.engine.base.Engine UPDATE customer_address SET is_primary=%(is_primary)s WHERE customer_address.customer_id = %(customer_id_1)s AND customer_address.is_primary                                                   
2020-11-22 09:04:26,566 INFO sqlalchemy.engine.base.Engine {'is_primary': False, 'customer_id_1': 1}                         
2020-11-22 09:04:26,567 INFO sqlalchemy.engine.base.Engine INSERT INTO customer_address (customer_id, is_primary) VALUES (%(customer_id)s, %(is_primary)s) RETURNING customer_address.id                                                                  
2020-11-22 09:04:26,567 INFO sqlalchemy.engine.base.Engine {'customer_id': 1, 'is_primary': True}                            
2020-11-22 09:04:26,568 INFO sqlalchemy.engine.base.Engine COMMIT

另请注意，如果您一次为客户创建多个地址，则此方法不起作用：

# Will violate the constraint
for _ in range(5):
    session.add(Customer_address(customer_id=1, is_primary=True))
session.commit()  

如果这可能在您的代码中发生，您可以考虑在客户的地址集合或地址实例上侦听 events。