<s> '''<STR_LIT>''' <EOL> from datetime import datetime <EOL> import inspect <EOL> from sqlalchemy import Table , Column , and_ , desc <EOL> from sqlalchemy . orm import mapper , MapperExtension , EXT_CONTINUE , object_session <EOL> from elixir import Integer , DateTime <EOL> from elixir . statements import Statement <EOL> from elixir . properties import EntityBuilder <EOL> from elixir . entity import getmembers <EOL> __all__ = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> __doc_all__ = [ ] <EOL> def get_entity_where ( instance ) : <EOL> clauses = [ ] <EOL> for column in instance . table . primary_key . columns : <EOL> instance_value = getattr ( instance , column . name ) <EOL> clauses . append ( column == instance_value ) <EOL> return and_ ( * clauses ) <EOL> def get_history_where ( instance ) : <EOL> clauses = [ ] <EOL> history_columns = instance . __history_table__ . primary_key . columns <EOL> for column in instance . table . primary_key . columns : <EOL> instance_value = getattr ( instance , column . name ) <EOL> history_column = getattr ( history_columns , column . name ) <EOL> clauses . append ( history_column == instance_value ) <EOL> return and_ ( * clauses ) <EOL> class VersionedMapperExtension ( MapperExtension ) : <EOL> def before_insert ( self , mapper , connection , instance ) : <EOL> version_colname , timestamp_colname = instance . __class__ . __versioned_column_names__ <EOL> setattr ( instance , version_colname , <NUM_LIT:1> ) <EOL> setattr ( instance , timestamp_colname , datetime . now ( ) ) <EOL> return EXT_CONTINUE <EOL> def before_update ( self , mapper , connection , instance ) : <EOL> old_values = instance . table . select ( get_entity_where ( instance ) ) . execute ( ) . fetchone ( ) <EOL> ignored = instance . __class__ . __ignored_fields__ <EOL> version_colname , timestamp_colname = instance . __class__ . __versioned_column_names__ <EOL> for key in instance . table . c . keys ( ) : <EOL> if key in ignored : <EOL> continue <EOL> if getattr ( instance , key ) != old_values [ key ] : <EOL> dict_values = dict ( old_values . items ( ) ) <EOL> connection . execute ( <EOL> instance . __class__ . __history_table__ . insert ( ) , dict_values ) <EOL> old_version = getattr ( instance , version_colname ) <EOL> setattr ( instance , version_colname , old_version + <NUM_LIT:1> ) <EOL> setattr ( instance , timestamp_colname , datetime . now ( ) ) <EOL> break <EOL> return EXT_CONTINUE <EOL> def before_delete ( self , mapper , connection , instance ) : <EOL> connection . execute ( instance . __history_table__ . delete ( <EOL> get_history_where ( instance ) <EOL> ) ) <EOL> return EXT_CONTINUE <EOL> versioned_mapper_extension = VersionedMapperExtension ( ) <EOL> class VersionedEntityBuilder ( EntityBuilder ) : <EOL> def __init__ ( self , entity , ignore = None , check_concurrent = False , <EOL> column_names = None ) : <EOL> self . entity = entity <EOL> self . add_mapper_extension ( versioned_mapper_extension ) <EOL> self . check_concurrent = check_concurrent <EOL> if column_names is None : <EOL> column_names = [ '<STR_LIT:version>' , '<STR_LIT>' ] <EOL> entity . __versioned_column_names__ = column_names <EOL> if ignore is None : <EOL> ignore = [ ] <EOL> ignore . extend ( column_names ) <EOL> entity . __ignored_fields__ = ignore <EOL> def create_non_pk_cols ( self ) : <EOL> version_colname , timestamp_colname = self . entity . __versioned_column_names__ <EOL> self . add_table_column ( Column ( version_colname , Integer ) ) <EOL> self . add_table_column ( Column ( timestamp_colname , DateTime ) ) <EOL> if self . check_concurrent : <EOL> self . entity . _descriptor . version_id_col = '<STR_LIT>' <EOL> def after_table ( self ) : <EOL> entity = self . entity <EOL> version_colname , timestamp_colname = entity . __versioned_column_names__ <EOL> after_revert_events = [ ] <EOL> for name , func in getmembers ( entity , inspect . ismethod ) : <EOL> if getattr ( func , '<STR_LIT>' , False ) : <EOL> after_revert_events . append ( func ) <EOL> skipped_columns = [ version_colname ] <EOL> if self . check_concurrent : <EOL> skipped_columns . append ( '<STR_LIT>' ) <EOL> columns = [ <EOL> column . copy ( ) for column in entity . table . c <EOL> if column . name not in skipped_columns <EOL> ] <EOL> columns . append ( Column ( version_colname , Integer , primary_key = True ) ) <EOL> table = Table ( entity . table . name + '<STR_LIT>' , entity . table . metadata , <EOL> * columns <EOL> ) <EOL> entity . __history_table__ = table <EOL> class Version ( object ) : <EOL> pass <EOL> Version . __name__ = entity . __name__ + '<STR_LIT>' <EOL> Version . __versioned_entity__ = entity <EOL> mapper ( Version , entity . __history_table__ ) <EOL> version_col = getattr ( table . c , version_colname ) <EOL> timestamp_col = getattr ( table . c , timestamp_colname ) <EOL> def get_versions ( self ) : <EOL> v = object_session ( self ) . query ( Version ) . filter ( get_history_where ( self ) ) . order_by ( version_col ) . all ( ) <EOL> v . append ( self ) <EOL> return v <EOL> def get_as_of ( self , dt ) : <EOL> if getattr ( self , timestamp_colname ) < dt : <EOL> return self <EOL> sess = object_session ( self ) <EOL> query = sess . query ( Version ) . filter ( and_ ( get_history_where ( self ) , <EOL> timestamp_col <= dt ) ) . order_by ( desc ( timestamp_col ) ) . limit ( <NUM_LIT:1> ) <EOL> return query . first ( ) <EOL> def revert_to ( self , to_version ) : <EOL> if isinstance ( to_version , Version ) : <EOL> to_version = getattr ( to_version , version_colname ) <EOL> old_version = table . select ( and_ ( <EOL> get_history_where ( self ) , <EOL> version_col == to_version <EOL> ) ) . execute ( ) . fetchone ( ) <EOL> entity . table . update ( get_entity_where ( self ) ) . execute ( <EOL> dict ( old_version . items ( ) ) <EOL> ) <EOL> table . delete ( and_ ( get_history_where ( self ) , <EOL> version_col >= to_version ) ) . execute ( ) <EOL> self . expire ( ) <EOL> for event in after_revert_events : <EOL> event ( self ) <EOL> def revert ( self ) : <EOL> assert getattr ( self , version_colname ) > <NUM_LIT:1> <EOL> self . revert_to ( getattr ( self , version_colname ) - <NUM_LIT:1> ) <EOL> def compare_with ( self , version ) : <EOL> differences = { } <EOL> for column in self . table . c : <EOL> if column . name in ( version_colname , '<STR_LIT>' ) : <EOL> continue <EOL> this = getattr ( self , column . name ) <EOL> that = getattr ( version , column . name ) <EOL> if this != that : <EOL> differences [ column . name ] = ( this , that ) <EOL> return differences <EOL> entity . versions = property ( get_versions ) <EOL> entity . get_as_of = get_as_of <EOL> entity . revert_to = revert_to <EOL> entity . revert = revert <EOL> entity . compare_with = compare_with <EOL> Version . compare_with = compare_with <EOL> acts_as_versioned = Statement ( VersionedEntityBuilder ) <EOL> def after_revert ( func ) : <EOL> """<STR_LIT>""" <EOL> func . _elixir_after_revert = True <EOL> return func </s>
