SQLAlchemy中的Percolumn时间戳

Question

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

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（它将由一个datestamp和一个string组成）、DateStampedRelationship等，但我担心使用多个表的额外间接性会提高合并的速度

• 我可以使用Python修饰符和参数，在运行时和创建表之前，动态地将额外的datestamp列添加到模型中

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

• 也许我可以利用sqlalchemy.types.TypeDecorator来构建一个新的数据类型，但它似乎对从（不透明类型）->>；（sqlalchemy类型）而不是将两个类型捆绑在一起感兴趣。

对于我要做的事情有没有最佳实践？在

编辑：我正在寻找

1. 声明某些字段有日期戳的更简洁的方法
2. 迭代日期戳字段的方法，这样merge函数就不必知道它在合并哪个键
3. 使日期戳在值更改时保持最新的方法

Answer 1

通用合并

您可以轻松地在查询对象中迭代（thing，datestamp）。例如，如果要获取地址和日期戳，可以执行以下操作：

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

它将返回一组（address，addressDatestamp）元组。（它们实际上是namedtuples，但是您可以使用索引来代替）。如果要更新一堆属性，实际上不需要这样做。动态执行此操作的一种方法是传递merge属性元组列表和一个tuple为（Person，foreignPerson）的查询，并执行以下操作：

^{pr2}$

这将检查每个属性的datestamp，如果外来属性是较新的，则存储该属性（+还存储日期）。在

如果属性的格式始终是<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()替换为您正在使用的任何基类…假设您使用的是ORM）。在

你可以变得更花哨一些，编写一个元类来查找字符串中的所有列并向它们添加日期戳+创建适当的合并并创建适当更新时间戳的方法，但这可能比您需要的要复杂得多。在