Temporal Options stored on Model
| 49 | |
| 50 | |
| 51 | class TemporalOption: |
| 52 | """ Temporal Options stored on Model """ |
| 53 | def __init__(self, |
| 54 | history_models: typing.Dict[T_PROPS, nine.Type[TemporalProperty]], |
| 55 | temporal_props: typing.Iterable[T_PROPS], |
| 56 | clock_model: nine.Type[EntityClock], |
| 57 | activity_cls: nine.Type[TemporalActivityMixin] = None, |
| 58 | allow_persist_on_commit: bool = False): |
| 59 | self.history_models = history_models |
| 60 | self.temporal_props = temporal_props |
| 61 | |
| 62 | self.clock_model = clock_model |
| 63 | self.activity_cls = activity_cls |
| 64 | |
| 65 | self.allow_persist_on_commit = allow_persist_on_commit |
| 66 | |
| 67 | @property |
| 68 | def clock_table(self): |
| 69 | """ DEPRECATED: use .clock_model instead -- Clock Model for this Model""" |
| 70 | warnings.warn( |
| 71 | 'use TemporalOption.clock_model instead', |
| 72 | PendingDeprecationWarning) |
| 73 | return self.clock_model |
| 74 | |
| 75 | @property |
| 76 | def history_tables(self): |
| 77 | """ DEPRECATED: use .history_models instead -- list of history models for this Model""" |
| 78 | warnings.warn( |
| 79 | 'use TemporalOption.history_models instead', |
| 80 | PendingDeprecationWarning) |
| 81 | return self.history_models |
| 82 | |
| 83 | @staticmethod |
| 84 | def make_clock(effective_lower: dt.datetime, |
| 85 | vclock_lower: int, |
| 86 | **kwargs) -> _ClockSet: |
| 87 | """construct a clock set tuple""" |
| 88 | effective_upper = kwargs.get('effective_upper', None) |
| 89 | vclock_upper = kwargs.get('vclock_upper', None) |
| 90 | |
| 91 | effective = psql_extras.DateTimeTZRange( |
| 92 | effective_lower, effective_upper) |
| 93 | vclock = psql_extras.NumericRange(vclock_lower, vclock_upper) |
| 94 | |
| 95 | return _ClockSet(effective, vclock) |
| 96 | |
| 97 | def record_history(self, |
| 98 | clocked: 'Clocked', |
| 99 | session: orm.Session, |
| 100 | timestamp: dt.datetime): |
| 101 | """record all history for a given clocked object""" |
| 102 | new_tick = self._get_new_tick(clocked) |
| 103 | |
| 104 | is_strict_mode = session.info[STRICT_MODE_KEY] |
| 105 | vclock_history = attributes.get_history(clocked, 'vclock') |
| 106 | is_vclock_unchanged = (vclock_history.unchanged and |
| 107 | new_tick == vclock_history.unchanged[0]) |
| 108 |