def build_history_class(cls, prop, schema=None):
""" build a sql alchemy table for given prop
Args:
cls (Clocked): class to refer back to
prop (orm.RelationshipProperty|orm.ColumnProperty): property to build a history class for
schema (Optional[str]): schema to use for history table
Returns:
_TemporalProperty: history class with table
"""
class_name = "%s%s_%s" % (cls.__name__, 'History', prop.key)
table = build_history_table(cls, prop, schema)
model = type(
class_name,
(_TemporalProperty, declarative.declarative_base(metadata=cls.metadata)),
dict(
__table__=table,
entity=orm.relationship(cls, backref=orm.backref('%s_history' % prop.key, lazy='dynamic')),
)
)
if isinstance(prop, orm.RelationshipProperty):
mapper = sa.inspect(model)
rel = orm.relationship(
prop.argument,
primaryjoin=getattr(model, prop.info['temporal_on']) == prop.argument.id, # todo different shaped FKs
lazy="noload") # write only rel
mapper.add_property(prop.key, rel)
return model
and
def build_history_table(cls, prop, schema=None):
""" build a sql alchemy table for given prop
Args:
cls (Clocked): class to refer back to
prop (orm.ColumnProperty|orm.RelationshipProperty): property to build a history table for
schema (Optional[str]): schema to use for table, if None will use schema of given prop
Returns:
sa.Table: Table for property
"""
if isinstance(prop, orm.RelationshipProperty):
assert 'temporal_on' in prop.info, 'cannot temporal-ize a property without temporal_on=True'
prop_ = prop.parent.get_property(prop.info['temporal_on']) # converts rel prop to fk prop
assert prop_.parent.local_table is prop.parent.local_table
property_key = prop_.key
columns = (_copy_column(col) for col in prop_.columns)
else:
property_key = prop.key
columns = (_copy_column(col) for col in prop.columns)
local_table = cls.__table__
table_name = _truncate_identifier('%s_%s_%s' % (local_table.name, 'history', property_key))
index_name = _truncate_identifier('%s_effective_idx' % table_name)
effective_exclude_name = _truncate_identifier('%s_excl_effective' % table_name)
vclock_exclude_name = _truncate_identifier('%s_excl_vclock' % table_name)
constraints = [
sa.Index(index_name, 'effective', postgresql_using='gist'),
sap.ExcludeConstraint(
(sa.cast(sa.text('entity_id'), sap.TEXT), '='), ('effective', '&&'),
name=effective_exclude_name,
),
sap.ExcludeConstraint(
(sa.cast(sa.text('entity_id'), sap.TEXT), '='), ('vclock', '&&'),
name=vclock_exclude_name
),
]
foreign_key = getattr(cls, 'id') # TODO make this support different shape pks
return sa.Table(table_name, prop.parent.class_.metadata,
sa.Column('id', sap.UUID(as_uuid=True), default=uuid.uuid4, primary_key=True),
sa.Column('effective', sap.TSTZRANGE, default=effective_now, nullable=False),
sa.Column('vclock', sap.INT4RANGE, nullable=False),
sa.Column('entity_id', sa.ForeignKey(foreign_key)),
*columns,
*constraints,
schema=schema or local_table.schema,
keep_existing=True) # memoization ftw
not through the
prop.parent.class_Fixes joined table inheritance.
and