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
+from sqlalchemy import event, util
import datetime
from sqlalchemy.orm.properties import RelationshipProperty
super_fks = []
def _col_copy(col):
+ orig = col
col = col.copy()
+ orig.info['history_copy'] = col
col.unique = False
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} # add column.info to identify
+ # columns specific to versioning
+
for column in local_mapper.local_table.c:
if _is_versioning_col(column):
continue
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_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(
# "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))
+ cols.append(Column(
+ 'changed', DateTime,
+ default=datetime.datetime.utcnow,
+ info=version_meta))
if super_fks:
cols.append(ForeignKeyConstraint(*zip(*super_fks)))
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
- )
+ 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:
from unittest import TestCase
from sqlalchemy.ext.declarative import declarative_base
from .history_meta import Versioned, versioned_session
-from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Boolean
-from sqlalchemy.orm import clear_mappers, Session, deferred, relationship
+from sqlalchemy import create_engine, Column, Integer, String, \
+ ForeignKey, Boolean, select
+from sqlalchemy.orm import clear_mappers, Session, deferred, relationship, \
+ column_property
from sqlalchemy.testing import AssertsCompiledSQL, eq_, assert_raises
from sqlalchemy.testing.entities import ComparableEntity
from sqlalchemy.orm import exc as orm_exc
+import warnings
+
+warnings.simplefilter("error")
engine = None
class SubClassSeparatePk(BaseClass):
__tablename__ = 'subtable1'
- id = Column(Integer, primary_key=True)
+ id = column_property(
+ Column(Integer, primary_key=True),
+ BaseClass.id
+ )
base_id = Column(Integer, ForeignKey('basetable.id'))
subdata1 = Column(String(50))
class SubClassSamePk(BaseClass):
__tablename__ = 'subtable2'
- id = Column(Integer, ForeignKey('basetable.id'), primary_key=True)
+ id = Column(
+ Integer, ForeignKey('basetable.id'), primary_key=True)
subdata2 = Column(String(50))
__mapper_args__ = {'polymorphic_identity': 'same'}
class SubClass(BaseClass):
__tablename__ = 'subtable'
- id = Column(Integer, primary_key=True)
+ id = column_property(
+ Column(Integer, primary_key=True),
+ BaseClass.id
+ )
base_id = Column(Integer, ForeignKey('basetable.id'))
subdata1 = Column(String(50))
q = sess.query(SubSubHistory)
self.assert_compile(
q,
+
+
"SELECT "
"subsubtable_history.id AS subsubtable_history_id, "
"subtable_history.id AS subtable_history_id, "
"basetable_history.id AS basetable_history_id, "
+ "subsubtable_history.changed AS subsubtable_history_changed, "
+ "subtable_history.changed AS subtable_history_changed, "
+ "basetable_history.changed AS basetable_history_changed, "
+
"basetable_history.name AS basetable_history_name, "
"basetable_history.type AS basetable_history_type, "
"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, "
name='ss1', subdata1='sd11',
subdata2='sd22', version=2))
+ def test_joined_inheritance_changed(self):
+ class BaseClass(Versioned, self.Base, ComparableEntity):
+ __tablename__ = 'basetable'
+
+ id = Column(Integer, primary_key=True)
+ name = Column(String(50))
+ type = Column(String(20))
+
+ __mapper_args__ = {
+ 'polymorphic_on': type,
+ 'polymorphic_identity': 'base'
+ }
+
+ class SubClass(BaseClass):
+ __tablename__ = 'subtable'
+
+ id = Column(Integer, ForeignKey('basetable.id'), primary_key=True)
+
+ __mapper_args__ = {'polymorphic_identity': 'sep'}
+
+ self.create_tables()
+
+ BaseClassHistory = BaseClass.__history_mapper__.class_
+ SubClassHistory = SubClass.__history_mapper__.class_
+ sess = self.session
+ s1 = SubClass(name='s1')
+ sess.add(s1)
+ sess.commit()
+
+ s1.name = 's2'
+ sess.commit()
+ actual_changed_base = sess.scalar(
+ select([BaseClass.__history_mapper__.local_table.c.changed]))
+ actual_changed_sub = sess.scalar(
+ select([SubClass.__history_mapper__.local_table.c.changed]))
+ h1 = sess.query(BaseClassHistory).first()
+ eq_(h1.changed, actual_changed_base)
+ eq_(h1.changed, actual_changed_sub)
+
+ h1 = sess.query(SubClassHistory).first()
+ eq_(h1.changed, actual_changed_base)
+ eq_(h1.changed, actual_changed_sub)
def test_single_inheritance(self):
class BaseClass(Versioned, self.Base, ComparableEntity):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
