specifically asks for a column element, no longer
misleads with incorrect information about
text() or literal(). [ticket:1863]
+
+ - Dynamic attributes don't support collection
+ population - added an assertion for when
+ set_committed_value() is called, as well as
+ when joinedload() or subqueryload() options
+ are applied to a dynamic attribute, instead
+ of failure / silent failure. [ticket:1864]
- sql
- Changed the scheme used to generate truncated
accepts_scalar_loader = True
uses_objects = False
+ supports_population = True
def delete(self, state, dict_):
"""
uses_objects = False
+ supports_population = True
def __init__(self, class_, key, callable_,
class_manager, copy_function=None,
accepts_scalar_loader = False
uses_objects = True
+ supports_population = True
def __init__(self, class_, key, callable_,
trackparent=False, extension=None, copy_function=None,
"""
accepts_scalar_loader = False
uses_objects = True
+ supports_population = True
def __init__(self, class_, key, callable_,
typecallable=None, trackparent=False, extension=None,
class DynamicAttributeImpl(attributes.AttributeImpl):
uses_objects = True
accepts_scalar_loader = False
-
+ supports_population = False
+
def __init__(self, class_, key, typecallable,
target_mapper, order_by, query_class=None, **kwargs):
- super(DynamicAttributeImpl, self).__init__(class_, key, typecallable, **kwargs)
+ super(DynamicAttributeImpl, self).\
+ __init__(class_, key, typecallable, **kwargs)
self.target_mapper = target_mapper
self.order_by = order_by
if not query_class:
def get(self, state, dict_, passive=False):
if passive:
- return self._get_collection_history(state, passive=True).added_items
+ return self._get_collection_history(state,
+ passive=True).added_items
else:
return self.query_class(self, state)
def get_collection(self, state, dict_, user_data=None, passive=True):
if passive:
- return self._get_collection_history(state, passive=passive).added_items
+ return self._get_collection_history(state,
+ passive=passive).added_items
else:
- history = self._get_collection_history(state, passive=passive)
+ history = self._get_collection_history(state,
+ passive=passive)
return history.added_items + history.unchanged_items
def fire_append_event(self, state, dict_, value, initiator):
dict_[self.key] = True
return state.committed_state[self.key]
- def set(self, state, dict_, value, initiator, passive=attributes.PASSIVE_OFF):
+ def set(self, state, dict_, value, initiator,
+ passive=attributes.PASSIVE_OFF):
if initiator is self:
return
self._set_iterable(state, dict_, value)
def _set_iterable(self, state, dict_, iterable, adapter=None):
-
collection_history = self._modified_event(state, dict_)
new_values = list(iterable)
-
if state.has_identity:
old_collection = list(self.get(state, dict_))
else:
old_collection = []
-
- collections.bulk_replace(new_values, DynCollectionAdapter(self, state, old_collection), DynCollectionAdapter(self, state, new_values))
+ collections.bulk_replace(new_values, DynCollectionAdapter(self,
+ state, old_collection),
+ DynCollectionAdapter(self, state,
+ new_values))
def delete(self, *args, **kwargs):
raise NotImplementedError()
+ def set_committed_value(self, state, dict_, value):
+ raise NotImplementedError("Dynamic attributes don't support "
+ "collection population.")
+
def get_history(self, state, dict_, passive=False):
c = self._get_collection_history(state, passive)
- return attributes.History(c.added_items, c.unchanged_items, c.deleted_items)
+ return attributes.History(c.added_items, c.unchanged_items,
+ c.deleted_items)
def _get_collection_history(self, state, passive=False):
if self.key in state.committed_state:
def __session(self):
sess = object_session(self.instance)
- if sess is not None and self.autoflush and sess.autoflush and self.instance in sess:
+ if sess is not None and self.autoflush and sess.autoflush \
+ and self.instance in sess:
sess.flush()
if not has_identity(self.instance):
return None
deleted = util.IdentitySet(apply_to.deleted_items)
added = apply_to.added_items
coll = AppenderQuery(attr, state).autoflush(False)
- self.unchanged_items = [o for o in util.IdentitySet(coll) if o not in deleted]
+ self.unchanged_items = [o for o in util.IdentitySet(coll)
+ if o not in deleted]
self.added_items = apply_to.added_items
self.deleted_items = apply_to.deleted_items
else:
]
def create_row_processor(self, context, path, mapper, row, adapter):
+ if not self.parent.class_manager[self.key].impl.supports_population:
+ raise sa_exc.InvalidRequestError(
+ "'%s' does not support object "
+ "population - eager loading cannot be applied." %
+ self)
+
path = path + (self.key,)
path = interfaces._reduce_path(path)
**kwargs):
"""Add a left outer join to the statement thats being constructed."""
+
if not context.query._enable_eagerloads:
return
return False
def create_row_processor(self, context, path, mapper, row, adapter):
+ if not self.parent.class_manager[self.key].impl.supports_population:
+ raise sa_exc.InvalidRequestError(
+ "'%s' does not support object "
+ "population - eager loading cannot be applied." %
+ self)
+
path = path + (self.key,)
+
eager_adapter = self._create_eager_adapter(
context,
from sqlalchemy.test.schema import Table, Column
from sqlalchemy.orm import mapper, relationship, create_session, Query, attributes
from sqlalchemy.orm.dynamic import AppenderMixin
-from sqlalchemy.test.testing import eq_, AssertsCompiledSQL
+from sqlalchemy.test.testing import eq_, AssertsCompiledSQL, assert_raises_message
from sqlalchemy.util import function_named
from test.orm import _base, _fixtures
q.filter(User.id==7).all())
self.assert_sql_count(testing.db, go, 2)
+ @testing.resolve_artifact_names
+ def test_no_populate(self):
+ mapper(User, users, properties={
+ 'addresses':dynamic_loader(mapper(Address, addresses))
+ })
+ u1 = User()
+ assert_raises_message(
+ NotImplementedError,
+ "Dynamic attributes don't support collection population.",
+ attributes.set_committed_value, u1, 'addresses', []
+ )
+
@testing.resolve_artifact_names
def test_m2m(self):
mapper(Order, orders, properties={
-"""basic tests of eager loaded attributes"""
+"""tests of joined-eager loaded attributes"""
from sqlalchemy.test.testing import eq_, is_, is_not_
import sqlalchemy as sa
from sqlalchemy.test import testing
-from sqlalchemy.orm import joinedload, deferred, undefer, joinedload_all, backref
-from sqlalchemy import Integer, String, Date, ForeignKey, and_, select, func
+from sqlalchemy.orm import joinedload, deferred, undefer, \
+ joinedload_all, backref
+from sqlalchemy import Integer, String, Date, ForeignKey, and_, select, \
+ func
from sqlalchemy.test.schema import Table, Column
-from sqlalchemy.orm import mapper, relationship, create_session, lazyload, aliased
-from sqlalchemy.test.testing import eq_, assert_raises
+from sqlalchemy.orm import mapper, relationship, create_session, \
+ lazyload, aliased
+from sqlalchemy.test.testing import eq_, assert_raises, \
+ assert_raises_message
from sqlalchemy.test.assertsql import CompiledSQL
from test.orm import _base, _fixtures
from sqlalchemy.util import OrderedDict as odict
)
self.assert_sql_count(testing.db, go, count)
-
+ @testing.resolve_artifact_names
+ def test_disable_dynamic(self):
+ """test no joined option on a dynamic."""
+
+ mapper(User, users, properties={
+ 'addresses':relationship(Address, lazy="dynamic")
+ })
+ mapper(Address, addresses)
+ sess = create_session()
+ assert_raises_message(
+ sa.exc.InvalidRequestError,
+ "User.addresses' does not support object population - eager loading cannot be applied.",
+ sess.query(User).options(joinedload(User.addresses)).first,
+ )
@testing.resolve_artifact_names
def test_many_to_many(self):
from sqlalchemy.test.schema import Table, Column
from sqlalchemy import Integer, String, ForeignKey, bindparam
from sqlalchemy.orm import backref, subqueryload, subqueryload_all, \
- mapper, relationship, clear_mappers,\
- create_session, lazyload, aliased, joinedload,\
- deferred, undefer
-from sqlalchemy.test.testing import eq_, assert_raises
+ mapper, relationship, clear_mappers, create_session, lazyload, \
+ aliased, joinedload, deferred, undefer
+from sqlalchemy.test.testing import eq_, assert_raises, \
+ assert_raises_message
from sqlalchemy.test.assertsql import CompiledSQL
from test.orm import _base, _fixtures
import sqlalchemy as sa
self.assert_sql_count(testing.db, go, 2)
+ @testing.resolve_artifact_names
+ def test_disable_dynamic(self):
+ """test no subquery option on a dynamic."""
+
+ mapper(User, users, properties={
+ 'addresses':relationship(Address, lazy="dynamic")
+ })
+ mapper(Address, addresses)
+ sess = create_session()
+
+ # previously this would not raise, but would emit
+ # the query needlessly and put the result nowhere.
+ assert_raises_message(
+ sa.exc.InvalidRequestError,
+ "User.addresses' does not support object population - eager loading cannot be applied.",
+ sess.query(User).options(subqueryload(User.addresses)).first,
+ )
@testing.resolve_artifact_names
def test_many_to_many(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
