from . import attributes
from . import exc as orm_exc
from . import path_registry
-from . import strategy_options
from .base import _DEFER_FOR_STATE
from .base import _RAISE_FOR_STATE
from .base import _SET_DEFERRED_EXPIRED
attribute_names = attribute_names.intersection(mapper.attrs.keys())
if mapper.inherits and not mapper.concrete:
- # because we are using Core to produce a select() that we
- # pass to the Query, we aren't calling setup() for mapped
- # attributes; in 1.0 this means deferred attrs won't get loaded
- # by default
statement = mapper._optimized_get_statement(state, attribute_names)
if statement is not None:
- # this was previously aliased(mapper, statement), however,
- # statement is a select() and Query's coercion now raises for this
- # since you can't "select" from a "SELECT" statement. only
- # from_statement() allows this.
- # note: using from_statement() here means there is an adaption
- # with adapt_on_names set up. the other option is to make the
- # aliased() against a subquery which affects the SQL.
from .query import FromStatement
- stmt = FromStatement(mapper, statement).options(
- strategy_options.Load(mapper).undefer("*")
- )
+ # undefer() isn't needed here because statement has the
+ # columns needed already, this implicitly undefers that column
+ stmt = FromStatement(mapper, statement)
result = load_on_ident(
session,
from .base import _DEFER_FOR_STATE
from .base import _RAISE_FOR_STATE
from .base import _SET_DEFERRED_EXPIRED
+from .base import PASSIVE_OFF
from .context import _column_descriptions
from .context import ORMCompileState
from .context import QueryContext
if self.raiseload:
self._invoke_raise_load(state, passive, "raise")
- if (
- loading.load_on_ident(
- session,
- sql.select(localparent).set_label_style(
- LABEL_STYLE_TABLENAME_PLUS_COL
- ),
- state.key,
- only_load_props=group,
- refresh_state=state,
- )
- is None
- ):
- raise orm_exc.ObjectDeletedError(state)
+ loading.load_scalar_attributes(
+ state.mapper, state, set(group), PASSIVE_OFF
+ )
return attributes.ATTR_WAS_SET
class OptimizedGetOnDeferredTest(fixtures.MappedTest):
- """test that the 'optimized get' path accommodates deferred columns."""
+ """test that the 'optimized get' path accommodates deferred columns.
+
+ Original issue tested is #3468, where loading of a deferred column
+ in an inherited subclass would fail.
+
+ At some point, the logic tested was no longer used and a less efficient
+ query was used to load these columns, but the test here did not inspect
+ the SQL such that this would be detected.
+
+ Test was then revised to more carefully test and now targets
+ #7463 as well.
+
+ """
@classmethod
def define_tables(cls, metadata):
Column(
"id", Integer, primary_key=True, test_needs_autoincrement=True
),
+ Column("type", String(10)),
)
Table(
"b",
A, B = cls.classes("A", "B")
a, b = cls.tables("a", "b")
- cls.mapper_registry.map_imperatively(A, a)
+ cls.mapper_registry.map_imperatively(A, a, polymorphic_on=a.c.type)
cls.mapper_registry.map_imperatively(
B,
b,
inherits=A,
+ polymorphic_identity="b",
properties={
"data": deferred(b.c.data),
"expr": column_property(b.c.data + "q", deferred=True),
b1 = B(data="x")
sess.add(b1)
sess.flush()
+ b_id = b1.id
- eq_(b1.expr, "xq")
+ with self.sql_execution_asserter(testing.db) as asserter:
+ eq_(b1.expr, "xq")
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT b.data || :data_1 AS anon_1 "
+ "FROM b WHERE :param_1 = b.id",
+ [{"param_1": b_id, "data_1": "q"}],
+ )
+ )
def test_expired_column(self):
A, B = self.classes("A", "B")
b1 = B(data="x")
sess.add(b1)
sess.flush()
+ b_id = b1.id
sess.expire(b1, ["data"])
+ with self.sql_execution_asserter(testing.db) as asserter:
+ eq_(b1.data, "x")
+ # uses efficient statement w/o JOIN to a
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT b.data AS b_data FROM b WHERE :param_1 = b.id",
+ [{"param_1": b_id}],
+ )
+ )
+
+ def test_load_from_unloaded_subclass(self):
+ A, B = self.classes("A", "B")
+ sess = fixture_session()
+ b1 = B(data="x")
+ sess.add(b1)
+ sess.commit()
+ b_id = b1.id
+ sess.close()
+
+ # load polymorphically in terms of A, so that B needs another
+ # SELECT
+ b1 = sess.execute(select(A)).scalar()
+
+ # it's not loaded
+ assert "data" not in b1.__dict__
+
+ # but it loads successfully when requested
+ with self.sql_execution_asserter(testing.db) as asserter:
+ eq_(b1.data, "x")
+
+ # uses efficient statement w/o JOIN to a
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT b.data AS b_data FROM b WHERE :param_1 = b.id",
+ [{"param_1": b_id}],
+ )
+ )
+
+ def test_load_from_expired_subclass(self):
+ A, B = self.classes("A", "B")
+ sess = fixture_session()
+ b1 = B(data="x")
+ sess.add(b1)
+ sess.commit()
+ b_id = b1.id
+ sess.close()
+
+ b1 = sess.execute(select(A)).scalar()
+
+ # it's not loaded
+ assert "data" not in b1.__dict__
+
eq_(b1.data, "x")
+ sess.expire(b1, ["data"])
+
+ with self.sql_execution_asserter(testing.db) as asserter:
+ eq_(b1.data, "x")
+
+ # uses efficient statement w/o JOIN to a
+ asserter.assert_(
+ CompiledSQL(
+ "SELECT b.data AS b_data FROM b WHERE :param_1 = b.id",
+ [{"param_1": b_id}],
+ )
+ )
class JoinedNoFKSortingTest(fixtures.MappedTest):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
