1

我正在尝试在我的 SQLAlchemy 数据源中的实体和从其他地方（外部 REST API）拉入的实体之间执行每个字段的合并。我想做的是这样的：

class Person:
    __tablename__ = "people"
    id = Column(Integer,primary_key=True)
    name = Column(String)
    nameDatestamp = Column(DateTime)
    address = Column(String)
    addressDatestamp = Column(DateTime)

def merge(myPerson, foreignPerson):
   if myPerson.nameDateStamp < foreignPerson.nameDateStamp:
      myPerson.name = foreignPerson.name
      myPerson.nameDateStamp = foreignPerson.nameDateStamp
   if myPerson.addressDatestamp < foreignPerson.addressDateStamp:
      myPerson.addressDatestamp = foreignPerson.addressDateStamp
      myPerson.address = foreignPerson.address

对于很多课程和很多领域。这似乎太冗长而不能成为最佳实践。

  • 我可以引入新的数据模型，例如 DateStampedString（它将由一个日期戳和一个字符串组成）、DateStampedRelationship 等，但我担心使用多个表的额外间接性会提高合并的速度

  • 我可以使用带有参数的 Python 装饰器，这些参数会在运行时和创建表之前将额外的日期戳列动态添加到我的模型中，例如

    @datestamp(name,address)
class Person:
   ...

  • 也许我可以以某种方式利用sqlalchemy.types.TypeDecorator, 来构建一个新的数据类型，但它似乎对从 (opaque type)->(sqlalchemy type) 开始而不是将两种类型捆绑在一起感兴趣。

我正在尝试做的事情是否有最佳实践？

编辑：我正在寻找

  1. 一种更简洁的方式来声明某些字段有日期戳
  2. 某种方式来迭代带日期标记的字段，这样我的合并函数就不必知道它正在合并哪个键
  3. 每当值更改时，通过某种方式使日期戳保持最新
4

1 回答 1

3

通用合并

您可以通过查询对象轻松地迭代 ( thing, )。datestamp例如，如果您想获取地址和日期戳，您可以这样做：

session.query(Person.address, Person.addressDatestamp).all()

这将返回一组 ( address, addressDatestamp) 元组。（它们实际上是命名元组，但您可以只使用索引）。如果您有一堆要更新的属性，则实际上不需要这样做。您可以动态执行此操作的一种方法是传递合并属性元组列表和作为 (Person, foreignPerson) 元组的查询并执行以下操作：

attrs = [("address", "addressDatestamp"), ("name", "nameDatestamp")]
person_tuples = # some way to generate (Person, ForeignPerson) tuples
def merge(attrs, person_tuples):
    for person, foreign in person_tuples:
        for attr, date in attrs:
            if getattr(person, date) < getattr(foreign, date):
                setattr(person, attr) = getattr(foreign, attr)
                setattr(person, date) = getattr(foreign, date)

    return person_tuples

这将检查每个属性的日期戳，然后存储属性，如果外国是较新的（+ 也存储日期）。

如果您的属性始终在表单中<attr>，<attr>Datestamp那么您可以将其缩短为：

attrs = ["name", "address"]
def merge(attrs, person_tuples):
    for person, foreign in person_tuples:
        for attr in attrs:
            date = attr + "Datestamp"
            if getattr(person, date) < getattr(foreign, date):
                setattr(person, attr) = getattr(foreign, attr)
                setattr(person, date) = setattr(foreign, date)

如果该属性有时可能不存在，您可以将 getattr 调用更改为getattr(object, attr, default)，它不会引发错误。

动态类

如果您希望能够动态生成带有日期戳的模型，您可以使用元类（稍微复杂一点，特别是因为它会干扰 SQLA 的声明性基础等），或者您可以创建一个类工厂，如下所示：

def datestamped_factory(class_name, attrlist, timestamp="Datestamp", superclass_list=None):
    superclass_list = superclass or (object,)
    cols = dict((attr, Column(String)) for attr in attrlist)
    cols.update(dict((attr + timestamp, Column(DateTime)) for attr in attrlist)
    cols["timestamped_attrs"] = attrlist
    # create a merge specific to the class (so only need to pass person_tuples)
    cols["merge"] = classmethod(lambda cls, person_tuples: merge(cls.timestamped_attrs, person_tuples))
    return type(class_name, superclass_list, cols)

（可以将其添加到工厂cols["class_merge"] = classmethod(lambda cls, person_tuples: merge(cls.timestamped_attrs, person_tuples）））

要创建您的 person 方法，您可以执行以下操作：

class Base(sqlalchemy.declarative_base()):
     id = Column(Integer, primary_key=True)

Person = datestamped__factory("Person", ["name", "address"], superclass_list = (Base,))
Person.__tablename__ = "person"

（替换sqlalchemy.declarative_base()为您正在使用的任何 Base 类......假设您正在使用 ORM）。

您可能会更花哨并编写一个元类，该元类可以找到所有作为字符串的列并向它们添加日期戳+创建适当的合并并创建适当地更新时间戳的方法，但这可能比您需要的更花哨。

于 2012-08-25T01:12:56.780 回答