Add support for testing of super types

This commit is contained in:
Konsta Vesterinen
2014-03-05 15:36:11 +02:00
parent 2ec373443a
commit 4730bb9cd1
4 changed files with 101 additions and 7 deletions

View File

@@ -4,6 +4,14 @@ Changelog
Here you can see the full list of changes between each SQLAlchemy-Utils release. Here you can see the full list of changes between each SQLAlchemy-Utils release.
0.25.0 (2014-03-05)
^^^^^^^^^^^^^^^^^^^
- Added single table inheritance support for generic_relationship
- Added support for comparing class super types with generic relationships
- BC break: In order to support different inheritance strategies generic_relationship now uses class names as discriminators instead of table names.
0.24.4 (2014-03-05) 0.24.4 (2014-03-05)
^^^^^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^^^^

View File

@@ -29,25 +29,85 @@ Generic relationship is a form of relationship that supports creating a 1 to man
# Some general usage to attach an event to a user. # Some general usage to attach an event to a user.
us_1 = User() user = User()
cu_1 = Customer() customer = Customer()
session.add_all([us_1, cu_1]) session.add_all([user, customer])
session.commit() session.commit()
ev = Event() ev = Event()
ev.object = us_1 ev.object = user
session.add(ev) session.add(ev)
session.commit() session.commit()
# Find the event we just made. # Find the event we just made.
session.query(Event).filter_by(object=us_1).first() session.query(Event).filter_by(object=user).first()
# Find any events that are bound to users. # Find any events that are bound to users.
session.query(Event).filter(Event.object.is_type(User)).all() session.query(Event).filter(Event.object.is_type(User)).all()
Inheritance
^^^^^^^^^^^
::
class Employee(self.Base):
__tablename__ = 'employee'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.String(50))
type = sa.Column(sa.String(20))
__mapper_args__ = {
'polymorphic_on': type,
'polymorphic_identity': 'employee'
}
class Manager(Employee):
__mapper_args__ = {
'polymorphic_identity': 'manager'
}
class Engineer(Employee):
__mapper_args__ = {
'polymorphic_identity': 'engineer'
}
class Activity(self.Base):
__tablename__ = 'event'
id = sa.Column(sa.Integer, primary_key=True)
object_type = sa.Column(sa.Unicode(255))
object_id = sa.Column(sa.Integer, nullable=False)
object = generic_relationship(object_type, object_id)
Now same as before we can add some objects::
manager = Manager()
session.add(manager)
session.commit()
activity = Activity()
activity.object = manager
session.add(activity)
session.commit()
# Find the activity we just made.
session.query(Event).filter_by(object=manager).first()
We can even test super types::
# Find any events that are bound to users.
session.query(Activity).filter(Event.object.is_type(Employee)).all()
Abstract base classes Abstract base classes
^^^^^^^^^^^^^^^^^^^^^ ^^^^^^^^^^^^^^^^^^^^^

View File

@@ -1,6 +1,7 @@
from collections import Iterable from collections import Iterable
import six import six
import sqlalchemy as sa
from sqlalchemy.ext.hybrid import hybrid_property from sqlalchemy.ext.hybrid import hybrid_property
from sqlalchemy.orm import attributes, class_mapper from sqlalchemy.orm import attributes, class_mapper
from sqlalchemy.orm import ColumnProperty from sqlalchemy.orm import ColumnProperty
@@ -154,8 +155,16 @@ class GenericRelationshipProperty(MapperProperty):
return ~(self == other) return ~(self == other)
def is_type(self, other): def is_type(self, other):
discriminator = unicode(other.__name__) mapper = sa.inspect(other)
return self.property._discriminator_col == discriminator # Iterate through the weak sequence in order to get the actual
# mappers
class_names = [unicode(other.__name__)]
class_names.extend([
unicode(submapper.class_.__name__)
for submapper in mapper._inheriting_mappers
])
return self.property._discriminator_col.in_(class_names)
def instrument_class(self, mapper): def instrument_class(self, mapper):
attributes.register_attribute( attributes.register_attribute(

View File

@@ -142,3 +142,20 @@ class TestGenericRelationship(TestCase):
statement = self.Event.object.is_type(self.Manager) statement = self.Event.object.is_type(self.Manager)
q = self.session.query(self.Event).filter(statement) q = self.session.query(self.Event).filter(statement)
assert q.first() is not None assert q.first() is not None
def test_compare_super_type(self):
manager1 = self.Manager()
manager2 = self.Manager()
self.session.add_all([manager1, manager2])
self.session.commit()
event1 = self.Event(object=manager1)
event2 = self.Event(object=manager2)
self.session.add_all([event1, event2])
self.session.commit()
statement = self.Event.object.is_type(self.Employee)
q = self.session.query(self.Event).filter(statement)
assert q.first() is not None