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して新しいデータ型を構築することもできますが、2 つの型を一緒にバンドルするのではなく、(不透明な型)->(sqlalchemy 型) から移行することに関心があるようです。

私がやろうとしていることのベストプラクティスはありますか?

編集：私は探しています

  1. 一部のフィールドに日付スタンプがあることを宣言するより簡潔な方法
  2. マージ関数がどのキーをマージしているかを知る必要がないように、日付スタンプ付きフィールドを反復処理する方法
  3. 値が変更されるたびに日付スタンプを最新の状態に保つ何らかの方法
4

1 に答える 1

3

一般的なマージ

クエリ オブジェクトを介して( thing, ) を簡単に反復処理できます。datestampたとえば、住所と日付スタンプを取得したい場合は、次のようにします。

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

address( , addressDatestamp) タプルのセットを返します。(実際には名前付きタプルですが、代わりにインデックスを使用できます)。更新したい属性がたくさんある場合は、実際にこれを行う必要はありません。これを動的に行う方法の 1 つは、属性タプルのリストと (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()（ ORMを使用していると仮定して、使用している基本クラスに置き換えてください）。

文字列であるすべての列を見つけてそれらに日付スタンプを追加し、適切なマージを作成し、タイムスタンプを適切に更新するメソッドを作成するメタクラスをより洗練して作成することもできますが、それはおそらく必要以上に凝っています。

于 2012-08-25T01:12:56.780 に答える