"""Versioned mixin class and other utilities."""
from sqlalchemy.ext.declarative import declared_attr
-from sqlalchemy.orm import mapper, class_mapper, attributes, object_mapper
-from sqlalchemy.orm.exc import UnmappedClassError, UnmappedColumnError
+from sqlalchemy.orm import mapper, attributes, object_mapper
+from sqlalchemy.orm.exc import UnmappedColumnError
from sqlalchemy import Table, Column, ForeignKeyConstraint, Integer, DateTime
from sqlalchemy import event
+import datetime
from sqlalchemy.orm.properties import RelationshipProperty
def col_references_table(col, table):
return True
return False
+def _is_versioning_col(col):
+ return "version_meta" in col.info
+
def _history_mapper(local_mapper):
cls = local_mapper.class_
if not super_mapper or local_mapper.local_table is not super_mapper.local_table:
cols = []
for column in local_mapper.local_table.c:
- if column.name == 'version':
+ if _is_versioning_col(column):
continue
col = column.copy()
if super_mapper:
super_fks.append(('version', super_history_mapper.local_table.c.version))
- cols.append(Column('version', Integer, primary_key=True, autoincrement=False))
- cols.append(Column('changed', DateTime, default=datetime.datetime.utcnow))
- else:
- cols.append(Column('version', Integer, primary_key=True, autoincrement=False))
- cols.append(Column('changed', DateTime, default=datetime.datetime.utcnow))
+
+ version_meta = {"version_meta": True} # add column.info to identify
+ # columns specific to versioning
+
+ # "version" stores the integer version id. This column is
+ # required.
+ cols.append(Column('version', Integer, primary_key=True,
+ autoincrement=False, info=version_meta))
+
+ # "changed" column stores the UTC timestamp of when the
+ # history row was created.
+ # This column is optional and can be omitted.
+ cols.append(Column('changed', DateTime,
+ default=datetime.datetime.utcnow,
+ 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
+ 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
if hasattr(obj, '__history_mapper__'):
yield obj
-def create_version(obj, session, deleted = False):
+def create_version(obj, session, deleted=False):
obj_mapper = object_mapper(obj)
history_mapper = obj.__history_mapper__
history_cls = history_mapper.class_
continue
for hist_col in hm.local_table.c:
- if hist_col.key == 'version' or hist_col.key == 'changed':
+ if _is_versioning_col(hist_col):
continue
obj_col = om.local_table.c[hist_col.key]
for obj in versioned_objects(session.dirty):
create_version(obj, session)
for obj in versioned_objects(session.deleted):
- create_version(obj, session, deleted = True)
+ create_version(obj, session, deleted=True)
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import clear_mappers, Session, deferred, relationship
from sqlalchemy.testing import AssertsCompiledSQL, eq_, assert_raises
-from sqlalchemy.testing.entities import BasicEntity, ComparableEntity
+from sqlalchemy.testing.entities import ComparableEntity
from sqlalchemy.orm import exc as orm_exc
engine = None
-
def setup():
global engine
engine = create_engine('sqlite://', echo=True)
name = Column(String(50))
type = Column(String(20))
- __mapper_args__ = {'polymorphic_on':type, 'polymorphic_identity':'base'}
+ __mapper_args__ = {'polymorphic_on': type, 'polymorphic_identity': 'base'}
class SubClassSeparatePk(BaseClass):
__tablename__ = 'subtable1'
base_id = Column(Integer, ForeignKey('basetable.id'))
subdata1 = Column(String(50))
- __mapper_args__ = {'polymorphic_identity':'sep'}
+ __mapper_args__ = {'polymorphic_identity': 'sep'}
class SubClassSamePk(BaseClass):
__tablename__ = 'subtable2'
id = Column(Integer, ForeignKey('basetable.id'), primary_key=True)
subdata2 = Column(String(50))
- __mapper_args__ = {'polymorphic_identity':'same'}
+ __mapper_args__ = {'polymorphic_identity': 'same'}
self.create_tables()
sess = self.session
base1.name = 'base1mod'
same1.subdata2 = 'same1subdatamod'
- sep1.name ='sep1mod'
+ sep1.name = 'sep1mod'
sess.commit()
BaseClassHistory = BaseClass.__history_mapper__.class_
q = sess.query(SubSubHistory)
self.assert_compile(
q,
- "SELECT subsubtable_history.id AS subsubtable_history_id, "
+ "SELECT "
+
+ "subsubtable_history.id AS subsubtable_history_id, "
"subtable_history.id AS subtable_history_id, "
"basetable_history.id AS basetable_history_id, "
+
"basetable_history.name AS basetable_history_name, "
+
"basetable_history.type AS basetable_history_type, "
+
"subsubtable_history.version AS subsubtable_history_version, "
"subtable_history.version AS subtable_history_version, "
"basetable_history.version AS basetable_history_version, "
+
+ "subsubtable_history.changed AS subsubtable_history_changed, "
+ "subtable_history.changed AS subtable_history_changed, "
+ "basetable_history.changed AS basetable_history_changed, "
+
"subtable_history.base_id AS subtable_history_base_id, "
"subtable_history.subdata1 AS subtable_history_subdata1, "
"subsubtable_history.subdata2 AS subsubtable_history_subdata2 "
id = Column(Integer, primary_key=True)
name = Column(String(50))
type = Column(String(50))
- __mapper_args__ = {'polymorphic_on':type, 'polymorphic_identity':'base'}
+ __mapper_args__ = {'polymorphic_on': type,
+ 'polymorphic_identity': 'base'}
class SubClass(BaseClass):
subname = Column(String(50), unique=True)
- __mapper_args__ = {'polymorphic_identity':'sub'}
+ __mapper_args__ = {'polymorphic_identity': 'sub'}
self.create_tables()
sess = self.session
sess.commit()
- b1.name='b1modified'
+ b1.name = 'b1modified'
BaseClassHistory = BaseClass.__history_mapper__.class_
SubClassHistory = SubClass.__history_mapper__.class_
eq_(
- sess.query(BaseClassHistory).order_by(BaseClassHistory.id, BaseClassHistory.version).all(),
+ sess.query(BaseClassHistory).order_by(BaseClassHistory.id,
+ BaseClassHistory.version).all(),
[BaseClassHistory(id=1, name='b1', type='base', version=1)]
)
- sc.name ='s1modified'
- b1.name='b1modified2'
+ sc.name = 's1modified'
+ b1.name = 'b1modified2'
eq_(
- sess.query(BaseClassHistory).order_by(BaseClassHistory.id, BaseClassHistory.version).all(),
+ sess.query(BaseClassHistory).order_by(BaseClassHistory.id,
+ BaseClassHistory.version).all(),
[
BaseClassHistory(id=1, name='b1', type='base', version=1),
BaseClassHistory(id=1, name='b1modified', type='base', version=2),
# test the unique constraint on the subclass
# column
- sc.name ="modifyagain"
+ sc.name = "modifyagain"
sess.flush()
def test_unique(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
