0.7.9
=====
+- orm
+ - [bug] Fixed bug mostly local to new
+ AbstractConcreteBase helper where the "type"
+ attribute from the superclass would not
+ be overridden on the subclass to produce the
+ "reserved for base" error message, instead placing
+ a do-nothing attribute there. This was inconsistent
+ vs. using ConcreteBase as well as all the behavior
+ of classical concrete mappings, where the "type"
+ column from the polymorphic base would be explicitly
+ disabled on subclasses, unless overridden
+ explicitly.
+
- sql
- [bug] Fixed CTE bug whereby positional
bound parameters present in the CTEs themselves
properties = util.importlater('sqlalchemy.orm', 'properties')
class DescriptorProperty(MapperProperty):
- """:class:`.MapperProperty` which proxies access to a
+ """:class:`.MapperProperty` which proxies access to a
user-defined descriptor."""
doc = None
self.key = key
if hasattr(prop, 'get_history'):
- def get_history(self, state, dict_,
+ def get_history(self, state, dict_,
passive=attributes.PASSIVE_OFF):
return prop.get_history(state, dict_, passive)
create_proxied_attribute(self.descriptor)\
(
self.parent.class_,
- self.key,
+ self.key,
self.descriptor,
lambda: self._comparator_factory(mapper),
doc=self.doc,
self._setup_event_handlers()
def do_init(self):
- """Initialization which occurs after the :class:`.CompositeProperty`
+ """Initialization which occurs after the :class:`.CompositeProperty`
has been associated with its parent mapper.
"""
self._setup_arguments_on_columns()
def _create_descriptor(self):
- """Create the Python descriptor that will serve as
+ """Create the Python descriptor that will serve as
the access point on instances of the mapped class.
"""
values = [getattr(instance, key) for key in self._attribute_keys]
# current expected behavior here is that the composite is
- # created on access if the object is persistent or if
- # col attributes have non-None. This would be better
+ # created on access if the object is persistent or if
+ # col attributes have non-None. This would be better
# if the composite were created unconditionally,
# but that would be a behavioral change.
if self.key not in dict_ and (
- state.key is not None or
+ state.key is not None or
not _none_set.issuperset(values)
):
dict_[self.key] = self.composite_class(*values)
setattr(instance, key, None)
else:
for key, value in zip(
- self._attribute_keys,
+ self._attribute_keys,
value.__composite_values__()):
setattr(instance, key, value)
return
# if column elements aren't loaded, skip.
- # __get__() will initiate a load for those
+ # __get__() will initiate a load for those
# columns
for k in self._attribute_keys:
if k not in dict_:
#assert self.key not in dict_
dict_[self.key] = self.composite_class(
- *[state.dict[key] for key in
+ *[state.dict[key] for key in
self._attribute_keys]
)
def insert_update_handler(mapper, connection, state):
"""After an insert or update, some columns may be expired due
to server side defaults, or re-populated due to client side
- defaults. Pop out the composite value here so that it
+ defaults. Pop out the composite value here so that it
recreates.
-
+
"""
state.dict.pop(self.key, None)
- event.listen(self.parent, 'after_insert',
+ event.listen(self.parent, 'after_insert',
insert_update_handler, raw=True)
- event.listen(self.parent, 'after_update',
+ event.listen(self.parent, 'after_update',
insert_update_handler, raw=True)
event.listen(self.parent, 'load', load_handler, raw=True, propagate=True)
event.listen(self.parent, 'refresh', load_handler, raw=True, propagate=True)
return str(self.parent.class_.__name__) + "." + self.key
class ConcreteInheritedProperty(DescriptorProperty):
- """A 'do nothing' :class:`.MapperProperty` that disables
+ """A 'do nothing' :class:`.MapperProperty` that disables
an attribute on a concrete subclass that is only present
on the inherited mapper, not the concrete classes' mapper.
Cases where this occurs include:
- * When the superclass mapper is mapped against a
- "polymorphic union", which includes all attributes from
+ * When the superclass mapper is mapped against a
+ "polymorphic union", which includes all attributes from
all subclasses.
* When a relationship() is configured on an inherited mapper,
but not on the subclass mapper. Concrete mappers require
- that relationship() is configured explicitly on each
- subclass.
+ that relationship() is configured explicitly on each
+ subclass.
"""
def warn():
raise AttributeError("Concrete %s does not implement "
"attribute %r at the instance level. Add this "
- "property explicitly to %s." %
+ "property explicitly to %s." %
(self.parent, self.key, self.parent))
class NoninheritedConcreteProp(object):
class SynonymProperty(DescriptorProperty):
- def __init__(self, name, map_column=None,
+ def __init__(self, name, map_column=None,
descriptor=None, comparator_factory=None,
doc=None):
self.name = name
if self.key not in parent.mapped_table.c:
raise sa_exc.ArgumentError(
"Can't compile synonym '%s': no column on table "
- "'%s' named '%s'"
+ "'%s' named '%s'"
% (self.name, parent.mapped_table.description, self.key))
elif parent.mapped_table.c[self.key] in \
parent._columntoproperty and \
raise sa_exc.ArgumentError(
"Can't call map_column=True for synonym %r=%r, "
"a ColumnProperty already exists keyed to the name "
- "%r for column %r" %
+ "%r for column %r" %
(self.key, self.name, self.name, self.key)
)
p = properties.ColumnProperty(parent.mapped_table.c[self.key])
parent._configure_property(
- self.name, p,
- init=init,
+ self.name, p,
+ init=init,
setparent=True)
p._mapped_by_synonym = self.key
self.inherits._inheriting_mappers.add(self)
self.passive_updates = self.inherits.passive_updates
self._all_tables = self.inherits._all_tables
+ for key, prop in mapper._props.iteritems():
+ if key not in self._props and \
+ not self._should_exclude(key, key, local=False,
+ column=None):
+ self._adapt_inherited_property(key, prop, False)
+
def _set_polymorphic_on(self, polymorphic_on):
self.polymorphic_on = polymorphic_on
# polymorphic_union.
# we'll make a separate ColumnProperty for it.
instrument = True
-
key = getattr(col, 'key', None)
if key:
if self._should_exclude(col.key, col.key, False, col):
== 'cobol')).first(), c2)
# ensure that the Manager mapper was compiled with the Manager id
- # column as higher priority. this ensures that "Manager.id"
+ # column as higher priority. this ensures that "Manager.id"
# is appropriately treated as the "id" column in the "manager"
# table (reversed from 0.6's behavior.)
sess.add(e1)
sess.flush()
sess.expunge_all()
- eq_(sess.query(Person).first(),
+ eq_(sess.query(Person).first(),
Engineer(primary_language='java', name='dilbert'))
def test_add_sub_parentcol_after_the_fact(self):
@testing.fails_if(lambda: True, "Not implemented until 0.7")
def test_foreign_keys_with_col(self):
- """Test that foreign keys that reference a literal 'id' subclass
+ """Test that foreign keys that reference a literal 'id' subclass
'id' attribute behave intuitively.
See [ticket:1892].
sa.orm.configure_mappers() # no exceptions here
def test_foreign_keys_with_col(self):
- """Test that foreign keys that reference a literal 'id' subclass
+ """Test that foreign keys that reference a literal 'id' subclass
'id' attribute behave intuitively.
See [ticket:1892].
from test.orm.test_events import _RemoveListeners
class ConcreteInhTest(_RemoveListeners, DeclarativeTestBase):
- def _roundtrip(self, Employee, Manager, Engineer, Boss, polymorphic=True):
+ def _roundtrip(self, Employee, Manager, Engineer, Boss,
+ polymorphic=True, explicit_type=False):
Base.metadata.create_all()
sess = create_session()
e1 = Engineer(name='dilbert', primary_language='java')
m1 = Manager(name='dogbert', golf_swing='fore!')
e3 = Engineer(name='vlad', primary_language='cobol')
b1 = Boss(name="pointy haired")
+
+ if polymorphic:
+ for obj in [e1, e2, m1, e3, b1]:
+ if explicit_type:
+ eq_(obj.type, obj.__mapper__.polymorphic_identity)
+ else:
+ assert_raises_message(
+ AttributeError,
+ "does not implement attribute .?'type' "
+ "at the instance level.",
+ getattr, obj, "type"
+ )
+ else:
+ assert "type" not in Engineer.__dict__
+ assert "type" not in Manager.__dict__
+ assert "type" not in Boss.__dict__
+
sess.add_all([e1, e2, m1, e3, b1])
sess.flush()
sess.expunge_all()
test_needs_autoincrement=True), Column('name'
, String(50)), Column('primary_language',
String(50)))
- managers = Table('managers', Base.metadata,
- Column('id',Integer, primary_key=True, test_needs_autoincrement=True),
- Column('name', String(50)),
+ managers = Table('managers', Base.metadata,
+ Column('id',Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('name', String(50)),
Column('golf_swing', String(50))
)
- boss = Table('boss', Base.metadata,
- Column('id',Integer, primary_key=True, test_needs_autoincrement=True),
- Column('name', String(50)),
+ boss = Table('boss', Base.metadata,
+ Column('id',Integer, primary_key=True, test_needs_autoincrement=True),
+ Column('name', String(50)),
Column('golf_swing', String(50))
)
punion = polymorphic_union({
- 'engineer': engineers,
+ 'engineer': engineers,
'manager' : managers,
'boss': boss}, 'type', 'punion')
class Manager(Employee):
__tablename__ = 'manager'
- employee_id = Column(Integer, primary_key=True,
+ employee_id = Column(Integer, primary_key=True,
test_needs_autoincrement=True)
name = Column(String(50))
golf_swing = Column(String(40))
__mapper_args__ = {
- 'polymorphic_identity':'manager',
+ 'polymorphic_identity':'manager',
'concrete':True}
class Boss(Manager):
__tablename__ = 'boss'
- employee_id = Column(Integer, primary_key=True,
+ employee_id = Column(Integer, primary_key=True,
test_needs_autoincrement=True)
name = Column(String(50))
golf_swing = Column(String(40))
__mapper_args__ = {
- 'polymorphic_identity':'boss',
+ 'polymorphic_identity':'boss',
'concrete':True}
class Engineer(Employee):
__tablename__ = 'engineer'
- employee_id = Column(Integer, primary_key=True,
+ employee_id = Column(Integer, primary_key=True,
test_needs_autoincrement=True)
name = Column(String(50))
primary_language = Column(String(40))
- __mapper_args__ = {'polymorphic_identity':'engineer',
+ __mapper_args__ = {'polymorphic_identity':'engineer',
'concrete':True}
self._roundtrip(Employee, Manager, Engineer, Boss)
def test_concrete_extension(self):
class Employee(ConcreteBase, Base, fixtures.ComparableEntity):
__tablename__ = 'employee'
- employee_id = Column(Integer, primary_key=True,
+ employee_id = Column(Integer, primary_key=True,
test_needs_autoincrement=True)
name = Column(String(50))
__mapper_args__ = {
- 'polymorphic_identity':'employee',
+ 'polymorphic_identity':'employee',
'concrete':True}
class Manager(Employee):
__tablename__ = 'manager'
- employee_id = Column(Integer, primary_key=True,
+ employee_id = Column(Integer, primary_key=True,
test_needs_autoincrement=True)
name = Column(String(50))
golf_swing = Column(String(40))
__mapper_args__ = {
- 'polymorphic_identity':'manager',
+ 'polymorphic_identity':'manager',
'concrete':True}
class Boss(Manager):
__tablename__ = 'boss'
- employee_id = Column(Integer, primary_key=True,
+ employee_id = Column(Integer, primary_key=True,
test_needs_autoincrement=True)
name = Column(String(50))
golf_swing = Column(String(40))
__mapper_args__ = {
- 'polymorphic_identity':'boss',
+ 'polymorphic_identity':'boss',
'concrete':True}
class Engineer(Employee):
__tablename__ = 'engineer'
- employee_id = Column(Integer, primary_key=True,
+ employee_id = Column(Integer, primary_key=True,
test_needs_autoincrement=True)
name = Column(String(50))
primary_language = Column(String(40))
- __mapper_args__ = {'polymorphic_identity':'engineer',
+ __mapper_args__ = {'polymorphic_identity':'engineer',
'concrete':True}
self._roundtrip(Employee, Manager, Engineer, Boss)
+
+ def test_ok_to_override_type_from_abstract(self):
+ class Employee(AbstractConcreteBase, Base, fixtures.ComparableEntity):
+ pass
+
+ class Manager(Employee):
+ __tablename__ = 'manager'
+ employee_id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
+ name = Column(String(50))
+ golf_swing = Column(String(40))
+
+ @property
+ def type(self):
+ return "manager"
+
+ __mapper_args__ = {
+ 'polymorphic_identity': "manager",
+ 'concrete':True}
+
+ class Boss(Manager):
+ __tablename__ = 'boss'
+ employee_id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
+ name = Column(String(50))
+ golf_swing = Column(String(40))
+
+ @property
+ def type(self):
+ return "boss"
+
+ __mapper_args__ = {
+ 'polymorphic_identity': "boss",
+ 'concrete':True}
+
+ class Engineer(Employee):
+ __tablename__ = 'engineer'
+ employee_id = Column(Integer, primary_key=True,
+ test_needs_autoincrement=True)
+ name = Column(String(50))
+ primary_language = Column(String(40))
+
+ @property
+ def type(self):
+ return "engineer"
+ __mapper_args__ = {'polymorphic_identity': "engineer",
+ 'concrete':True}
+ self._roundtrip(Employee, Manager, Engineer, Boss, explicit_type=True)
polymorphic_identity='hacker')
session = create_session()
tom = Manager('Tom', 'knows how to manage things')
+
+ assert_raises_message(AttributeError,
+ "does not implement attribute .?'type' at the instance level.",
+ setattr, tom, "type", "sometype")
+
jerry = Engineer('Jerry', 'knows how to program')
hacker = Hacker('Kurt', 'Badass', 'knows how to hack')
+
+ assert_raises_message(AttributeError,
+ "does not implement attribute .?'type' at the instance level.",
+ setattr, hacker, "type", "sometype")
+
session.add_all((tom, jerry, hacker))
session.flush()
})
mapper(Dest, dest_table, properties={
- 'many_a': relationship(A,back_populates='some_dest'),
+ 'many_a': relationship(A,back_populates='some_dest'),
'many_b': relationship(B,back_populates='some_dest')
})
sess = sessionmaker()()
self.tables.dest_table)
- ajoin = polymorphic_union({'a': a_table, 'b': b_table, 'c':c_table},
+ ajoin = polymorphic_union({'a': a_table, 'b': b_table, 'c':c_table},
'type','ajoin')
mapper(
A,
mapper(Dest, dest_table, properties={
'many_a': relationship(A,
- back_populates='some_dest',
+ back_populates='some_dest',
order_by=ajoin.c.id)
}
)
def go():
eq_(
[
- Dest(many_a=[A(aname='a1'),
- B(bname='b1'),
+ Dest(many_a=[A(aname='a1'),
+ B(bname='b1'),
B(bname='b2'),
- C(cname='c1')]),
+ C(cname='c1')]),
Dest(many_a=[A(aname='a2'), C(cname='c2')])],
sess.query(Dest).options(joinedload(Dest.many_a)).order_by(Dest.id).all())
self.classes.Dest,
self.tables.dest_table)
- ajoin = polymorphic_union({'a': a_table, 'b': b_table, 'c':c_table},
+ ajoin = polymorphic_union({'a': a_table, 'b': b_table, 'c':c_table},
'type','ajoin')
mapper(
A,
mapper(Dest, dest_table, properties={
'many_a': relationship(A,
- back_populates='some_dest',
+ back_populates='some_dest',
order_by=ajoin.c.id)
}
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
