SQL Alchemy 根据多个值检查重复记录

Question

我想根据属性检查数据库实例中的现有行。

这是我的模型：

class BasketModel(Base):
    __tablename__ = 'basket'
    id = sqlalchemy.Column(sqlalchemy.Integer, primary_key=True)
    org_id = sqlalchemy.Column(sqlalchemy.Integer, sqlalchemy.ForeignKey('organization.id'))
    first_attribute = sqlalchemy.Column(sqlalchemy.String(2))
    second_attribute = sqlalchemy.Column(sqlalchemy.Integer)
    ... # many more attributes

在我的关系模型中，如果新记录的所有非关键属性（唯一的 id 除外）都相等，则新记录被视为重复。

entityA = BasketModel(org_id=1, first_attribute="Hello", second_attribute="World")
entityB = BasketModel(org_id=1, first_attribute="Hello", second_attribute="World")
entityA == entityB # Should be True

现在，如果我想添加一条新记录，我可以简单地做（感谢 ORM 抽象）：

session.add(entityA)

但是如果我想根据值检查现有记录，我必须检查每个字段（这非常难看，不可缩放且令人沮丧）：

# create entity A
entityA = BasketModel(org_id=1, first_attribute="Hello", second_attribute="World")

# add entity A 
session.add(entityA)
session.flush()
session.commit()

# Now A is in the Database

#create entity B
entityB = BasketModel(org_id=1, first_attribute="Hello", second_attribute="World")


# Check if any record exists with those attributes

session.query(BasketModel).filter(
                BasketModel.org_id == entityB.org_id,
                BasketModel.first_attribute == entityB.first_attribute,
                BasketModel.second_attribute == entityB.second_attribute,
                ... # and I have many more
                ).count()

这“工作正常”，因为它根据每个字段的值输出我所期望的重复行。但它的编码非常困难。

有没有办法做这样的事情，检查每个非键属性并检查是否相等？（奇怪的是我没有找到，因为我很确定每条记录都映射到 ORM 中的一种幕后集合……可以“轻松”检查发生的事件……

session.query(BasketModel).exists(entityB) # I did not find any API for that...

Answer 1

解决此问题的一种可能方法是将非键列的值的散列存储在具有唯一约束的列中。这样，我们只需要在插入或更新记录时针对单个列进行检查。

class MyModel(Base):
    id = Column(Integer, primary_key=True)
    record_hash = Column(String, unique=True)
    ...

哈希值是这样构造的（未经测试）：

import hashlib
...

mymodel = MyModel(attr1='a', attr2='b',...)
non_key_cols = sorted(
    filter(lambda c: c.name != 'record_hash' and not c.primary_key, MyModel.__table__.columns
    )
)
hash_vals = [str(getattr(mymodel, col)) for col in non_key_cols]
hash_ = hashlib.sha256(''.join(hash_vals)).hexdigest()
mymodel.record_hash = hash_
session.add(mymodel)

毫无疑问，使用 SQLAlchemy 有更简洁的方法可以做到这一点——也许使用 a function to provide a default value，但上面的代码传达了一般的想法。

如果创建或更新具有相同非键值的实例，hash_record 上的唯一约束将触发 IntegrityError 并阻止插入。

请注意，更新记录时必须重新计算 hash_record，如果添加或删除列或以这种方式更改列，则必须为整个 table 重新计算它们的值的字符串表示形式发生了变化。

Answer 2

检查是否已有值不是你的工作。您的工作是惰性数据并最终捕获错误“重复数据” 添加到您的 table 字段上的 uniq-index。您最终可以将所有字段散列为一个，但这又不是您的工作。但我很确定数据库已经优化得很好。

Answer 3

您可以使用 UniqueConstraint 并在插入时捕获数据库错误如果你有 sqlalchemy 这个块应该运行

import sqlalchemy as sa
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker,declarative_base

engine = sa.create_engine('sqlite:///foo_3.db')
Session = sessionmaker(engine)
Base = declarative_base()

class BasketModel(Base):
    __tablename__ = 'basket'
    id = sa.Column(sa.Integer, primary_key=True)
    org_id = sa.Column(sa.Integer, sa.ForeignKey('organization.id'))
    first_attribute = sa.Column(sa.String(2))
    second_attribute = sa.Column(sa.Integer)
    
    __table_args__ = (
        (sa.UniqueConstraint("first_attribute","second_attribute","org_id"),)
      )

class Organization(Base):
    __tablename__ = 'organization'
    id = sa.Column(sa.Integer, primary_key=True)
    
Base.metadata.create_all(bind=engine)
with Session() as s:
    s.add(Organization())
    s.add(Organization())
    s.commit()
    orgs=s.query(Organization).all()
    print([org.id for org in orgs])
    s.add(BasketModel(org_id=2,first_attribute="AB",second_attribute=2))
    s.commit()
    res=s.query(BasketModel).filter_by(org_id=2).all()
    print(res)
    try:
        s.add(BasketModel(org_id=2,first_attribute="AB",second_attribute=2))
        s.commit()
    except IntegrityError as e:
        print("duplicate row",e)

[1, 2]
[<__main__.BasketModel object at 0x7f459e9221f0>]
duplicate row (sqlite3.IntegrityError) UNIQUE constraint failed: basket.first_attribute, basket.second_attribute, basket.org_id
[SQL: INSERT INTO basket (org_id, first_attribute, second_attribute) VALUES (?, ?, ?)]
[parameters: (2, 'AB', 2)]
(Background on this error at: https://sqlalche.me/e/14/gkpj)

SQL Alchemy 根据多个值检查重复记录

SQL Alchemy check for duplicate records based on multiple values

python

orm

sqlalchemy