--- /dev/null
+.. change::
+ :tags: bug, orm, declarative
+ :tickets: 8705
+
+ Changed a fundamental configuration behavior of :class:`.Mapper`, where
+ :class:`_schema.Column` objects that are explicitly present in the
+ :paramref:`_orm.Mapper.properties` dictionary, either directly or enclosed
+ within a mapper property object, will now be mapped within the order of how
+ they appear within the mapped :class:`.Table` (or other selectable) itself
+ (assuming they are in fact part of that table's list of columns), thereby
+ maintaining the same order of columns in the mapped selectable as is
+ instrumented on the mapped class, as well as what renders in an ORM SELECT
+ statement for that mapper. Previously (where "previously" means since
+ version 0.0.1), :class:`.Column` objects in the
+ :paramref:`_orm.Mapper.properties` dictionary would always be mapped first,
+ ahead of when the other columns in the mapped :class:`.Table` would be
+ mapped, causing a discrepancy in the order in which the mapper would
+ assign attributes to the mapped class as well as the order in which they
+ would render in statements.
+
+ The change most prominently takes place in the way that Declarative
+ assigns declared columns to the :class:`.Mapper`, specifically how
+ :class:`.Column` (or :func:`_orm.mapped_column`) objects are handled
+ when they have a DDL name that is explicitly different from the mapped
+ attribute name, as well as when constructs such as :func:`_orm.deferred`
+ etc. are used. The new behavior will see the column ordering within
+ the mapped :class:`.Table` being the same order in which the attributes
+ are mapped onto the class, assigned within the :class:`.Mapper` itself,
+ and rendered in ORM statements such as SELECT statements, independent
+ of how the :class:`_schema.Column` was configured against the
+ :class:`.Mapper`.
>>> from sqlalchemy.orm import undefer
>>> book = session.scalar(select(Book).where(Book.id == 2).options(undefer(Book.summary)))
- {opensql}SELECT book.summary, book.id, book.owner_id, book.title
+ {opensql}SELECT book.id, book.owner_id, book.title, book.summary
FROM book
WHERE book.id = ?
[...] (2,)
>>> book = session.scalar(
... select(Book).where(Book.id == 2).options(undefer_group("book_attrs"))
... )
- {opensql}SELECT book.summary, book.cover_photo, book.id, book.owner_id, book.title
+ {opensql}SELECT book.id, book.owner_id, book.title, book.summary, book.cover_photo
FROM book
WHERE book.id = ?
[...] (2,)
a wildcard::
>>> book = session.scalar(select(Book).where(Book.id == 3).options(undefer("*")))
- {opensql}SELECT book.summary, book.cover_photo, book.id, book.owner_id, book.title
+ {opensql}SELECT book.id, book.owner_id, book.title, book.summary, book.cover_photo
FROM book
WHERE book.id = ?
[...] (3,)
... .options(undefer("*"))
... .execution_options(populate_existing=True)
... )
- {opensql}SELECT book.summary, book.cover_photo, book.id, book.owner_id, book.title
+ {opensql}SELECT book.id, book.owner_id, book.title, book.summary, book.cover_photo
FROM book
WHERE book.id = ?
[...] (2,)
def _configure_properties(self) -> None:
- # TODO: consider using DedupeColumnCollection
self.columns = self.c = sql_base.ColumnCollection() # type: ignore
# object attribute names mapped to MapperProperty objects
# table columns mapped to MapperProperty
self._columntoproperty = _ColumnMapping(self)
- # load custom properties
+ explicit_col_props_by_column: Dict[
+ KeyedColumnElement[Any], Tuple[str, ColumnProperty[Any]]
+ ] = {}
+ explicit_col_props_by_key: Dict[str, ColumnProperty[Any]] = {}
+
+ # step 1: go through properties that were explicitly passed
+ # in the properties dictionary. For Columns that are local, put them
+ # aside in a separate collection we will reconcile with the Table
+ # that's given. For other properties, set them up in _props now.
if self._init_properties:
- for key, prop in self._init_properties.items():
- self._configure_property(key, prop, False)
+ for key, prop_arg in self._init_properties.items():
+
+ if not isinstance(prop_arg, MapperProperty):
+ possible_col_prop = self._make_prop_from_column(
+ key, prop_arg
+ )
+ else:
+ possible_col_prop = prop_arg
+
+ # issue #8705. if the explicit property is actually a
+ # Column that is local to the local Table, don't set it up
+ # in ._props yet, integrate it into the order given within
+ # the Table.
+ if isinstance(possible_col_prop, properties.ColumnProperty):
+ given_col = possible_col_prop.columns[0]
+ if self.local_table.c.contains_column(given_col):
+ explicit_col_props_by_key[key] = possible_col_prop
+ explicit_col_props_by_column[given_col] = (
+ key,
+ possible_col_prop,
+ )
+ continue
- # pull properties from the inherited mapper if any.
+ self._configure_property(key, possible_col_prop, False)
+
+ # step 2: pull properties from the inherited mapper. reconcile
+ # columns with those which are explicit above. for properties that
+ # are only in the inheriting mapper, set them up as local props
if self.inherits:
- for key, prop in self.inherits._props.items():
- if key not in self._props and not self._should_exclude(
- key, key, local=False, column=None
- ):
- self._adapt_inherited_property(key, prop, False)
+ for key, inherited_prop in self.inherits._props.items():
+ if self._should_exclude(key, key, local=False, column=None):
+ continue
+
+ incoming_prop = explicit_col_props_by_key.get(key)
+ if incoming_prop:
+
+ new_prop = self._reconcile_prop_with_incoming_columns(
+ key,
+ inherited_prop,
+ warn_only=False,
+ incoming_prop=incoming_prop,
+ )
+ explicit_col_props_by_key[key] = new_prop
+ explicit_col_props_by_column[incoming_prop.columns[0]] = (
+ key,
+ new_prop,
+ )
+ elif key not in self._props:
+ self._adapt_inherited_property(key, inherited_prop, False)
+
+ # step 3. Iterate through all columns in the persist selectable.
+ # this includes not only columns in the local table / fromclause,
+ # but also those columns in the superclass table if we are joined
+ # inh or single inh mapper. map these columns as well. additional
+ # reconciliation against inherited columns occurs here also.
- # create properties for each column in the mapped table,
- # for those columns which don't already map to a property
for column in self.persist_selectable.columns:
- if column in self._columntoproperty:
+
+ if column in explicit_col_props_by_column:
+ # column was explicitly passed to properties; configure
+ # it now in the order in which it corresponds to the
+ # Table / selectable
+ key, prop = explicit_col_props_by_column[column]
+ self._configure_property(key, prop, False)
+ continue
+
+ elif column in self._columntoproperty:
continue
column_key = (self.column_prefix or "") + column.key
)
if not isinstance(prop_arg, MapperProperty):
- prop = self._property_from_column(key, prop_arg)
+ prop: MapperProperty[Any] = self._property_from_column(
+ key, prop_arg
+ )
else:
prop = prop_arg
return prop
+ def _make_prop_from_column(
+ self,
+ key: str,
+ column: Union[
+ Sequence[KeyedColumnElement[Any]], KeyedColumnElement[Any]
+ ],
+ ) -> ColumnProperty[Any]:
+
+ columns = util.to_list(column)
+ mapped_column = []
+ for c in columns:
+ mc = self.persist_selectable.corresponding_column(c)
+ if mc is None:
+ mc = self.local_table.corresponding_column(c)
+ if mc is not None:
+ # if the column is in the local table but not the
+ # mapped table, this corresponds to adding a
+ # column after the fact to the local table.
+ # [ticket:1523]
+ self.persist_selectable._refresh_for_new_column(mc)
+ mc = self.persist_selectable.corresponding_column(c)
+ if mc is None:
+ raise sa_exc.ArgumentError(
+ "When configuring property '%s' on %s, "
+ "column '%s' is not represented in the mapper's "
+ "table. Use the `column_property()` function to "
+ "force this column to be mapped as a read-only "
+ "attribute." % (key, self, c)
+ )
+ mapped_column.append(mc)
+ return properties.ColumnProperty(*mapped_column)
+
+ def _reconcile_prop_with_incoming_columns(
+ self,
+ key: str,
+ existing_prop: MapperProperty[Any],
+ warn_only: bool,
+ incoming_prop: Optional[ColumnProperty[Any]] = None,
+ single_column: Optional[KeyedColumnElement[Any]] = None,
+ ) -> ColumnProperty[Any]:
+
+ if incoming_prop and (
+ self.concrete
+ or not isinstance(existing_prop, properties.ColumnProperty)
+ ):
+ return incoming_prop
+
+ existing_column = existing_prop.columns[0]
+
+ if incoming_prop and existing_column in incoming_prop.columns:
+ return incoming_prop
+
+ if incoming_prop is None:
+ assert single_column is not None
+ incoming_column = single_column
+ equated_pair_key = (existing_prop.columns[0], incoming_column)
+ else:
+ assert single_column is None
+ incoming_column = incoming_prop.columns[0]
+ equated_pair_key = (incoming_column, existing_prop.columns[0])
+
+ if (
+ (
+ not self._inherits_equated_pairs
+ or (equated_pair_key not in self._inherits_equated_pairs)
+ )
+ and not existing_column.shares_lineage(incoming_column)
+ and existing_column is not self.version_id_col
+ and incoming_column is not self.version_id_col
+ ):
+ msg = (
+ "Implicitly combining column %s with column "
+ "%s under attribute '%s'. Please configure one "
+ "or more attributes for these same-named columns "
+ "explicitly."
+ % (
+ existing_prop.columns[-1],
+ incoming_column,
+ key,
+ )
+ )
+ if warn_only:
+ util.warn(msg)
+ else:
+ raise sa_exc.InvalidRequestError(msg)
+
+ # existing properties.ColumnProperty from an inheriting
+ # mapper. make a copy and append our column to it
+ # breakpoint()
+ new_prop = existing_prop.copy()
+
+ new_prop.columns.insert(0, incoming_column)
+ self._log(
+ "inserting column to existing list "
+ "in properties.ColumnProperty %s",
+ key,
+ )
+ return new_prop # type: ignore
+
@util.preload_module("sqlalchemy.orm.descriptor_props")
def _property_from_column(
self,
key: str,
- prop_arg: Union[KeyedColumnElement[Any], MapperProperty[Any]],
- ) -> MapperProperty[Any]:
+ column: KeyedColumnElement[Any],
+ ) -> ColumnProperty[Any]:
"""generate/update a :class:`.ColumnProperty` given a
- :class:`_schema.Column` object."""
+ :class:`_schema.Column` or other SQL expression object."""
+
descriptor_props = util.preloaded.orm_descriptor_props
- # we were passed a Column or a list of Columns;
- # generate a properties.ColumnProperty
- columns = util.to_list(prop_arg)
- column = columns[0]
prop = self._props.get(key)
if isinstance(prop, properties.ColumnProperty):
- if (
- (
- not self._inherits_equated_pairs
- or (prop.columns[0], column)
- not in self._inherits_equated_pairs
- )
- and not prop.columns[0].shares_lineage(column)
- and prop.columns[0] is not self.version_id_col
- and column is not self.version_id_col
- ):
- warn_only = prop.parent is not self
- msg = (
- "Implicitly combining column %s with column "
- "%s under attribute '%s'. Please configure one "
- "or more attributes for these same-named columns "
- "explicitly." % (prop.columns[-1], column, key)
- )
- if warn_only:
- util.warn(msg)
- else:
- raise sa_exc.InvalidRequestError(msg)
-
- # existing properties.ColumnProperty from an inheriting
- # mapper. make a copy and append our column to it
- prop = prop.copy()
- prop.columns.insert(0, column)
- self._log(
- "inserting column to existing list "
- "in properties.ColumnProperty %s" % (key)
+ return self._reconcile_prop_with_incoming_columns(
+ key,
+ prop,
+ single_column=column,
+ warn_only=prop.parent is not self,
)
- return prop
elif prop is None or isinstance(
prop, descriptor_props.ConcreteInheritedProperty
):
- mapped_column = []
- for c in columns:
- mc = self.persist_selectable.corresponding_column(c)
- if mc is None:
- mc = self.local_table.corresponding_column(c)
- if mc is not None:
- # if the column is in the local table but not the
- # mapped table, this corresponds to adding a
- # column after the fact to the local table.
- # [ticket:1523]
- self.persist_selectable._refresh_for_new_column(mc)
- mc = self.persist_selectable.corresponding_column(c)
- if mc is None:
- raise sa_exc.ArgumentError(
- "When configuring property '%s' on %s, "
- "column '%s' is not represented in the mapper's "
- "table. Use the `column_property()` function to "
- "force this column to be mapped as a read-only "
- "attribute." % (key, self, c)
- )
- mapped_column.append(mc)
- return properties.ColumnProperty(*mapped_column)
+ return self._make_prop_from_column(key, column)
else:
raise sa_exc.ArgumentError(
"WARNING: when configuring property '%s' on %s, "
session = Session()
self.assert_compile(
session.query(Document),
- "SELECT pjoin.documenttype AS pjoin_documenttype, "
- "pjoin.id AS pjoin_id, pjoin.type AS pjoin_type FROM "
+ "SELECT "
+ "pjoin.id AS pjoin_id, pjoin.documenttype AS pjoin_documenttype, "
+ "pjoin.type AS pjoin_type FROM "
"(SELECT offers.id AS id, offers.documenttype AS documenttype, "
"'offer' AS type FROM offers) AS pjoin",
)
sess = fixture_session()
self.assert_compile(
sess.query(aliased(A)).filter_by(value="foo"),
- "SELECT a_1.value AS a_1_value, a_1.id AS a_1_id "
+ "SELECT a_1.id AS a_1_id, a_1.value AS a_1_value "
"FROM a AS a_1 WHERE upper(a_1.value) = upper(:upper_1)",
)
sess = fixture_session()
self.assert_compile(
sess.query(A).filter_by(value="foo"),
- "SELECT a.value AS a_value, a.id AS a_id "
+ "SELECT a.id AS a_id, a.value AS a_value "
"FROM a WHERE foo(a.value) + bar(a.value) = :param_1",
)
sess = fixture_session()
self.assert_compile(
sess.query(aliased(A)).filter_by(value="foo"),
- "SELECT a_1.value AS a_1_value, a_1.id AS a_1_id "
+ "SELECT a_1.id AS a_1_id, a_1.value AS a_1_value "
"FROM a AS a_1 WHERE foo(a_1.value) + bar(a_1.value) = :param_1",
)
sess = fixture_session()
self.assert_compile(
sess.query(A).filter(A.value(5) == "foo"),
- "SELECT a.value AS a_value, a.id AS a_id "
+ "SELECT a.id AS a_id, a.value AS a_value "
"FROM a WHERE foo(a.value, :foo_1) + :foo_2 = :param_1",
)
a1 = aliased(A)
self.assert_compile(
sess.query(a1).filter(a1.value(5) == "foo"),
- "SELECT a_1.value AS a_1_value, a_1.id AS a_1_id "
+ "SELECT a_1.id AS a_1_id, a_1.value AS a_1_value "
"FROM a AS a_1 WHERE foo(a_1.value, :foo_1) + :foo_2 = :param_1",
)
self.assert_compile(
query,
- "SELECT bank_account.balance AS bank_account_balance, "
- "bank_account.id AS bank_account_id FROM bank_account "
+ "SELECT bank_account.id AS bank_account_id, "
+ "bank_account.balance AS bank_account_balance "
+ "FROM bank_account "
"WHERE bank_account.balance = :balance_1",
checkparams={"balance_1": Decimal("9886.110000")},
)
)
self.assert_compile(
query,
- "SELECT bank_account.balance AS bank_account_balance, "
- "bank_account.id AS bank_account_id "
+ "SELECT bank_account.id AS bank_account_id, "
+ "bank_account.balance AS bank_account_balance "
"FROM bank_account "
"WHERE :balance_1 * bank_account.balance > :param_1 "
"AND :balance_2 * bank_account.balance < :param_2",
)
self.assert_compile(
query,
- "SELECT bank_account.balance AS bank_account_balance, "
- "bank_account.id AS bank_account_id FROM bank_account "
+ "SELECT bank_account.id AS bank_account_id, "
+ "bank_account.balance AS bank_account_balance "
+ "FROM bank_account "
"WHERE :balance_1 * bank_account.balance > "
":param_1 * :balance_2 * bank_account.balance",
checkparams={
# note in the original, the same bound parameter is used twice
self.assert_compile(
expr,
- "SELECT test.some_id AS test_some_id, "
+ "SELECT "
"CAST(left(test.some_id, :left_1) AS INTEGER) AS anon_1, "
- "test.id AS test_id FROM test WHERE "
+ "test.id AS test_id, test.some_id AS test_some_id FROM test WHERE "
"CAST(left(test.some_id, :left_1) AS INTEGER) = :param_1",
checkparams={"left_1": 6, "param_1": 123456},
)
# the same value however
self.assert_compile(
expr2,
- "SELECT test.some_id AS test_some_id, "
- "CAST(left(test.some_id, :left_1) AS INTEGER) AS anon_1, "
- "test.id AS test_id FROM test WHERE "
+ "SELECT CAST(left(test.some_id, :left_1) AS INTEGER) AS anon_1, "
+ "test.id AS test_id, test.some_id AS test_some_id FROM test WHERE "
"CAST(left(test.some_id, :left_2) AS INTEGER) = :param_1",
checkparams={"left_1": 6, "left_2": 6, "param_1": 123456},
)
+import random
+
import sqlalchemy as sa
from sqlalchemy import CheckConstraint
from sqlalchemy import event
from sqlalchemy import Index
from sqlalchemy import inspect
from sqlalchemy import Integer
+from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy import UniqueConstraint
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy.orm import joinedload
from sqlalchemy.orm import mapped_column
+from sqlalchemy.orm import MappedColumn
from sqlalchemy.orm import Mapper
from sqlalchemy.orm import registry
from sqlalchemy.orm import relationship
mt = MyTable(id=5)
eq_(mt.id, 5)
+
+
+class NamedAttrOrderingTest(fixtures.TestBase):
+ """test for #8705"""
+
+ @testing.combinations(
+ "decl_base_fn",
+ "decl_base_base",
+ "classical_mapping",
+ argnames="mapping_style",
+ )
+ def test_ordering_of_attrs_cols_named_or_unnamed(self, mapping_style):
+ def make_name():
+ uppercase = random.randint(1, 3) == 1
+ name = "".join(
+ random.choice("abcdefghijklmnopqrstuvxyz")
+ for i in range(random.randint(4, 10))
+ )
+ if uppercase:
+ name = random.choice("ABCDEFGHIJKLMNOP") + name
+ return name
+
+ def make_column(assign_col_name):
+ use_key = random.randint(1, 3) == 1
+ use_name = random.randint(1, 3) == 1
+
+ args = []
+ kw = {}
+ name = col_name = make_name()
+
+ if use_name:
+ use_different_name = random.randint(1, 3) != 3
+ if use_different_name:
+ col_name = make_name()
+
+ args.append(col_name)
+ elif assign_col_name:
+ args.append(col_name)
+
+ if use_key:
+ kw["key"] = name
+ expected_c_name = name
+ else:
+ expected_c_name = col_name
+
+ args.append(Integer)
+
+ if mapping_style.startswith("decl"):
+ use_mapped_column = random.randint(1, 2) == 1
+ else:
+ use_mapped_column = False
+
+ if use_mapped_column:
+ col = mapped_column(*args, **kw)
+ else:
+ col = Column(*args, **kw)
+
+ use_explicit_property = (
+ not use_mapped_column and random.randint(1, 6) == 1
+ )
+ if use_explicit_property:
+ col_prop = column_property(col)
+ else:
+ col_prop = col
+
+ return name, expected_c_name, col, col_prop
+
+ assign_col_name = mapping_style == "classical_mapping"
+
+ names = [
+ make_column(assign_col_name) for i in range(random.randint(10, 15))
+ ]
+ len_names = len(names)
+
+ pk_col = names[random.randint(0, len_names - 1)][2]
+ if isinstance(pk_col, MappedColumn):
+ pk_col.column.primary_key = True
+ else:
+ pk_col.primary_key = True
+
+ names_only = [name for name, _, _, _ in names]
+ col_names_only = [col_name for _, col_name, _, _ in names]
+ cols_only = [col for _, _, col, _ in names]
+
+ if mapping_style in ("decl_base_fn", "decl_base_base"):
+ if mapping_style == "decl_base_fn":
+ Base = declarative_base()
+ elif mapping_style == "decl_base_base":
+
+ class Base(DeclarativeBase):
+ pass
+
+ else:
+ assert False
+
+ clsdict = {
+ "__tablename__": "new_table",
+ }
+ clsdict.update({name: colprop for name, _, _, colprop in names})
+
+ new_cls = type("NewCls", (Base,), clsdict)
+
+ elif mapping_style == "classical_mapping":
+
+ class new_cls:
+ pass
+
+ reg = registry()
+ t = Table("new_table", reg.metadata, *cols_only)
+
+ reg.map_imperatively(
+ new_cls,
+ t,
+ properties={
+ key: colprop
+ for key, col_name, col, colprop in names
+ if col_name != key
+ },
+ )
+ else:
+ assert False
+
+ eq_(new_cls.__table__.c.keys(), col_names_only)
+ eq_(new_cls.__mapper__.attrs.keys(), names_only)
+ eq_(list(new_cls._sa_class_manager.keys()), names_only)
+ eq_([k for k in new_cls.__dict__ if not k.startswith("_")], names_only)
+
+ stmt = select(new_cls)
+ eq_(stmt.selected_columns.keys(), col_names_only)
s = fixture_session()
self.assert_compile(
s.query(A),
- "SELECT a.x + :x_1 AS anon_1, a.x AS a_x, a.id AS a_id FROM a",
+ "SELECT a.x + :x_1 AS anon_1, a.id AS a_id, a.x AS a_x FROM a",
)
self.assert_compile(
s.query(B),
- "SELECT b.x + :x_1 AS anon_1, b.x AS b_x, b.id AS b_id FROM b",
+ "SELECT b.x + :x_1 AS anon_1, b.id AS b_id, b.x AS b_x FROM b",
)
@testing.requires.predictable_gc
self.assert_compile(select(User), "SELECT users.id FROM users")
self.assert_compile(
select(User).options(undefer(User.data)),
- "SELECT users.data, users.id FROM users",
+ "SELECT users.id, users.data FROM users",
)
def test_deferred_kw(self, decl_base):
self.assert_compile(select(User), "SELECT users.id FROM users")
self.assert_compile(
select(User).options(undefer(User.data)),
- "SELECT users.data, users.id FROM users",
+ "SELECT users.id, users.data FROM users",
)
@testing.combinations(
to_assert = [
# refresh john
CompiledSQL(
- "SELECT users.age_int AS users_age_int, "
- "users.id AS users_id, users.name AS users_name FROM users "
+ "SELECT users.id AS users_id, users.name AS users_name, "
+ "users.age_int AS users_age_int FROM users "
"WHERE users.id = :pk_1",
[{"pk_1": 1}],
),
# refresh jill
CompiledSQL(
- "SELECT users.age_int AS users_age_int, "
- "users.id AS users_id, users.name AS users_name FROM users "
+ "SELECT users.id AS users_id, users.name AS users_name, "
+ "users.age_int AS users_age_int FROM users "
"WHERE users.id = :pk_1",
[{"pk_1": 3}],
),
to_assert.append(
# refresh jane for partial attributes
CompiledSQL(
- "SELECT users.age_int AS users_age_int, "
- "users.name AS users_name FROM users "
+ "SELECT users.name AS users_name, "
+ "users.age_int AS users_age_int FROM users "
"WHERE users.id = :pk_1",
[{"pk_1": 4}],
)
asserter.assert_(
# refresh john
CompiledSQL(
- "SELECT users.age_int AS users_age_int, "
- "users.id AS users_id, users.name AS users_name FROM users "
+ "SELECT users.id AS users_id, users.name AS users_name, "
+ "users.age_int AS users_age_int FROM users "
"WHERE users.id = :pk_1",
[{"pk_1": 1}],
),
# refresh jill
CompiledSQL(
- "SELECT users.age_int AS users_age_int, "
- "users.id AS users_id, users.name AS users_name FROM users "
+ "SELECT users.id AS users_id, users.name AS users_name, "
+ "users.age_int AS users_age_int FROM users "
"WHERE users.id = :pk_1",
[{"pk_1": 3}],
),
from sqlalchemy.testing import assert_raises
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
# PK col
assert s2.id == s2.base_id != 15
+ def test_subclass_renames_superclass_col_single_inh(self, decl_base):
+ """tested as part of #8705.
+
+ The step where we configure columns mapped to specific keys must
+ take place even if the given column is already in _columntoproperty,
+ as would be the case if the superclass maps that column already.
+
+ """
+
+ class A(decl_base):
+ __tablename__ = "a"
+
+ id = Column(Integer, primary_key=True)
+ a_data = Column(String)
+
+ class B(A):
+ b_data = column_property(A.__table__.c.a_data)
+
+ is_(A.a_data.property.columns[0], A.__table__.c.a_data)
+ is_(B.a_data.property.columns[0], A.__table__.c.a_data)
+ is_(B.b_data.property.columns[0], A.__table__.c.a_data)
+
+ def test_column_setup_sanity_check(self, decl_base):
+ class A(decl_base):
+ __tablename__ = "a"
+
+ id = Column(Integer, primary_key=True)
+ a_data = Column(String)
+
+ class B(A):
+ __tablename__ = "b"
+ id = Column(Integer, ForeignKey("a.id"), primary_key=True)
+ b_data = Column(String)
+
+ is_(A.id.property.parent, inspect(A))
+ # overlapping cols get a new prop on the subclass, with cols merged
+ is_(B.id.property.parent, inspect(B))
+ eq_(B.id.property.columns, [B.__table__.c.id, A.__table__.c.id])
+
+ # totally independent cols remain w/ parent on the originating
+ # mapper
+ is_(B.a_data.property.parent, inspect(A))
+ is_(B.b_data.property.parent, inspect(B))
+
def test_override_implicit(self):
# this is originally [ticket:1111].
# the pattern here is now disallowed by [ticket:1892]
Base, base, properties={"id": base.c.base_id}
)
- def go():
+ with expect_raises_message(
+ sa_exc.InvalidRequestError,
+ "Implicitly combining column base.base_id with column "
+ "subtable.base_id under attribute 'id'. Please configure one "
+ "or more attributes for these same-named columns explicitly.",
+ ):
self.mapper_registry.map_imperatively(
Sub,
subtable,
properties={"id": subtable.c.base_id},
)
- # Sub mapper compilation needs to detect that "base.c.base_id"
- # is renamed in the inherited mapper as "id", even though
- # it has its own "id" property. It then generates
- # an exception in 0.7 due to the implicit conflict.
- assert_raises(sa_exc.InvalidRequestError, go)
-
def test_pk_fk_different(self):
class Base:
pass
go,
CompiledSQL(
"SELECT base.id AS base_id, sub.id AS sub_id, "
+ "base.data AS base_data, base.type AS base_type, "
"base.counter AS base_counter, "
"sub.subcounter AS sub_subcounter, "
- "base.data AS base_data, base.type AS base_type, "
"sub.sub AS sub_sub, sub.subcounter2 AS sub_subcounter2 "
"FROM base LEFT OUTER JOIN sub ON base.id = sub.id "
"WHERE base.id = :pk_1",
) if autoalias else nullcontext():
self.assert_compile(
q,
- "SELECT people.type AS people_type, engineers.id AS "
+ "SELECT engineers.id AS "
"engineers_id, "
- "people.id AS people_id, "
+ "people.id AS people_id, people.type AS people_type, "
"engineers.primary_language AS engineers_primary_language, "
"engineers.manager_id AS engineers_manager_id, "
- "people_1.type AS people_1_type, "
"managers_1.id AS managers_1_id, "
- "people_1.id AS people_1_id, seen_1.id AS seen_1_id, "
+ "people_1.id AS people_1_id, people_1.type AS people_1_type, "
+ "seen_1.id AS seen_1_id, "
"seen_1.timestamp AS seen_1_timestamp, "
"seen_2.id AS seen_2_id, "
"seen_2.timestamp AS seen_2_timestamp "
self.assert_compile(
q,
"SELECT anon_1.id "
- "FROM (SELECT users.name AS name, "
- "users.id AS id FROM users) AS anon_1",
+ "FROM (SELECT users.id AS id, users.name AS name "
+ "FROM users) AS anon_1",
)
# testing deferred opts separately for deterministic SQL generation
self.assert_compile(
q,
- "SELECT anon_1.name, anon_1.id "
- "FROM (SELECT users.name AS name, "
- "users.id AS id FROM users) AS anon_1",
+ "SELECT anon_1.id, anon_1.name "
+ "FROM (SELECT users.id AS id, users.name AS name "
+ "FROM users) AS anon_1",
)
q = select(u1).options(undefer(u1.name_upper))
self.assert_compile(
q,
"SELECT upper(anon_1.name) AS upper_1, anon_1.id "
- "FROM (SELECT users.name AS name, "
- "users.id AS id FROM users) AS anon_1",
+ "FROM (SELECT users.id AS id, users.name AS name "
+ "FROM users) AS anon_1",
)
def test_non_deferred_subq_one(self, non_deferred_fixture):
self.assert_compile(
stmt,
- "WITH RECURSIVE anon_1(name, id) AS "
- "(SELECT users.name AS name, users.id AS id FROM users "
- "UNION ALL SELECT users.name AS name, users.id AS id "
+ "WITH RECURSIVE anon_1(id, name) AS "
+ "(SELECT users.id AS id, users.name AS name FROM users "
+ "UNION ALL SELECT users.id AS id, users.name AS name "
"FROM users JOIN anon_1 ON anon_1.id = users.id) "
"SELECT users.id FROM users JOIN anon_1 ON users.id = anon_1.id",
)
# testing deferred opts separately for deterministic SQL generation
self.assert_compile(
stmt.options(undefer(User.name_upper)),
- "WITH RECURSIVE anon_1(name, id) AS "
- "(SELECT users.name AS name, users.id AS id FROM users "
- "UNION ALL SELECT users.name AS name, users.id AS id "
+ "WITH RECURSIVE anon_1(id, name) AS "
+ "(SELECT users.id AS id, users.name AS name FROM users "
+ "UNION ALL SELECT users.id AS id, users.name AS name "
"FROM users JOIN anon_1 ON anon_1.id = users.id) "
"SELECT upper(users.name) AS upper_1, users.id "
"FROM users JOIN anon_1 ON users.id = anon_1.id",
self.assert_compile(
stmt.options(undefer(User.name)),
- "WITH RECURSIVE anon_1(name, id) AS "
- "(SELECT users.name AS name, users.id AS id FROM users "
- "UNION ALL SELECT users.name AS name, users.id AS id "
+ "WITH RECURSIVE anon_1(id, name) AS "
+ "(SELECT users.id AS id, users.name AS name FROM users "
+ "UNION ALL SELECT users.id AS id, users.name AS name "
"FROM users JOIN anon_1 ON anon_1.id = users.id) "
- "SELECT users.name, users.id "
+ "SELECT users.id, users.name "
"FROM users JOIN anon_1 ON users.id = anon_1.id",
)
self.assert_compile(
select(u_alias),
- "SELECT anon_1.id FROM ((SELECT users.name AS name, "
- "users.id AS id FROM users "
- "WHERE users.id = :id_1 UNION SELECT users.name AS name, "
- "users.id AS id "
+ "SELECT anon_1.id FROM ((SELECT users.id AS id, "
+ "users.name AS name "
+ "FROM users "
+ "WHERE users.id = :id_1 UNION SELECT users.id AS id, "
+ "users.name AS name "
"FROM users WHERE users.id = :id_2) "
- "UNION SELECT users.name AS name, users.id AS id "
+ "UNION SELECT users.id AS id, users.name AS name "
"FROM users WHERE users.id = :id_3) AS anon_1",
)
self.assert_compile(
select(u_alias).options(undefer(u_alias.name)),
- "SELECT anon_1.name, anon_1.id FROM "
- "((SELECT users.name AS name, users.id AS id FROM users "
- "WHERE users.id = :id_1 UNION SELECT users.name AS name, "
- "users.id AS id "
+ "SELECT anon_1.id, anon_1.name FROM "
+ "((SELECT users.id AS id, users.name AS name FROM users "
+ "WHERE users.id = :id_1 UNION SELECT users.id AS id, "
+ "users.name AS name "
"FROM users WHERE users.id = :id_2) "
- "UNION SELECT users.name AS name, users.id AS id "
+ "UNION SELECT users.id AS id, users.name AS name "
"FROM users WHERE users.id = :id_3) AS anon_1",
)
)
self.assert_compile(
select(Order).options(undefer_group("g1")),
- "SELECT orders.isopen, orders.description, orders.id, "
- "orders.user_id, orders.address_id FROM orders",
+ "SELECT orders.id, orders.user_id, orders.address_id, "
+ "orders.isopen, orders.description "
+ "FROM orders",
)
else:
self.assert_compile(
go,
[
(
- "SELECT orders.user_id AS orders_user_id, "
+ "SELECT orders.id AS orders_id, "
+ "orders.user_id AS orders_user_id, "
+ "orders.address_id AS orders_address_id, "
"orders.description AS orders_description, "
- "orders.isopen AS orders_isopen, "
- "orders.id AS orders_id, "
- "orders.address_id AS orders_address_id "
+ "orders.isopen AS orders_isopen "
"FROM orders ORDER BY orders.id",
{},
)
go,
[
(
- "SELECT orders.user_id AS orders_user_id, "
- "orders.description AS orders_description, "
- "orders.isopen AS orders_isopen, "
+ "SELECT "
"orders.id AS orders_id, "
- "orders.address_id AS orders_address_id "
+ "orders.user_id AS orders_user_id, "
+ "orders.address_id AS orders_address_id, "
+ "orders.description AS orders_description, "
+ "orders.isopen AS orders_isopen "
"FROM orders ORDER BY orders.id",
{},
)
go,
[
(
- "SELECT orders.user_id AS orders_user_id, "
- "orders.description AS orders_description, "
- "orders.isopen AS orders_isopen, "
+ "SELECT "
"orders.id AS orders_id, "
- "orders.address_id AS orders_address_id "
+ "orders.user_id AS orders_user_id, "
+ "orders.address_id AS orders_address_id, "
+ "orders.description AS orders_description, "
+ "orders.isopen AS orders_isopen "
"FROM orders ORDER BY orders.id",
{},
)
{"id_1": 7},
),
(
- "SELECT orders.user_id AS orders_user_id, "
- "orders.description "
- "AS orders_description, orders.isopen AS orders_isopen, "
- "orders.id AS orders_id, orders.address_id "
- "AS orders_address_id "
+ "SELECT orders.id AS orders_id, "
+ "orders.user_id AS orders_user_id, "
+ "orders.address_id AS orders_address_id, "
+ "orders.description AS orders_description, "
+ "orders.isopen AS orders_isopen "
"FROM orders WHERE :param_1 = orders.user_id "
"ORDER BY orders.id",
{"param_1": 7},
{"id_1": 7},
),
(
- "SELECT orders.user_id AS orders_user_id, "
- "orders.description "
- "AS orders_description, orders.isopen AS orders_isopen, "
+ "SELECT "
"orders.id AS orders_id, "
+ "orders.user_id AS orders_user_id, "
"orders.address_id AS orders_address_id, "
+ "orders.description AS orders_description, "
+ "orders.isopen AS orders_isopen, "
"anon_1.users_id AS anon_1_users_id "
"FROM (SELECT users.id AS "
"users_id FROM users WHERE users.id = :id_1) AS anon_1 "
[
(
"SELECT users.id AS users_id, users.name AS users_name, "
+ "orders_1.id AS orders_1_id, "
"orders_1.user_id AS orders_1_user_id, "
+ "orders_1.address_id AS orders_1_address_id, "
"orders_1.description AS orders_1_description, "
- "orders_1.isopen AS orders_1_isopen, "
- "orders_1.id AS orders_1_id, orders_1.address_id AS "
- "orders_1_address_id FROM users "
+ "orders_1.isopen AS orders_1_isopen "
+ "FROM users "
"LEFT OUTER JOIN orders AS orders_1 ON users.id = "
"orders_1.user_id WHERE users.id = :id_1 "
"ORDER BY orders_1.id",
[
(
"SELECT users.id AS users_id, users.name AS users_name, "
- "orders_1.user_id AS orders_1_user_id, "
"lower(orders_1.description) AS lower_1, "
- "orders_1.isopen AS orders_1_isopen, "
"orders_1.id AS orders_1_id, "
+ "orders_1.user_id AS orders_1_user_id, "
"orders_1.address_id AS orders_1_address_id, "
- "orders_1.description AS orders_1_description FROM users "
+ "orders_1.description AS orders_1_description, "
+ "orders_1.isopen AS orders_1_isopen "
+ "FROM users "
"LEFT OUTER JOIN orders AS orders_1 ON users.id = "
"orders_1.user_id WHERE users.id = :id_1 "
"ORDER BY orders_1.id",
q = sess.query(Order).options(Load(Order).undefer("*"))
self.assert_compile(
q,
- "SELECT orders.user_id AS orders_user_id, "
- "orders.description AS orders_description, "
- "orders.isopen AS orders_isopen, "
+ "SELECT "
"orders.id AS orders_id, "
- "orders.address_id AS orders_address_id "
+ "orders.user_id AS orders_user_id, "
+ "orders.address_id AS orders_address_id, "
+ "orders.description AS orders_description, "
+ "orders.isopen AS orders_isopen "
"FROM orders",
)
)
self.assert_compile(
q,
- "SELECT orders.description AS orders_description, "
- "orders.id AS orders_id, "
- "orders.user_id AS orders_user_id, "
+ "SELECT orders.id AS orders_id, orders.user_id AS orders_user_id, "
+ "orders.description AS orders_description, "
"orders.isopen AS orders_isopen FROM orders",
)
self.assert_compile(
q,
"SELECT "
- "addresses_1.id AS addresses_1_id, "
"users_1.id AS users_1_id, "
"users_1.name AS users_1_name, "
+ "addresses_1.id AS addresses_1_id, "
"addresses_1.user_id AS addresses_1_user_id, "
"addresses_1.email_address AS "
"addresses_1_email_address, "
self.assert_compile(
q,
"SELECT "
- "addresses_1.id AS addresses_1_id, "
"users_1.id AS users_1_id, "
"users_1.name AS users_1_name, "
+ "addresses_1.id AS addresses_1_id, "
"addresses_1.user_id AS addresses_1_user_id, "
"addresses_1.email_address AS "
"addresses_1_email_address, "
self.assert_compile(
stmt,
"SELECT anon_1.users_id AS anon_1_users_id FROM "
- "(SELECT users.name AS users_name, users.id AS users_id "
+ "(SELECT users.id AS users_id, users.name AS users_name "
"FROM users WHERE users.id = :id_1 "
"UNION "
- "SELECT users.name AS users_name, users.id AS users_id FROM users "
+ "SELECT users.id AS users_id, users.name AS users_name FROM users "
"WHERE users.id = :id_2) AS anon_1 ORDER BY anon_1.users_id",
)
stmt = s1.union(s2).options(undefer(User.name)).order_by(User.id)
self.assert_compile(
stmt,
- "SELECT anon_1.users_name AS anon_1_users_name, "
- "anon_1.users_id AS anon_1_users_id FROM "
- "(SELECT users.name AS users_name, users.id AS users_id "
+ "SELECT anon_1.users_id AS anon_1_users_id, "
+ "anon_1.users_name AS anon_1_users_name "
+ "FROM "
+ "(SELECT users.id AS users_id, users.name AS users_name "
"FROM users WHERE users.id = :id_1 "
"UNION "
- "SELECT users.name AS users_name, users.id AS users_id "
+ "SELECT users.id AS users_id, users.name AS users_name "
"FROM users WHERE users.id = :id_2) AS anon_1 "
"ORDER BY anon_1.users_id",
)
stmt,
"SELECT anon_1.anon_2_users_id AS anon_1_anon_2_users_id "
"FROM ("
- "SELECT anon_2.users_name AS anon_2_users_name, "
- "anon_2.users_id AS anon_2_users_id FROM "
- "(SELECT users.name AS users_name, users.id AS users_id "
+ "SELECT anon_2.users_id AS anon_2_users_id, "
+ "anon_2.users_name AS anon_2_users_name "
+ "FROM "
+ "(SELECT users.id AS users_id, users.name AS users_name "
"FROM users WHERE users.id = :id_1 UNION "
- "SELECT users.name AS users_name, users.id AS users_id "
+ "SELECT users.id AS users_id, users.name AS users_name "
"FROM users WHERE users.id = :id_2) AS anon_2 "
"UNION "
- "SELECT users.name AS users_name, users.id AS users_id FROM users "
+ "SELECT users.id AS users_id, users.name AS users_name FROM users "
"WHERE users.id = :id_3) AS anon_1 "
"ORDER BY anon_1.anon_2_users_id",
)
)
self.assert_compile(
stmt,
- "SELECT anon_1.anon_2_users_name AS anon_1_anon_2_users_name, "
- "anon_1.anon_2_users_id AS anon_1_anon_2_users_id "
+ "SELECT anon_1.anon_2_users_id AS anon_1_anon_2_users_id, "
+ "anon_1.anon_2_users_name AS anon_1_anon_2_users_name "
"FROM ("
- "SELECT anon_2.users_name AS anon_2_users_name, "
- "anon_2.users_id AS anon_2_users_id FROM "
- "(SELECT users.name AS users_name, users.id AS users_id "
+ "SELECT anon_2.users_id AS anon_2_users_id, "
+ "anon_2.users_name AS anon_2_users_name "
+ "FROM "
+ "(SELECT users.id AS users_id, users.name AS users_name "
"FROM users WHERE users.id = :id_1 UNION "
- "SELECT users.name AS users_name, users.id AS users_id "
+ "SELECT users.id AS users_id, users.name AS users_name "
"FROM users WHERE users.id = :id_2) AS anon_2 "
"UNION "
- "SELECT users.name AS users_name, users.id AS users_id FROM users "
+ "SELECT users.id AS users_id, users.name AS users_name FROM users "
"WHERE users.id = :id_3) AS anon_1 "
"ORDER BY anon_1.anon_2_users_id",
)
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
