)
return self._set_column_strategy(
- (key,), {"query_expression": True}, opts={"expression": expression}
+ (key,), {"query_expression": True}, extra_criteria=(expression,)
)
def selectin_polymorphic(self, classes: Iterable[Type[Any]]) -> Self:
attrs: Tuple[_AttrType, ...],
strategy: Optional[_StrategySpec],
opts: Optional[_OptsType] = None,
+ extra_criteria: Optional[Tuple[Any, ...]] = None,
) -> Self:
strategy_key = self._coerce_strat(strategy)
_COLUMN_TOKEN,
opts=opts,
attr_group=attrs,
+ extra_criteria=extra_criteria,
)
return self
attr_group: Optional[_AttrGroupType] = None,
propagate_to_loaders: bool = True,
reconcile_to_other: Optional[bool] = None,
+ extra_criteria: Optional[Tuple[Any, ...]] = None,
) -> Self:
raise NotImplementedError()
found_crit = False
def process(opt: _LoadElement) -> _LoadElement:
- if not opt._extra_criteria:
- return opt
-
nonlocal orig_cache_key, replacement_cache_key, found_crit
found_crit = True
- # avoid generating cache keys for the queries if we don't
- # actually have any extra_criteria options, which is the
- # common case
if orig_cache_key is None or replacement_cache_key is None:
orig_cache_key = orig_query._generate_cache_key()
replacement_cache_key = context.query._generate_cache_key()
)
for crit in opt._extra_criteria
)
+
return opt
+ # avoid generating cache keys for the queries if we don't
+ # actually have any extra_criteria options, which is the
+ # common case
new_context = tuple(
process(value._clone()) if value._extra_criteria else value
for value in self.context
attr_group: Optional[_AttrGroupType] = None,
propagate_to_loaders: bool = True,
reconcile_to_other: Optional[bool] = None,
+ extra_criteria: Optional[Tuple[Any, ...]] = None,
) -> Self:
# for individual strategy that needs to propagate, set the whole
# Load container to also propagate, so that it shows up in
propagate_to_loaders,
attr_group=attr_group,
reconcile_to_other=reconcile_to_other,
+ extra_criteria=extra_criteria,
)
if load_element:
self.context += (load_element,)
propagate_to_loaders,
attr_group=attr_group,
reconcile_to_other=reconcile_to_other,
+ extra_criteria=extra_criteria,
)
else:
load_element = _AttributeStrategyLoad.create(
propagate_to_loaders,
attr_group=attr_group,
reconcile_to_other=reconcile_to_other,
+ extra_criteria=extra_criteria,
)
if load_element:
attr_group=None,
propagate_to_loaders=True,
reconcile_to_other=None,
+ extra_criteria=None,
):
assert attrs is not None
attr = attrs[0]
if opts:
self.local_opts = util.immutabledict(opts)
+ assert extra_criteria is None
+
def options(self, *opts: _AbstractLoad) -> Self:
raise NotImplementedError("Star option does not support sub-options")
return effective_path
- def _init_path(self, path, attr, wildcard_key, attr_group, raiseerr):
+ def _init_path(
+ self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria
+ ):
"""Apply ORM attributes and/or wildcard to an existing path, producing
a new path.
raiseerr: bool = True,
attr_group: Optional[_AttrGroupType] = None,
reconcile_to_other: Optional[bool] = None,
+ extra_criteria: Optional[Tuple[Any, ...]] = None,
) -> _LoadElement:
"""Create a new :class:`._LoadElement` object."""
else:
opt._reconcile_to_other = None
- path = opt._init_path(path, attr, wildcard_key, attr_group, raiseerr)
+ path = opt._init_path(
+ path, attr, wildcard_key, attr_group, raiseerr, extra_criteria
+ )
if not path:
return None # type: ignore
replacement.local_opts = replacement.local_opts.union(
existing.local_opts
)
- replacement._extra_criteria += replacement._extra_criteria
+ replacement._extra_criteria += existing._extra_criteria
return replacement
elif replacement.path.is_token:
# use 'last one wins' logic for wildcard options. this is also
is_class_strategy = False
is_token_strategy = False
- def _init_path(self, path, attr, wildcard_key, attr_group, raiseerr):
+ def _init_path(
+ self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria
+ ):
assert attr is not None
self._of_type = None
self._path_with_polymorphic_path = None
# from an attribute. This appears to have been an artifact of how
# _UnboundLoad / Load interacted together, which was opaque and
# poorly defined.
- self._extra_criteria = attr._extra_criteria
+ if extra_criteria:
+ assert not attr._extra_criteria
+ self._extra_criteria = extra_criteria
+ else:
+ self._extra_criteria = attr._extra_criteria
if getattr(attr, "_of_type", None):
ac = attr._of_type
is_class_strategy = False
is_token_strategy = True
- def _init_path(self, path, attr, wildcard_key, attr_group, raiseerr):
+ def _init_path(
+ self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria
+ ):
# assert isinstance(attr, str) or attr is None
if attr is not None:
default_token = attr.endswith(_DEFAULT_TOKEN)
__visit_name__ = "class_strategy_load_element"
- def _init_path(self, path, attr, wildcard_key, attr_group, raiseerr):
+ def _init_path(
+ self, path, attr, wildcard_key, attr_group, raiseerr, extra_criteria
+ ):
return path
def _prepare_for_compile_state(
import sqlalchemy as sa
from sqlalchemy import Column
from sqlalchemy import column
+from sqlalchemy import ForeignKey
from sqlalchemy import func
from sqlalchemy import inspect
from sqlalchemy import Integer
from sqlalchemy.orm import Load
from sqlalchemy.orm import load_only
from sqlalchemy.orm import Query
+from sqlalchemy.orm import query_expression
from sqlalchemy.orm import relationship
from sqlalchemy.orm import selectinload
from sqlalchemy.orm import Session
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import int_within_variance
from sqlalchemy.testing import ne_
+from sqlalchemy.testing.entities import ComparableMixin
from sqlalchemy.testing.fixtures import DeclarativeMappedTest
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.testing.util import count_cache_key_tuples
int_within_variance(29796, total_size(ck), 0.05)
else:
testing.skip_test("python platform not available")
+
+
+class WithExpresionLoaderOptTest(DeclarativeMappedTest):
+ """test #10570"""
+
+ @classmethod
+ def setup_classes(cls):
+ Base = cls.DeclarativeBasic
+
+ class A(ComparableMixin, Base):
+ __tablename__ = "a"
+
+ id = Column(Integer, primary_key=True)
+ data = Column(String(30))
+ bs = relationship("B")
+
+ class B(ComparableMixin, Base):
+ __tablename__ = "b"
+ id = Column(Integer, primary_key=True)
+ a_id = Column(ForeignKey("a.id"))
+ boolean = query_expression()
+ data = Column(String(30))
+
+ @classmethod
+ def insert_data(cls, connection):
+ A, B = cls.classes("A", "B")
+
+ with Session(connection) as s:
+ s.add(A(bs=[B(data="a"), B(data="b"), B(data="c")]))
+ s.commit()
+
+ @testing.combinations(
+ joinedload, lazyload, defaultload, selectinload, subqueryload
+ )
+ def test_from_opt(self, loadopt):
+ A, B = self.classes("A", "B")
+
+ def go(value):
+ with Session(testing.db) as sess:
+ objects = sess.execute(
+ select(A).options(
+ loadopt(A.bs).options(
+ with_expression(B.boolean, B.data == value)
+ )
+ )
+ ).scalars()
+ if loadopt is joinedload:
+ objects = objects.unique()
+ eq_(
+ objects.all(),
+ [
+ A(
+ bs=[
+ B(data="a", boolean=value == "a"),
+ B(data="b", boolean=value == "b"),
+ B(data="c", boolean=value == "c"),
+ ]
+ )
+ ],
+ )
+
+ go("b")
+ go("c")
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
