from ...orm.base import _mapper_or_none
from ...orm.clsregistry import _resolver
from ...orm.decl_base import _DeferredMapperConfig
-from ...orm.decl_base import _get_immediate_cls_attr
from ...orm.util import polymorphic_union
from ...schema import Table
from ...util import OrderedDict
attribute to :class:`_declarative.ConcreteBase` so that the
virtual discriminator column name can be customized.
+ .. versionchanged:: 1.4.2 The ``_concrete_discriminator_name`` attribute
+ need only be placed on the basemost class to take correct effect for
+ all subclasses. An explicit error message is now raised if the
+ mapped column names conflict with the discriminator name, whereas
+ in the 1.3.x series there would be some warnings and then a non-useful
+ query would be generated.
+
.. seealso::
:class:`.AbstractConcreteBase`
return
discriminator_name = (
- _get_immediate_cls_attr(cls, "_concrete_discriminator_name")
- or "type"
+ getattr(cls, "_concrete_discriminator_name", None) or "type"
)
mappers = list(m.self_and_descendants)
mappers.append(mn)
discriminator_name = (
- _get_immediate_cls_attr(cls, "_concrete_discriminator_name")
- or "type"
+ getattr(cls, "_concrete_discriminator_name", None) or "type"
)
pjoin = cls._create_polymorphic_union(mappers, discriminator_name)
disallow_is_literal=True,
name_is_truncatable=isinstance(name, _truncated_label),
)
- # TODO: want to remove this assertion at some point. all
- # _make_proxy() implementations will give us back the key that
- # is our "name" in the first place. based on this we can
- # safely return our "self.key" as the key here, to support a new
- # case where the key and name are separate.
- assert key == self.name
+
+ # there was a note here to remove this assertion, which was here
+ # to determine if we later could support a use case where
+ # the key and name of a label are separate. But I don't know what
+ # that case was. For now, this is an unexpected case that occurs
+ # when a label name conflicts with other columns and select()
+ # is attempting to disambiguate an explicit label, which is not what
+ # the user would want. See issue #6090.
+ if key != self.name:
+ raise exc.InvalidRequestError(
+ "Label name %s is being renamed to an anonymous label due "
+ "to disambiguation "
+ "which is not supported right now. Please use unique names "
+ "for explicit labels." % (self.name)
+ )
e._propagate_attrs = selectable._propagate_attrs
e._proxies.append(self)
+from sqlalchemy import exc as sa_exc
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
+from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy import testing
from sqlalchemy.ext import declarative as decl
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import mock
+from sqlalchemy.testing.assertions import expect_raises_message
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
class DeclarativeTestBase(fixtures.TestBase, testing.AssertsExecutionResults):
+ __dialect__ = "default"
+
def setup_test(self):
global Base
Base = decl.declarative_base(testing.db)
self._roundtrip(Employee, Manager, Engineer, Boss)
+ def test_concrete_extension_warn_for_overlap(self):
+ class Employee(ConcreteBase, Base, fixtures.ComparableEntity):
+ __tablename__ = "employee"
+
+ employee_id = Column(
+ Integer, primary_key=True, test_needs_autoincrement=True
+ )
+ name = Column(String(50))
+ __mapper_args__ = {
+ "polymorphic_identity": "employee",
+ "concrete": True,
+ }
+
+ class Manager(Employee):
+ __tablename__ = "manager"
+ employee_id = Column(
+ Integer, primary_key=True, test_needs_autoincrement=True
+ )
+ type = Column(String(50))
+ __mapper_args__ = {
+ "polymorphic_identity": "manager",
+ "concrete": True,
+ }
+
+ with expect_raises_message(
+ sa_exc.InvalidRequestError,
+ "Polymorphic union can't use 'type' as the discriminator "
+ "column due to mapped column "
+ r"Column\('type', String\(length=50\), table=<manager>\); "
+ "please "
+ "apply the 'typecolname' argument; this is available on "
+ "ConcreteBase as '_concrete_discriminator_name'",
+ ):
+ configure_mappers()
+
+ def test_concrete_extension_warn_concrete_disc_resolves_overlap(self):
+ class Employee(ConcreteBase, Base, fixtures.ComparableEntity):
+ _concrete_discriminator_name = "_type"
+
+ __tablename__ = "employee"
+
+ employee_id = Column(
+ Integer, primary_key=True, test_needs_autoincrement=True
+ )
+ name = Column(String(50))
+ __mapper_args__ = {
+ "polymorphic_identity": "employee",
+ "concrete": True,
+ }
+
+ class Manager(Employee):
+ __tablename__ = "manager"
+ employee_id = Column(
+ Integer, primary_key=True, test_needs_autoincrement=True
+ )
+ type = Column(String(50))
+ __mapper_args__ = {
+ "polymorphic_identity": "manager",
+ "concrete": True,
+ }
+
+ configure_mappers()
+ self.assert_compile(
+ select(Employee),
+ "SELECT pjoin.employee_id, pjoin.name, pjoin._type, pjoin.type "
+ "FROM (SELECT employee.employee_id AS employee_id, "
+ "employee.name AS name, CAST(NULL AS VARCHAR(50)) AS type, "
+ "'employee' AS _type FROM employee UNION ALL "
+ "SELECT manager.employee_id AS employee_id, "
+ "CAST(NULL AS VARCHAR(50)) AS name, manager.type AS type, "
+ "'manager' AS _type FROM manager) AS pjoin",
+ )
+
+ def test_abs_concrete_extension_warn_for_overlap(self):
+ class Employee(AbstractConcreteBase, Base, fixtures.ComparableEntity):
+ name = Column(String(50))
+ __mapper_args__ = {
+ "polymorphic_identity": "employee",
+ "concrete": True,
+ }
+
+ class Manager(Employee):
+ __tablename__ = "manager"
+ employee_id = Column(
+ Integer, primary_key=True, test_needs_autoincrement=True
+ )
+ type = Column(String(50))
+ __mapper_args__ = {
+ "polymorphic_identity": "manager",
+ "concrete": True,
+ }
+
+ with expect_raises_message(
+ sa_exc.InvalidRequestError,
+ "Polymorphic union can't use 'type' as the discriminator "
+ "column due to mapped column "
+ r"Column\('type', String\(length=50\), table=<manager>\); "
+ "please "
+ "apply the 'typecolname' argument; this is available on "
+ "ConcreteBase as '_concrete_discriminator_name'",
+ ):
+ configure_mappers()
+
+ def test_abs_concrete_extension_warn_concrete_disc_resolves_overlap(self):
+ class Employee(AbstractConcreteBase, Base, fixtures.ComparableEntity):
+ _concrete_discriminator_name = "_type"
+
+ name = Column(String(50))
+ __mapper_args__ = {
+ "polymorphic_identity": "employee",
+ "concrete": True,
+ }
+
+ class Manager(Employee):
+ __tablename__ = "manager"
+ employee_id = Column(
+ Integer, primary_key=True, test_needs_autoincrement=True
+ )
+ type = Column(String(50))
+ __mapper_args__ = {
+ "polymorphic_identity": "manager",
+ "concrete": True,
+ }
+
+ configure_mappers()
+ self.assert_compile(
+ select(Employee),
+ "SELECT pjoin.name, pjoin.employee_id, pjoin.type, pjoin._type "
+ "FROM (SELECT manager.name AS name, manager.employee_id AS "
+ "employee_id, manager.type AS type, 'manager' AS _type "
+ "FROM manager) AS pjoin",
+ )
+
def test_has_inherited_table_doesnt_consider_base(self):
class A(Base):
__tablename__ = "a"
from sqlalchemy.testing import is_
from sqlalchemy.testing import is_not
from sqlalchemy.testing import ne_
+from sqlalchemy.testing.assertions import expect_raises_message
metadata = MetaData()
checkparams={"param_1": 5},
)
+ @testing.combinations((True,), (False,))
+ def test_broken_select_same_named_explicit_cols(self, use_anon):
+ # this is issue #6090. the query is "wrong" and we dont know how
+ # to render this right now.
+ stmt = select(
+ table1.c.col1,
+ table1.c.col2,
+ literal_column("col2").label(None if use_anon else "col2"),
+ ).select_from(table1)
+
+ if use_anon:
+ self.assert_compile(
+ select(stmt.subquery()),
+ "SELECT anon_1.col1, anon_1.col2, anon_1.col2_1 FROM "
+ "(SELECT table1.col1 AS col1, table1.col2 AS col2, "
+ "col2 AS col2_1 FROM table1) AS anon_1",
+ )
+ else:
+ # the keys here are not critical as they are not what was
+ # requested anyway, maybe should raise here also.
+ eq_(stmt.selected_columns.keys(), ["col1", "col2", "col2_1"])
+ with expect_raises_message(
+ exc.InvalidRequestError,
+ "Label name col2 is being renamed to an anonymous "
+ "label due to "
+ "disambiguation which is not supported right now. Please use "
+ "unique names for explicit labels.",
+ ):
+ select(stmt.subquery()).compile()
+
def test_select_label_grouped_still_corresponds(self):
label = select(table1.c.col1).label("foo")
label2 = label.self_group()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
