mirror of
https://github.com/GSA/notifications-api.git
synced 2025-12-16 10:12:32 -05:00
17f5742e2c
`create_history` used to initialise an empty model instance and assign attributes to it one by one. Assigned attributes are collected by iterating over internal state of the modified record and stored in a dictionary, so the order in which they are set on the history instance is undefined. This works fine for most model instances, but recent changes to the Template models require `template_type` to be set before `reply_to`, otherwise Template doesn't know which column to use to persist the `reply_to` value. This means that we need to order the attributes when they're being assigned. We achieved this by defining a constructor for Template objects that assigns `template_type` first. In order to make it work with `create_history` we replace individual `setattr` calls with a single call to the Template constructor which passes the entire data dictionary as keyword arguments. This works because all persisted attributes are defined as columns and SQLAlchemy accepts all column values as constructor arguments.
221 lines
7.3 KiB
Python
221 lines
7.3 KiB
Python
"""Versioned mixin class and other utilities.
|
|
|
|
This is an adapted version of:
|
|
|
|
https://bitbucket.org/zzzeek/sqlalchemy/raw/master/examples/versioned_history/history_meta.py
|
|
|
|
It does not use the create_version function from the orginal which looks for changes to models
|
|
as we just insert a copy of a model to the history table on create or update.
|
|
|
|
Also it does not add a created_at timestamp to the history table as we already have created_at
|
|
and updated_at timestamps.
|
|
|
|
Lastly when to create a version is done manually in dao_utils version decorator and not via
|
|
session events.
|
|
|
|
"""
|
|
import datetime
|
|
from sqlalchemy.ext.declarative import declared_attr
|
|
from sqlalchemy.orm import mapper, attributes, object_mapper
|
|
from sqlalchemy.orm.properties import RelationshipProperty, ColumnProperty
|
|
from sqlalchemy import Table, Column, ForeignKeyConstraint, Integer
|
|
from sqlalchemy import util
|
|
|
|
|
|
def col_references_table(col, table):
|
|
for fk in col.foreign_keys:
|
|
if fk.references(table):
|
|
return True
|
|
return False
|
|
|
|
|
|
def _is_versioning_col(col):
|
|
return "version_meta" in col.info
|
|
|
|
|
|
def _history_mapper(local_mapper): # noqa (C901 too complex)
|
|
cls = local_mapper.class_
|
|
|
|
# set the "active_history" flag
|
|
# on on column-mapped attributes so that the old version
|
|
# of the info is always loaded (currently sets it on all attributes)
|
|
for prop in local_mapper.iterate_properties:
|
|
getattr(local_mapper.class_, prop.key).impl.active_history = True
|
|
|
|
super_mapper = local_mapper.inherits
|
|
super_history_mapper = getattr(cls, '__history_mapper__', None)
|
|
|
|
polymorphic_on = None
|
|
super_fks = []
|
|
|
|
def _col_copy(col):
|
|
orig = col
|
|
col = col.copy()
|
|
orig.info['history_copy'] = col
|
|
col.unique = False
|
|
|
|
# if the column is nullable, we could end up overwriting an on-purpose null value with a default.
|
|
# if it's not nullable, however, the default may be relied upon to correctly set values within the database,
|
|
# so we should preserve it
|
|
if col.nullable:
|
|
col.default = col.server_default = None
|
|
return col
|
|
|
|
properties = util.OrderedDict()
|
|
if not super_mapper or \
|
|
local_mapper.local_table is not super_mapper.local_table:
|
|
cols = []
|
|
version_meta = {"version_meta": True}
|
|
for column in local_mapper.local_table.c:
|
|
if _is_versioning_col(column):
|
|
continue
|
|
|
|
col = _col_copy(column)
|
|
|
|
if super_mapper and \
|
|
col_references_table(column, super_mapper.local_table):
|
|
super_fks.append(
|
|
(
|
|
col.key,
|
|
list(super_history_mapper.local_table.primary_key)[0]
|
|
)
|
|
)
|
|
|
|
cols.append(col)
|
|
|
|
if column is local_mapper.polymorphic_on:
|
|
polymorphic_on = col
|
|
|
|
orig_prop = local_mapper.get_property_by_column(column)
|
|
# carry over column re-mappings
|
|
if len(orig_prop.columns) > 1 or \
|
|
orig_prop.columns[0].key != orig_prop.key:
|
|
properties[orig_prop.key] = tuple(
|
|
col.info['history_copy'] for col in orig_prop.columns)
|
|
|
|
if super_mapper:
|
|
super_fks.append(
|
|
(
|
|
'version', super_history_mapper.local_table.c.version
|
|
)
|
|
)
|
|
|
|
# "version" stores the integer version id. This column is
|
|
# required.
|
|
cols.append(
|
|
Column(
|
|
'version', Integer, primary_key=True,
|
|
autoincrement=False, info=version_meta))
|
|
|
|
if super_fks:
|
|
cols.append(ForeignKeyConstraint(*zip(*super_fks)))
|
|
|
|
table = Table(
|
|
local_mapper.local_table.name + '_history',
|
|
local_mapper.local_table.metadata,
|
|
*cols,
|
|
schema=local_mapper.local_table.schema
|
|
)
|
|
else:
|
|
# single table inheritance. take any additional columns that may have
|
|
# been added and add them to the history table.
|
|
for column in local_mapper.local_table.c:
|
|
if column.key not in super_history_mapper.local_table.c:
|
|
col = _col_copy(column)
|
|
super_history_mapper.local_table.append_column(col)
|
|
table = None
|
|
|
|
if super_history_mapper:
|
|
bases = (super_history_mapper.class_,)
|
|
|
|
if table is not None:
|
|
properties['changed'] = (
|
|
(table.c.changed, ) +
|
|
tuple(super_history_mapper.attrs.changed.columns)
|
|
)
|
|
|
|
else:
|
|
bases = local_mapper.base_mapper.class_.__bases__
|
|
versioned_cls = type.__new__(type, "%sHistory" % cls.__name__, bases, {})
|
|
|
|
m = mapper(
|
|
versioned_cls,
|
|
table,
|
|
inherits=super_history_mapper,
|
|
polymorphic_on=polymorphic_on,
|
|
polymorphic_identity=local_mapper.polymorphic_identity,
|
|
properties=properties
|
|
)
|
|
cls.__history_mapper__ = m
|
|
|
|
if not super_history_mapper:
|
|
local_mapper.local_table.append_column(
|
|
Column('version', Integer, default=1, nullable=False)
|
|
)
|
|
local_mapper.add_property(
|
|
"version", local_mapper.local_table.c.version)
|
|
|
|
|
|
class Versioned(object):
|
|
@declared_attr
|
|
def __mapper_cls__(cls):
|
|
def map(cls, *arg, **kw):
|
|
mp = mapper(cls, *arg, **kw)
|
|
_history_mapper(mp)
|
|
return mp
|
|
return map
|
|
|
|
@classmethod
|
|
def get_history_model(cls):
|
|
history_mapper = cls.__history_mapper__
|
|
return history_mapper.class_
|
|
|
|
|
|
def create_history(obj, history_cls=None):
|
|
if not history_cls:
|
|
history_mapper = obj.__history_mapper__
|
|
history_cls = history_mapper.class_
|
|
|
|
obj_mapper = object_mapper(obj)
|
|
|
|
obj_state = attributes.instance_state(obj)
|
|
data = {}
|
|
for prop in obj_mapper.iterate_properties:
|
|
|
|
# expired object attributes and also deferred cols might not
|
|
# be in the dict. force it them load no matter what by using getattr().
|
|
if prop.key not in obj_state.dict:
|
|
getattr(obj, prop.key)
|
|
|
|
# if prop is a normal col just set it on history model
|
|
if isinstance(prop, ColumnProperty):
|
|
if not data.get(prop.key):
|
|
data[prop.key] = getattr(obj, prop.key)
|
|
|
|
# if the prop is a relationship property and there is a
|
|
# corresponding prop on hist object then set the
|
|
# relevant "_id" prop to the id of the current object.prop.id.
|
|
# This is so foreign keys get set on history when
|
|
# the source object is new and therefore property foo_id does
|
|
# not yet have a value before insert
|
|
|
|
elif isinstance(prop, RelationshipProperty):
|
|
if hasattr(history_cls, prop.key + '_id'):
|
|
foreign_obj = getattr(obj, prop.key)
|
|
# if it's a nullable relationship, foreign_obj will be None, and we actually want to record that
|
|
data[prop.key + '_id'] = getattr(foreign_obj, 'id', None)
|
|
|
|
if not obj.version:
|
|
obj.version = 1
|
|
obj.created_at = datetime.datetime.utcnow()
|
|
else:
|
|
obj.version += 1
|
|
now = datetime.datetime.utcnow()
|
|
obj.updated_at = now
|
|
data['updated_at'] = now
|
|
|
|
data['version'] = obj.version
|
|
data['created_at'] = obj.created_at
|
|
|
|
return history_cls(**data)
|