disabled on subclasses, unless overridden
explicitly.
+ - [bug] A warning is emitted when lazy='dynamic'
+ is combined with uselist=False. This is an
+ exception raise in 0.8.
+
- sql
- [bug] Fixed CTE bug whereby positional
bound parameters present in the CTEs themselves
The :func:`~.orm.dynamic_loader` function is essentially the same
as :func:`~.orm.relationship` with the ``lazy='dynamic'`` argument specified.
+.. warning::
+
+ The "dynamic" loader applies to **collections only**. It is not valid
+ to use "dynamic" loaders with many-to-one, one-to-one, or uselist=False
+ relationships. Newer versions of SQLAlchemy emit warnings or exceptions
+ in these cases.
Setting Noload
---------------
"""
from sqlalchemy import log, util
-from sqlalchemy import exc as sa_exc
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy.sql import operators
from sqlalchemy.orm import (
)
from sqlalchemy.orm.query import Query
from sqlalchemy.orm.util import has_identity
-from sqlalchemy.orm import attributes, collections
+from sqlalchemy.orm import collections
class DynaLoader(strategies.AbstractRelationshipLoader):
def init_class_attribute(self, mapper):
self.is_class_level = True
-
+ if not self.uselist:
+ util.warn(
+ "On relationship %s, 'dynamic' loaders cannot be used with "
+ "many-to-one/one-to-one relationships and/or "
+ "uselist=False." % self.parent_property)
strategies._register_attribute(self,
mapper,
useobject=True,
else:
return self.query_class(self, state)
- def get_collection(self, state, dict_, user_data=None,
+ def get_collection(self, state, dict_, user_data=None,
passive=attributes.PASSIVE_NO_INITIALIZE):
if passive is not attributes.PASSIVE_OFF:
return self._get_collection_history(state,
if self.key not in state.committed_state:
state.committed_state[self.key] = CollectionHistory(self, state)
- state.modified_event(dict_,
+ state.modified_event(dict_,
self,
attributes.NEVER_SET)
return state.committed_state[self.key]
def set(self, state, dict_, value, initiator,
- passive=attributes.PASSIVE_OFF,
+ passive=attributes.PASSIVE_OFF,
check_old=None, pop=False):
if initiator and initiator.parent_token is self.parent_token:
return
def get_all_pending(self, state, dict_):
c = self._get_collection_history(state, True)
return [
- (attributes.instance_state(x), x)
- for x in
+ (attributes.instance_state(x), x)
+ for x in
c.added_items + c.unchanged_items + c.deleted_items
]
else:
return c
- def append(self, state, dict_, value, initiator,
+ def append(self, state, dict_, value, initiator,
passive=attributes.PASSIVE_OFF):
if initiator is not self:
self.fire_append_event(state, dict_, value, initiator)
- def remove(self, state, dict_, value, initiator,
+ def remove(self, state, dict_, value, initiator,
passive=attributes.PASSIVE_OFF):
if initiator is not self:
self.fire_remove_event(state, dict_, value, initiator)
mapper = object_mapper(instance)
prop = mapper._props[self.attr.key]
self._criterion = prop.compare(
- operators.eq,
- instance,
- value_is_parent=True,
+ operators.eq,
+ instance,
+ value_is_parent=True,
alias_secondary=False)
if self.attr.order_by:
def append(self, item):
self.attr.append(
- attributes.instance_state(self.instance),
+ attributes.instance_state(self.instance),
attributes.instance_dict(self.instance), item, None)
def remove(self, item):
self.attr.remove(
- attributes.instance_state(self.instance),
+ attributes.instance_state(self.instance),
attributes.instance_dict(self.instance), item, None)
from test.lib.testing import eq_, ne_
import operator
-from sqlalchemy.orm import dynamic_loader, backref
+from sqlalchemy.orm import dynamic_loader, backref, configure_mappers
from test.lib import testing
-from sqlalchemy import Integer, String, ForeignKey, desc, select, func
+from sqlalchemy import Integer, String, ForeignKey, desc, select, func, exc
from test.lib.schema import Table, Column
from sqlalchemy.orm import mapper, relationship, create_session, Query, attributes
from sqlalchemy.orm.dynamic import AppenderMixin
u = q.filter(User.id==7).first()
self.assert_compile(
- u.addresses.statement,
+ u.addresses.statement,
"SELECT addresses.id, addresses.user_id, addresses.email_address FROM "
"addresses WHERE :param_1 = addresses.user_id",
use_default_dialect=True
)
+ def test_no_uselist_false(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+ mapper(Address, addresses)
+ mapper(User, users, properties={
+ "addresses": relationship(Address, lazy='dynamic', uselist=False)
+ })
+ assert_raises_message(
+ exc.SAWarning,
+ "On relationship User.addresses, 'dynamic' loaders cannot be "
+ "used with many-to-one/one-to-one relationships and/or "
+ "uselist=False.",
+ configure_mappers
+ )
+
+ def test_no_m2o(self):
+ users, Address, addresses, User = (self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User)
+ mapper(Address, addresses, properties={
+ 'user': relationship(User, lazy='dynamic')
+ })
+ mapper(User, users)
+ assert_raises_message(
+ exc.SAWarning,
+ "On relationship Address.user, 'dynamic' loaders cannot be "
+ "used with many-to-one/one-to-one relationships and/or "
+ "uselist=False.",
+ configure_mappers
+ )
+
def test_order_by(self):
users, Address, addresses, User = (self.tables.users,
self.classes.Address,
self.classes.User)
mapper(User, users, properties={
- 'addresses':dynamic_loader(mapper(Address, addresses))
+ 'addresses': dynamic_loader(mapper(Address, addresses))
})
sess = create_session()
u = sess.query(User).get(8)
eq_(
list(u.addresses.order_by(desc(Address.email_address))),
- [Address(email_address=u'ed@wood.com'), Address(email_address=u'ed@lala.com'),
+ [Address(email_address=u'ed@wood.com'), Address(email_address=u'ed@lala.com'),
Address(email_address=u'ed@bettyboop.com')]
)
assert o1 in i1.orders.all()
assert i1 in o1.items.all()
- @testing.exclude('mysql', 'between',
+ @testing.exclude('mysql', 'between',
((5, 1,49), (5, 1, 52)),
'https://bugs.launchpad.net/ubuntu/+source/mysql-5.1/+bug/706988')
def test_association_nonaliased(self):
self.classes.Item)
mapper(Order, orders, properties={
- 'items':relationship(Item, secondary=order_items,
- lazy="dynamic",
+ 'items':relationship(Item, secondary=order_items,
+ lazy="dynamic",
order_by=order_items.c.item_id)
})
mapper(Item, items)
use_default_dialect=True
)
- # filter criterion against the secondary table
+ # filter criterion against the secondary table
# works
eq_(
o.items.filter(order_items.c.item_id==2).all(),
sess.flush()
sess.commit()
u1.addresses.append(Address(email_address='foo@bar.com'))
- eq_(u1.addresses.order_by(Address.id).all(),
+ eq_(u1.addresses.order_by(Address.id).all(),
[Address(email_address='lala@hoho.com'), Address(email_address='foo@bar.com')])
sess.rollback()
eq_(u1.addresses.all(), [Address(email_address='lala@hoho.com')])
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
