context, path, mapper, result, adapter, populators)
propagate_options = context.propagate_options
- if propagate_options:
- load_path = context.query._current_path + path \
- if context.query._current_path.path else path
+ load_path = context.query._current_path + path \
+ if context.query._current_path.path else path
session_identity_map = context.session.identity_map
# full population routines. Objects here are either
# just created, or we are doing a populate_existing
- if isnew and propagate_options:
+ # be conservative about setting load_path when populate_existing
+ # is in effect; want to maintain options from the original
+ # load. see test_expire->test_refresh_maintains_deferred_options
+ if isnew and (propagate_options or not populate_existing):
state.load_options = propagate_options
state.load_path = load_path
_populate_full(
- context, row, state, dict_, isnew,
+ context, row, state, dict_, isnew, load_path,
loaded_instance, populate_existing, populators)
if isnew:
# and add to the "context.partials" collection.
to_load = _populate_partial(
- context, row, state, dict_, isnew,
+ context, row, state, dict_, isnew, load_path,
unloaded, populators)
if isnew:
def _populate_full(
- context, row, state, dict_, isnew,
+ context, row, state, dict_, isnew, load_path,
loaded_instance, populate_existing, populators):
if isnew:
# first time we are seeing a row with this identity.
populator(state, dict_, row)
for key, populator in populators["delayed"]:
populator(state, dict_, row)
+ elif load_path != state.load_path:
+ # new load path, e.g. object is present in more than one
+ # column position in a series of rows
+ state.load_path = load_path
+
+ # if we have data, and the data isn't in the dict, OK, let's put
+ # it in.
+ for key, getter in populators["quick"]:
+ if key not in dict_:
+ dict_[key] = getter(row)
+
+ # otherwise treat like an "already seen" row
+ for key, populator in populators["existing"]:
+ populator(state, dict_, row)
+ # TODO: allow "existing" populator to know this is
+ # a new path for the state:
+ # populator(state, dict_, row, new_path=True)
+
else:
- # have already seen rows with this identity.
+ # have already seen rows with this identity in this same path.
for key, populator in populators["existing"]:
populator(state, dict_, row)
+ # TODO: same path
+ # populator(state, dict_, row, new_path=False)
+
def _populate_partial(
- context, row, state, dict_, isnew,
+ context, row, state, dict_, isnew, load_path,
unloaded, populators):
+
if not isnew:
to_load = context.partials[state]
for key, populator in populators["existing"]:
)
+class SelfReferentialMultiPathTest(testing.fixtures.DeclarativeMappedTest):
+ """test for [ticket:3822]"""
+
+ @classmethod
+ def setup_classes(cls):
+ Base = cls.DeclarativeBasic
+
+ class Node(Base):
+ __tablename__ = 'node'
+
+ id = sa.Column(sa.Integer, primary_key=True)
+ parent_id = sa.Column(sa.ForeignKey('node.id'))
+ parent = relationship('Node', remote_side=[id])
+ name = sa.Column(sa.String(10))
+
+ @classmethod
+ def insert_data(cls):
+ Node = cls.classes.Node
+
+ session = Session()
+ session.add_all([
+ Node(id=1, name='name'),
+ Node(id=2, parent_id=1, name='name'),
+ Node(id=3, parent_id=1, name='name')
+ ])
+ session.commit()
+
+ def test_present_overrides_deferred(self):
+ Node = self.classes.Node
+
+ session = Session()
+
+ q = session.query(Node).options(
+ joinedload(Node.parent).load_only(Node.id, Node.parent_id)
+ )
+
+ # Node #1 will appear first as Node.parent and have
+ # deferred applied to Node.name. it will then appear
+ # as Node in the last row and "name" should be populated.
+ nodes = q.order_by(Node.id.desc()).all()
+
+ def go():
+ for node in nodes:
+ eq_(node.name, 'name')
+
+ self.assert_sql_count(testing.db, go, 0)
+
+
class InheritanceTest(_Polymorphic):
__dialect__ = 'default'
s.expire(u)
assert len(u.addresses) == 3
+ def test_refresh_maintains_deferred_options(self):
+ # testing a behavior that may have changed with
+ # [ticket:3822]
+ User, Address, Dingaling = self.classes(
+ "User", "Address", "Dingaling")
+ users, addresses, dingalings = self.tables(
+ "users", "addresses", "dingalings")
+
+ mapper(User, users, properties={
+ 'addresses': relationship(Address)
+ })
+
+ mapper(Address, addresses, properties={
+ 'dingalings': relationship(Dingaling)
+ })
+
+ mapper(Dingaling, dingalings)
+
+ s = create_session()
+ q = s.query(User).filter_by(name='fred').options(
+ sa.orm.lazyload('addresses').joinedload("dingalings"))
+
+ u1 = q.one()
+
+ # "addresses" is not present on u1, but when u1.addresses
+ # lazy loads, it should also joinedload dingalings. This is
+ # present in state.load_options and state.load_path. The
+ # refresh operation should not reset these attributes.
+ s.refresh(u1)
+
+ def go():
+ eq_(
+ u1.addresses,
+ [Address(
+ email_address='fred@fred.com',
+ dingalings=[Dingaling(data="ding 2/5")]
+ )]
+ )
+
+ self.assert_sql_count(testing.db, go, 1)
+
def test_refresh2(self):
"""test a hang condition that was occurring on expire/refresh"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
