from __future__ import annotations
import collections
-import itertools
from typing import Any
from typing import Dict
from typing import Literal
# to work with baked query, the parameters may have been
# updated since this query was created, so take these into account
- rows = list(q.params(self.params))
- for k, v in itertools.groupby(rows, lambda x: x[1:]):
- self._data[k].extend(vv[0] for vv in v)
+ data = self._data
+ for row in q.params(self.params):
+ # group plain tuples rather than Row slices, which would
+ # incur Row construction and Row.__eq__ per row
+ tup = row._to_tuple_instance()
+ data[tup[1:]].append(tup[0])
def loader(self, state, dict_, row):
if self._data is None:
"in_expr",
"pk_cols",
"zero_idx",
+ "n_pk",
"child_lookup_cols",
],
)
in_expr = fk_cols[0]
zero_idx = True
- return self.query_info(False, False, in_expr, pk_cols, zero_idx, None)
+ return self.query_info(
+ False, False, in_expr, pk_cols, zero_idx, len(pk_cols), None
+ )
def _init_for_omit_join_m2o(self):
pk_cols = self.mapper.primary_key
lookup_cols = [lazyloader._equated_columns[pk] for pk in pk_cols]
return self.query_info(
- True, False, in_expr, pk_cols, zero_idx, lookup_cols
+ True, False, in_expr, pk_cols, zero_idx, len(pk_cols), lookup_cols
)
def _init_for_join(self):
else:
in_expr = pk_cols[0]
zero_idx = True
- return self.query_info(False, True, in_expr, pk_cols, zero_idx, None)
+ return self.query_info(
+ False, True, in_expr, pk_cols, zero_idx, len(pk_cols), None
+ )
def init_class_attribute(self, mapper):
self.parent_property._get_strategy(
mapper = self.parent
+ # attribute keys for the lookup columns; when these are
+ # present in a state's dict, reading them directly is
+ # equivalent to the PASSIVE_NO_FETCH attribute lookup below.
+ # whether or not a key is present can vary per state, e.g.
+ # individual instances may have the attribute expired or
+ # deferred, so this is determined state-by-state
+ get_related_ident = mapper._state_ident_getter(
+ query_info.child_lookup_cols,
+ passive=attributes.PASSIVE_NO_FETCH,
+ )
+
for state, overwrite in states:
state_dict = state.dict
- related_ident = tuple(
- mapper._get_state_attr_by_column(
- state,
- state_dict,
- lk,
- passive=attributes.PASSIVE_NO_FETCH,
- )
- for lk in query_info.child_lookup_cols
- )
+ related_ident = get_related_ident(state, state_dict)
# if the loaded parent objects do not have the foreign key
# to the related item loaded, then degrade into the joined
# version of selectinload
]
in_expr = effective_entity._adapt_element(in_expr)
- bundle_ent = orm_util.Bundle("pk", *pk_cols)
- bundle_sql = bundle_ent.__clause_element__()
-
entity_sql = effective_entity.__clause_element__()
q = Select._create_raw_select(
- _raw_columns=[bundle_sql, entity_sql],
+ _raw_columns=[*pk_cols, entity_sql],
_compile_options=_ORMCompileState.default_compile_options,
_propagate_attrs={
"compile_state_plugin": "orm",
entity_sql, self.parent_property._join_condition.secondaryjoin
)
elif not query_info.load_with_join:
- # the Bundle we have in the "omit_join" case is against raw, non
- # annotated columns, so to ensure the Query knows its primary
- # entity, we add it explicitly. If we made the Bundle against
- # annotated columns, we hit a performance issue in this specific
- # case, which is detailed in issue #4347.
+ # the pk columns in the "omit_join" case are raw, non-annotated
+ # columns, so to ensure the Query knows its primary entity, we
+ # add it explicitly. Using annotated columns here would hit a
+ # performance issue detailed in issue #4347.
q = q.select_from(effective_entity)
else:
- # in the non-omit_join case, the Bundle is against the annotated/
- # mapped column of the parent entity, but the #4347 issue does not
- # occur in this case.
+ # in the non-omit_join case, the pk columns are against the
+ # annotated/mapped column of the parent entity, but the #4347
+ # issue does not occur in this case.
q = q.select_from(self._parent_alias).join(
getattr(self._parent_alias, self.parent_property.key).of_type(
effective_entity
chunksize,
):
uselist = self.uselist
+ n_pk = query_info.n_pk
# this sort is really for the benefit of the unit tests
our_keys = sorted(our_states)
while our_keys:
chunk = our_keys[0:chunksize]
our_keys = our_keys[chunksize:]
+ primary_keys = [
+ key[0] if query_info.zero_idx else key for key in chunk
+ ]
result = context.session.execute(
q,
- params={
- "primary_keys": [
- key[0] if query_info.zero_idx else key for key in chunk
- ]
- },
+ params={"primary_keys": primary_keys},
execution_options=execution_options,
)
if result.context is not None and result.context.requires_uniquing:
rows = result.unique()
else:
- # consume the result as plain tuples, skipping per-row
- # Row construction
rows = result._raw_all_tuples()
- data = {k: v for k, v in rows}
+ data = {row[:n_pk]: row[n_pk] for row in rows}
for key in chunk:
# for a real foreign key and no concurrent changes to the
self, our_states, query_info, q, context, execution_options, chunksize
):
uselist = self.uselist
+ n_pk = query_info.n_pk
_empty_result = () if uselist else None
while our_states:
if result.context is not None and result.context.requires_uniquing:
rows = result.unique()
else:
- # consume the result as plain tuples, skipping per-row
- # Row construction
rows = result._raw_all_tuples()
data = collections.defaultdict(list)
- for k, v in itertools.groupby(rows, lambda x: x[0]):
- data[k].extend(vv[1] for vv in v)
+ for row in rows:
+ data[row[:n_pk]].append(row[n_pk])
for key, state, state_dict, overwrite in chunk:
if not overwrite and self.key in state_dict:
eq_(a2.bs, [B2()])
eq_(a1.bs, [B()])
+
+
+class UselistFalseMultipleRowsTest(fixtures.DeclarativeMappedTest):
+ """test the warning emitted by the selectin loader for a
+ uselist=False relationship that matches more than one row"""
+
+ @classmethod
+ def setup_classes(cls):
+ Base = cls.DeclarativeBasic
+
+ class A(ComparableEntity, Base):
+ __tablename__ = "a"
+ id = Column(Integer, primary_key=True)
+ b = relationship("B", uselist=False)
+
+ class B(ComparableEntity, Base):
+ __tablename__ = "b"
+ id = Column(Integer, primary_key=True)
+ a_id = Column(ForeignKey("a.id"))
+
+ @classmethod
+ def insert_data(cls, connection):
+ A, B = cls.classes("A", "B")
+ s = Session(connection)
+ s.add(A(id=1))
+ s.add_all([B(id=1, a_id=1), B(id=2, a_id=1)])
+ s.commit()
+
+ def test_multiple_rows_uselist_false_warns(self):
+ A = self.classes.A
+ s = fixture_session()
+ with testing.expect_warnings(
+ "Multiple rows returned with uselist=False for "
+ "eagerly-loaded attribute 'A.b'"
+ ):
+ a1 = s.query(A).options(selectinload(A.b)).one()
+ assert a1.b is not None
+ assert a1.b.id in (1, 2)
+
+
+class SelectinM2ONoRelatedRowTest(fixtures.DeclarativeMappedTest):
+ """test many-to-one selectinload where the candidate foreign key
+ value matches no related row, or is NULL"""
+
+ @classmethod
+ def setup_classes(cls):
+ Base = cls.DeclarativeBasic
+
+ class A(ComparableEntity, Base):
+ __tablename__ = "a"
+ id = Column(Integer, primary_key=True)
+
+ class B(ComparableEntity, Base):
+ __tablename__ = "b"
+ id = Column(Integer, primary_key=True)
+ a_id = Column(Integer)
+ a = relationship("A", primaryjoin="foreign(B.a_id) == A.id")
+
+ @classmethod
+ def insert_data(cls, connection):
+ A, B = cls.classes("A", "B")
+ s = Session(connection)
+ s.add(A(id=1))
+ s.add_all(
+ [
+ B(id=1, a_id=1),
+ B(id=2, a_id=42),
+ B(id=3, a_id=None),
+ ]
+ )
+ s.commit()
+
+ def test_missing_and_null_fk(self):
+ A, B = self.classes("A", "B")
+ s = fixture_session()
+ bs = s.query(B).order_by(B.id).options(selectinload(B.a)).all()
+
+ with self.assert_statement_count(testing.db, 0):
+ eq_(bs[0].a, A(id=1))
+
+ # a_id value with no matching A row
+ is_(bs[1].a, None)
+
+ # NULL a_id
+ is_(bs[2].a, None)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
