<s> """<STR_LIT>""" <EOL> from sqlalchemy . ext . declarative import declared_attr <EOL> from sqlalchemy . orm import mapper , attributes , object_mapper <EOL> from sqlalchemy . orm . exc import UnmappedColumnError <EOL> from sqlalchemy import Table , Column , ForeignKeyConstraint , Integer , DateTime <EOL> from sqlalchemy import event , util <EOL> import datetime <EOL> from sqlalchemy . orm . properties import RelationshipProperty <EOL> def col_references_table ( col , table ) : <EOL> for fk in col . foreign_keys : <EOL> if fk . references ( table ) : <EOL> return True <EOL> return False <EOL> def _is_versioning_col ( col ) : <EOL> return "<STR_LIT>" in col . info <EOL> def _history_mapper ( local_mapper ) : <EOL> cls = local_mapper . class_ <EOL> for prop in local_mapper . iterate_properties : <EOL> getattr ( local_mapper . class_ , prop . key ) . impl . active_history = True <EOL> super_mapper = local_mapper . inherits <EOL> super_history_mapper = getattr ( cls , '<STR_LIT>' , None ) <EOL> polymorphic_on = None <EOL> super_fks = [ ] <EOL> def _col_copy ( col ) : <EOL> orig = col <EOL> col = col . copy ( ) <EOL> orig . info [ '<STR_LIT>' ] = col <EOL> col . unique = False <EOL> col . default = col . server_default = None <EOL> return col <EOL> properties = util . OrderedDict ( ) <EOL> if not super_mapper or local_mapper . local_table is not super_mapper . local_table : <EOL> cols = [ ] <EOL> version_meta = { "<STR_LIT>" : True } <EOL> for column in local_mapper . local_table . c : <EOL> if _is_versioning_col ( column ) : <EOL> continue <EOL> col = _col_copy ( column ) <EOL> if super_mapper and col_references_table ( column , super_mapper . local_table ) : <EOL> super_fks . append ( <EOL> ( <EOL> col . key , <EOL> list ( super_history_mapper . local_table . primary_key ) [ <NUM_LIT:0> ] <EOL> ) <EOL> ) <EOL> cols . append ( col ) <EOL> if column is local_mapper . polymorphic_on : <EOL> polymorphic_on = col <EOL> orig_prop = local_mapper . get_property_by_column ( column ) <EOL> if len ( orig_prop . columns ) > <NUM_LIT:1> or orig_prop . columns [ <NUM_LIT:0> ] . key != orig_prop . key : <EOL> properties [ orig_prop . key ] = tuple ( <EOL> col . info [ '<STR_LIT>' ] for col in orig_prop . columns ) <EOL> if super_mapper : <EOL> super_fks . append ( <EOL> ( <EOL> '<STR_LIT:version>' , super_history_mapper . local_table . c . version <EOL> ) <EOL> ) <EOL> cols . append ( <EOL> Column ( <EOL> '<STR_LIT:version>' , Integer , primary_key = True , <EOL> autoincrement = False , info = version_meta ) ) <EOL> cols . append ( Column ( <EOL> '<STR_LIT>' , DateTime , <EOL> default = datetime . datetime . utcnow , <EOL> info = version_meta ) ) <EOL> if super_fks : <EOL> cols . append ( ForeignKeyConstraint ( * zip ( * super_fks ) ) ) <EOL> table = Table ( <EOL> local_mapper . local_table . name + '<STR_LIT>' , <EOL> local_mapper . local_table . metadata , <EOL> * cols , <EOL> schema = local_mapper . local_table . schema <EOL> ) <EOL> else : <EOL> for column in local_mapper . local_table . c : <EOL> if column . key not in super_history_mapper . local_table . c : <EOL> col = _col_copy ( column ) <EOL> super_history_mapper . local_table . append_column ( col ) <EOL> table = None <EOL> if super_history_mapper : <EOL> bases = ( super_history_mapper . class_ , ) <EOL> if table is not None : <EOL> properties [ '<STR_LIT>' ] = ( <EOL> ( table . c . changed , ) + <EOL> tuple ( super_history_mapper . attrs . changed . columns ) <EOL> ) <EOL> else : <EOL> bases = local_mapper . base_mapper . class_ . __bases__ <EOL> versioned_cls = type . __new__ ( type , "<STR_LIT>" % cls . __name__ , bases , { } ) <EOL> m = mapper ( <EOL> versioned_cls , <EOL> table , <EOL> inherits = super_history_mapper , <EOL> polymorphic_on = polymorphic_on , <EOL> polymorphic_identity = local_mapper . polymorphic_identity , <EOL> properties = properties <EOL> ) <EOL> cls . __history_mapper__ = m <EOL> if not super_history_mapper : <EOL> local_mapper . local_table . append_column ( <EOL> Column ( '<STR_LIT:version>' , Integer , default = <NUM_LIT:1> , nullable = False ) <EOL> ) <EOL> local_mapper . add_property ( <EOL> "<STR_LIT:version>" , local_mapper . local_table . c . version ) <EOL> class Versioned ( object ) : <EOL> @ declared_attr <EOL> def __mapper_cls__ ( cls ) : <EOL> def map ( cls , * arg , ** kw ) : <EOL> mp = mapper ( cls , * arg , ** kw ) <EOL> _history_mapper ( mp ) <EOL> return mp <EOL> return map <EOL> def versioned_objects ( iter ) : <EOL> for obj in iter : <EOL> if hasattr ( obj , '<STR_LIT>' ) : <EOL> yield obj <EOL> def create_version ( obj , session , deleted = False ) : <EOL> obj_mapper = object_mapper ( obj ) <EOL> history_mapper = obj . __history_mapper__ <EOL> history_cls = history_mapper . class_ <EOL> obj_state = attributes . instance_state ( obj ) <EOL> attr = { } <EOL> obj_changed = False <EOL> for om , hm in zip ( <EOL> obj_mapper . iterate_to_root ( ) , <EOL> history_mapper . iterate_to_root ( ) <EOL> ) : <EOL> if hm . single : <EOL> continue <EOL> for hist_col in hm . local_table . c : <EOL> if _is_versioning_col ( hist_col ) : <EOL> continue <EOL> obj_col = om . local_table . c [ hist_col . key ] <EOL> try : <EOL> prop = obj_mapper . get_property_by_column ( obj_col ) <EOL> except UnmappedColumnError : <EOL> continue <EOL> if prop . key not in obj_state . dict : <EOL> getattr ( obj , prop . key ) <EOL> a , u , d = attributes . get_history ( obj , prop . key ) <EOL> if d : <EOL> attr [ prop . key ] = d [ <NUM_LIT:0> ] <EOL> obj_changed = True <EOL> elif u : <EOL> attr [ prop . key ] = u [ <NUM_LIT:0> ] <EOL> elif a : <EOL> attr [ prop . key ] = a [ <NUM_LIT:0> ] <EOL> obj_changed = True <EOL> if not obj_changed : <EOL> for prop in obj_mapper . iterate_properties : <EOL> if isinstance ( prop , RelationshipProperty ) and attributes . get_history ( <EOL> obj , prop . key , <EOL> passive = attributes . PASSIVE_NO_INITIALIZE ) . has_changes ( ) : <EOL> for p in prop . local_columns : <EOL> if p . foreign_keys : <EOL> obj_changed = True <EOL> break <EOL> if obj_changed is True : <EOL> break <EOL> if not obj_changed and not deleted : <EOL> return <EOL> attr [ '<STR_LIT:version>' ] = obj . version <EOL> hist = history_cls ( ) <EOL> for key , value in attr . items ( ) : <EOL> setattr ( hist , key , value ) <EOL> session . add ( hist ) <EOL> obj . version += <NUM_LIT:1> <EOL> def versioned_session ( session ) : <EOL> @ event . listens_for ( session , '<STR_LIT>' ) <EOL> def before_flush ( session , flush_context , instances ) : <EOL> for obj in versioned_objects ( session . dirty ) : <EOL> create_version ( obj , session ) <EOL> for obj in versioned_objects ( session . deleted ) : <EOL> create_version ( obj , session , deleted = True ) </s>
