--- /dev/null
+.. change::
+ :tags: feature, orm
+ :tickets: 8822
+
+ Added a new parameter :paramref:`_orm.mapped_column.use_existing_column` to
+ accommodate the use case of a single-table inheritance mapping that uses
+ the pattern of more than one subclass indicating the same column to take
+ place on the superclass. This pattern was previously possible by using
+ :func:`_orm.declared_attr` in conjunction with locating the existing column
+ in the ``.__table__`` of the superclass, however is now updated to work
+ with :func:`_orm.mapped_column` as well as with pep-484 typing, in a
+ simple and succinct way.
+
+ .. seealso::
+
+ :ref:`orm_inheritance_column_conflicts`
+
+
+
.. _orm_inheritance_column_conflicts:
-Resolving Column Conflicts
-+++++++++++++++++++++++++++
+Resolving Column Conflicts with ``use_existing_column``
++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Note in the previous section that the ``manager_name`` and ``engineer_info`` columns
are "moved up" to be applied to ``Employee.__table__``, as a result of their
.. sourcecode:: text
- sqlalchemy.exc.ArgumentError: Column 'start_date' on class
- <class '__main__.Manager'> conflicts with existing
- column 'employee.start_date'
+
+ sqlalchemy.exc.ArgumentError: Column 'start_date' on class Manager conflicts
+ with existing column 'employee.start_date'. If using Declarative,
+ consider using the use_existing_column parameter of mapped_column() to
+ resolve conflicts.
The above scenario presents an ambiguity to the Declarative mapping system that
-may be resolved by using
-:class:`.declared_attr` to define the :class:`_schema.Column` conditionally,
-taking care to return the **existing column** via the parent ``__table__``
-if it already exists::
+may be resolved by using the :paramref:`_orm.mapped_column.use_existing_column`
+parameter on :func:`_orm.mapped_column`, which instructs :func:`_orm.mapped_column`
+to look on the inheriting superclass present and use the column that's already
+mapped, if already present, else to map a new column::
+
from sqlalchemy import DateTime
- from sqlalchemy.orm import declared_attr
class Employee(Base):
"polymorphic_identity": "engineer",
}
- @declared_attr
- def start_date(cls) -> Mapped[datetime]:
- "Start date column, if not present already."
-
- # the DateTime type is required in the mapped_column
- # at the moment when used inside of a @declared_attr
- return Employee.__table__.c.get(
- "start_date", mapped_column(DateTime) # type: ignore
- )
+ start_date: Mapped[datetime] = mapped_column(use_existing_column=True)
class Manager(Employee):
"polymorphic_identity": "manager",
}
- @declared_attr
- def start_date(cls) -> Mapped[datetime]:
- "Start date column, if not present already."
-
- # the DateTime type is required in the mapped_column
- # at the moment when used inside of a @declared_attr
- return Employee.__table__.c.get(
- "start_date", mapped_column(DateTime) # type: ignore
- )
+ start_date: Mapped[datetime] = mapped_column(use_existing_column=True)
Above, when ``Manager`` is mapped, the ``start_date`` column is
-already present on the ``Employee`` class; by returning the existing
-:class:`_schema.Column` object, the declarative system recognizes that this
-is the same column to be mapped to the two different subclasses separately.
+already present on the ``Employee`` class, having been provided by the
+``Engineer`` mapping already. The :paramref:`_orm.mapped_column.use_existing_column`
+parameter indicates to :func:`_orm.mapped_column` that it should look for the
+requested :class:`_schema.Column` on the mapped :class:`.Table` for
+``Employee`` first, and if present, maintain that existing mapping. If not
+present, :func:`_orm.mapped_column` will map the column normally, adding it
+as one of the columns in the :class:`.Table` referred towards by the
+``Employee`` superclass.
+
+
+.. versionadded:: 2.0.0b4 - Added :paramref:`_orm.mapped_column.use_existing_column`,
+ which provides a 2.0-compatible means of mapping a column on an inheriting
+ subclass conditionally. The previous approach which combines
+ :class:`.declared_attr` with a lookup on the parent ``.__table__``
+ continues to function as well, but lacks :pep:`484` typing support.
+
A similar concept can be used with mixin classes (see :ref:`orm_mixins_toplevel`)
to define a particular series of columns and/or other mapped attributes
class HasStartDate:
- @declared_attr
- def start_date(cls) -> Mapped[datetime]:
- return cls.__table__.c.get(
- "start_date", mapped_column(DateTime) # type: ignore
- )
+ start_date: Mapped[datetime] = mapped_column(use_existing_column=True)
class Engineer(HasStartDate, Employee):
deferred: Union[_NoArg, bool] = _NoArg.NO_ARG,
deferred_group: Optional[str] = None,
deferred_raiseload: bool = False,
+ use_existing_column: bool = False,
name: Optional[str] = None,
type_: Optional[_TypeEngineArgument[Any]] = None,
autoincrement: Union[bool, Literal["auto", "ignore_fk"]] = "auto",
:ref:`orm_queryguide_deferred_raiseload`
+ :param use_existing_column: if True, will attempt to locate the given
+ column name on an inherited superclass (typically single inheriting
+ superclass), and if present, will not produce a new column, mapping
+ to the superclass column as though it were omitted from this class.
+ This is used for mixins that add new columns to an inherited superclass.
+
+ .. seealso::
+
+ :ref:`orm_inheritance_column_conflicts`
+
+ .. versionadded:: 2.0.0b4
+
:param default: Passed directly to the
:paramref:`_schema.Column.default` parameter if the
:paramref:`_orm.mapped_column.insert_default` parameter is not present.
primary_key=primary_key,
server_default=server_default,
server_onupdate=server_onupdate,
+ use_existing_column=use_existing_column,
quote=quote,
comment=comment,
system=system,
return None
+def _is_supercls_for_inherits(cls: Type[Any]) -> bool:
+ """return True if this class will be used as a superclass to set in
+ 'inherits'.
+
+ This includes deferred mapper configs that aren't mapped yet, however does
+ not include classes with _sa_decl_prepare_nocascade (e.g.
+ ``AbstractConcreteBase``); these concrete-only classes are not set up as
+ "inherits" until after mappers are configured using
+ mapper._set_concrete_base()
+
+ """
+ if _DeferredMapperConfig.has_cls(cls):
+ return not _get_immediate_cls_attr(
+ cls, "_sa_decl_prepare_nocascade", strict=True
+ )
+ # regular mapping
+ elif _is_mapped_class(cls):
+ return True
+ else:
+ return False
+
+
def _resolve_for_abstract_or_classical(cls: Type[Any]) -> Optional[Type[Any]]:
if cls is object:
return None
c = _resolve_for_abstract_or_classical(base_)
if c is None:
continue
- if _declared_mapping_info(
- c
- ) is not None and not _get_immediate_cls_attr(
- c, "_sa_decl_prepare_nocascade", strict=True
- ):
+
+ if _is_supercls_for_inherits(c) and c not in inherits_search:
inherits_search.append(c)
if inherits_search:
"allow_unmapped_annotations",
)
+ is_deferred = False
registry: _RegistryType
clsdict_view: _ClassDict
collected_annotations: Dict[str, _CollectedAnnotation]
self.classname, self.cls, registry._class_registry
)
+ self._setup_inheriting_mapper(mapper_kw)
+
self._extract_mappable_attributes()
self._extract_declared_columns()
self._setup_table(table)
- self._setup_inheritance(mapper_kw)
+ self._setup_inheriting_columns(mapper_kw)
self._early_mapping(mapper_kw)
# need to do this all the way up the hierarchy first
# (see #8190)
- class_mapped = (
- base is not cls
- and _declared_mapping_info(base) is not None
- and not _get_immediate_cls_attr(
- base, "_sa_decl_prepare_nocascade", strict=True
- )
- )
+ class_mapped = base is not cls and _is_supercls_for_inherits(base)
local_attributes_for_class = self._cls_attr_resolver(base)
if mapped_container is not None or annotation is None:
try:
value.declarative_scan(
+ self,
self.registry,
cls,
originating_module,
else:
return manager.registry.metadata
- def _setup_inheritance(self, mapper_kw: _MapperKwArgs) -> None:
- table = self.local_table
+ def _setup_inheriting_mapper(self, mapper_kw: _MapperKwArgs) -> None:
cls = self.cls
- table_args = self.table_args
- declared_columns = self.declared_columns
inherits = mapper_kw.get("inherits", None)
c = _resolve_for_abstract_or_classical(base_)
if c is None:
continue
- if _declared_mapping_info(
- c
- ) is not None and not _get_immediate_cls_attr(
- c, "_sa_decl_prepare_nocascade", strict=True
- ):
- if c not in inherits_search:
- inherits_search.append(c)
+
+ if _is_supercls_for_inherits(c) and c not in inherits_search:
+ inherits_search.append(c)
if inherits_search:
if len(inherits_search) > 1:
self.inherits = inherits
+ def _setup_inheriting_columns(self, mapper_kw: _MapperKwArgs) -> None:
+ table = self.local_table
+ cls = self.cls
+ table_args = self.table_args
+ declared_columns = self.declared_columns
+
if (
table is None
and self.inherits is None
if inherited_table.c[col.name] is col:
continue
raise exc.ArgumentError(
- "Column '%s' on class %s conflicts with "
- "existing column '%s'"
- % (col, cls, inherited_table.c[col.name])
+ f"Column '{col}' on class {cls.__name__} "
+ f"conflicts with existing column "
+ f"'{inherited_table.c[col.name]}'. If using "
+ f"Declarative, consider using the "
+ "use_existing_column parameter of mapped_column() "
+ "to resolve conflicts."
)
if col.primary_key:
raise exc.ArgumentError(
mapper_args["inherits"] = self.inherits
if self.inherits and not mapper_args.get("concrete", False):
+ # note the superclass is expected to have a Mapper assigned and
+ # not be a deferred config, as this is called within map()
+ inherited_mapper = class_mapper(self.inherits, False)
+ inherited_table = inherited_mapper.local_table
+
# single or joined inheritance
# exclude any cols on the inherited table which are
# not mapped on the parent class, to avoid
# mapping columns specific to sibling/nephew classes
- inherited_mapper = _declared_mapping_info(self.inherits)
- assert isinstance(inherited_mapper, Mapper)
- inherited_table = inherited_mapper.local_table
-
if "exclude_properties" not in mapper_args:
mapper_args["exclude_properties"] = exclude_properties = {
c.key
class _DeferredMapperConfig(_ClassScanMapperConfig):
_cls: weakref.ref[Type[Any]]
+ is_deferred = True
+
_configs: util.OrderedDict[
weakref.ref[Type[Any]], _DeferredMapperConfig
] = util.OrderedDict()
from .attributes import InstrumentedAttribute
from .attributes import QueryableAttribute
from .context import ORMCompileState
+ from .decl_base import _ClassScanMapperConfig
from .mapper import Mapper
from .properties import ColumnProperty
from .properties import MappedColumn
@util.preload_module("sqlalchemy.orm.properties")
def declarative_scan(
self,
+ decl_scan: _ClassScanMapperConfig,
registry: _RegistryType,
cls: Type[Any],
originating_module: Optional[str],
from .context import ORMCompileState
from .context import QueryContext
from .decl_api import RegistryType
+ from .decl_base import _ClassScanMapperConfig
from .loading import _PopulatorDict
from .mapper import Mapper
from .path_registry import AbstractEntityRegistry
def declarative_scan(
self,
+ decl_scan: _ClassScanMapperConfig,
registry: RegistryType,
cls: Type[Any],
originating_module: Optional[str],
from . import attributes
from . import strategy_options
from .base import _DeclarativeMapped
+from .base import class_mapper
from .descriptor_props import CompositeProperty
from .descriptor_props import ConcreteInheritedProperty
from .descriptor_props import SynonymProperty
from ._typing import _ORMColumnExprArgument
from ._typing import _RegistryType
from .base import Mapped
+ from .decl_base import _ClassScanMapperConfig
from .mapper import Mapper
from .session import Session
from .state import _InstallLoaderCallableProto
def declarative_scan(
self,
+ decl_scan: _ClassScanMapperConfig,
registry: _RegistryType,
cls: Type[Any],
originating_module: Optional[str],
"deferred_raiseload",
"_attribute_options",
"_has_dataclass_arguments",
+ "_use_existing_column",
)
deferred: bool
"attribute_options", _DEFAULT_ATTRIBUTE_OPTIONS
)
+ self._use_existing_column = kw.pop("use_existing_column", False)
+
self._has_dataclass_arguments = False
if attr_opts is not None and attr_opts != _DEFAULT_ATTRIBUTE_OPTIONS:
new._attribute_options = self._attribute_options
new._has_insert_default = self._has_insert_default
new._has_dataclass_arguments = self._has_dataclass_arguments
+ new._use_existing_column = self._use_existing_column
util.set_creation_order(new)
return new
def declarative_scan(
self,
+ decl_scan: _ClassScanMapperConfig,
registry: _RegistryType,
cls: Type[Any],
originating_module: Optional[str],
is_dataclass_field: bool,
) -> None:
column = self.column
+
+ if self._use_existing_column and decl_scan.inherits:
+ if decl_scan.is_deferred:
+ raise sa_exc.ArgumentError(
+ "Can't use use_existing_column with deferred mappers"
+ )
+ supercls_mapper = class_mapper(decl_scan.inherits, False)
+
+ column = self.column = supercls_mapper.local_table.c.get( # type: ignore # noqa: E501
+ key, column
+ )
+
if column.key is None:
column.key = key
if column.name is None:
from .base import Mapped
from .clsregistry import _class_resolver
from .clsregistry import _ModNS
+ from .decl_base import _ClassScanMapperConfig
from .dependency import DependencyProcessor
from .mapper import Mapper
from .query import Query
def declarative_scan(
self,
+ decl_scan: _ClassScanMapperConfig,
registry: _RegistryType,
cls: Type[Any],
originating_module: Optional[str],
+from __future__ import annotations
+
+from sqlalchemy import exc
from sqlalchemy import ForeignKey
from sqlalchemy import Integer
from sqlalchemy import String
from sqlalchemy.orm import decl_api as decl
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import exc as orm_exc
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
from sqlalchemy.orm.decl_base import _DeferredMapperConfig
from sqlalchemy.testing import assert_raises_message
from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
from sqlalchemy.testing.fixtures import fixture_session
from sqlalchemy.testing.schema import Column
def teardown_test(self):
clear_mappers()
+ @testing.fixture
+ def decl_base(self):
+ yield Base
+
class DeferredReflectBase(DeclarativeReflectionBase):
def teardown_test(self):
Column("bar_data", String(30)),
)
- def test_basic(self):
- class Foo(DeferredReflection, fixtures.ComparableEntity, Base):
+ def test_basic(self, decl_base):
+ class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
__tablename__ = "foo"
__mapper_args__ = {
"polymorphic_on": "type",
DeferredReflection.prepare(testing.db)
self._roundtrip()
- def test_add_subclass_column(self):
- class Foo(DeferredReflection, fixtures.ComparableEntity, Base):
+ def test_add_subclass_column(self, decl_base):
+ class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
__tablename__ = "foo"
__mapper_args__ = {
"polymorphic_on": "type",
DeferredReflection.prepare(testing.db)
self._roundtrip()
- def test_add_pk_column(self):
- class Foo(DeferredReflection, fixtures.ComparableEntity, Base):
+ def test_add_subclass_mapped_column(self, decl_base):
+ class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
+ __tablename__ = "foo"
+ __mapper_args__ = {
+ "polymorphic_on": "type",
+ "polymorphic_identity": "foo",
+ }
+
+ class Bar(Foo):
+ __mapper_args__ = {"polymorphic_identity": "bar"}
+ bar_data: Mapped[str]
+
+ DeferredReflection.prepare(testing.db)
+ self._roundtrip()
+
+ def test_subclass_mapped_column_no_existing(self, decl_base):
+ class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
+ __tablename__ = "foo"
+ __mapper_args__ = {
+ "polymorphic_on": "type",
+ "polymorphic_identity": "foo",
+ }
+
+ with expect_raises_message(
+ exc.ArgumentError,
+ "Can't use use_existing_column with deferred mappers",
+ ):
+
+ class Bar(Foo):
+ __mapper_args__ = {"polymorphic_identity": "bar"}
+ bar_data: Mapped[str] = mapped_column(use_existing_column=True)
+
+ def test_add_pk_column(self, decl_base):
+ class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
__tablename__ = "foo"
__mapper_args__ = {
"polymorphic_on": "type",
DeferredReflection.prepare(testing.db)
self._roundtrip()
+ def test_add_pk_mapped_column(self, decl_base):
+ class Foo(DeferredReflection, fixtures.ComparableEntity, decl_base):
+ __tablename__ = "foo"
+ __mapper_args__ = {
+ "polymorphic_on": "type",
+ "polymorphic_identity": "foo",
+ }
+ id: Mapped[int] = mapped_column(primary_key=True)
+
+ class Bar(Foo):
+ __mapper_args__ = {"polymorphic_identity": "bar"}
+
+ DeferredReflection.prepare(testing.db)
+ self._roundtrip()
+
class DeferredJoinedInhReflectionTest(DeferredInhReflectBase):
@classmethod
import sqlalchemy as sa
from sqlalchemy import ForeignKey
+from sqlalchemy import Identity
from sqlalchemy import Integer
from sqlalchemy import select
from sqlalchemy import String
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import declared_attr
from sqlalchemy.orm import deferred
+from sqlalchemy.orm import Mapped
+from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
+from sqlalchemy.orm import Session
from sqlalchemy.orm import with_polymorphic
from sqlalchemy.orm.decl_api import registry
from sqlalchemy.testing import assert_raises
[Person.__table__.c.name, Person.__table__.c.primary_language],
)
- @testing.skip_if(
- lambda: testing.against("oracle"),
- "Test has an empty insert in it at the moment",
- )
- def test_columns_single_inheritance_conflict_resolution(self):
+ @testing.variation("decl_type", ["legacy", "use_existing_column"])
+ def test_columns_single_inheritance_conflict_resolution(
+ self, connection, decl_base, decl_type
+ ):
"""Test that a declared_attr can return the existing column and it will
be ignored. this allows conditional columns to be added.
"""
- class Person(Base):
+ class Person(decl_base):
__tablename__ = "person"
- id = Column(Integer, primary_key=True)
+ id = Column(Integer, Identity(), primary_key=True)
class Engineer(Person):
"""single table inheritance"""
- @declared_attr
- def target_id(cls):
- return cls.__table__.c.get(
- "target_id", Column(Integer, ForeignKey("other.id"))
+ if decl_type.legacy:
+
+ @declared_attr
+ def target_id(cls):
+ return cls.__table__.c.get(
+ "target_id", Column(Integer, ForeignKey("other.id"))
+ )
+
+ elif decl_type.use_existing_column:
+ target_id: Mapped[int] = mapped_column(
+ ForeignKey("other.id"), use_existing_column=True
)
@declared_attr
"""single table inheritance"""
- @declared_attr
- def target_id(cls):
- return cls.__table__.c.get(
- "target_id", Column(Integer, ForeignKey("other.id"))
+ if decl_type.legacy:
+
+ @declared_attr
+ def target_id(cls):
+ return cls.__table__.c.get(
+ "target_id", Column(Integer, ForeignKey("other.id"))
+ )
+
+ elif decl_type.use_existing_column:
+ target_id: Mapped[int] = mapped_column(
+ ForeignKey("other.id"), use_existing_column=True
)
@declared_attr
def target(cls):
return relationship("Other")
- class Other(Base):
+ class Other(decl_base):
__tablename__ = "other"
- id = Column(Integer, primary_key=True)
+ id = Column(Integer, Identity(), primary_key=True)
is_(
Engineer.target_id.property.columns[0],
Manager.target_id.property.columns[0], Person.__table__.c.target_id
)
# do a brief round trip on this
- Base.metadata.create_all(testing.db)
- session = fixture_session()
- o1, o2 = Other(), Other()
- session.add_all(
- [Engineer(target=o1), Manager(target=o2), Manager(target=o1)]
- )
- session.commit()
- eq_(session.query(Engineer).first().target, o1)
+ decl_base.metadata.create_all(connection)
+ with Session(connection) as session:
+ o1, o2 = Other(), Other()
+ session.add_all(
+ [Engineer(target=o1), Manager(target=o2), Manager(target=o1)]
+ )
+ session.commit()
+ eq_(session.query(Engineer).first().target, o1)
- def test_columns_single_inheritance_conflict_resolution_pk(self):
+ @testing.variation("decl_type", ["legacy", "use_existing_column"])
+ def test_columns_single_inheritance_conflict_resolution_pk(
+ self, decl_base, decl_type
+ ):
"""Test #2472 in terms of a primary key column. This is
#4352.
"""
- class Person(Base):
+ class Person(decl_base):
__tablename__ = "person"
id = Column(Integer, primary_key=True)
"""single table inheritance"""
- @declared_attr
- def target_id(cls):
- return cls.__table__.c.get(
- "target_id", Column(Integer, primary_key=True)
+ if decl_type.legacy:
+
+ @declared_attr
+ def target_id(cls):
+ return cls.__table__.c.get(
+ "target_id", Column(Integer, primary_key=True)
+ )
+
+ elif decl_type.use_existing_column:
+ target_id: Mapped[int] = mapped_column(
+ primary_key=True, use_existing_column=True
)
class Manager(Person):
"""single table inheritance"""
- @declared_attr
- def target_id(cls):
- return cls.__table__.c.get(
- "target_id", Column(Integer, primary_key=True)
+ if decl_type.legacy:
+
+ @declared_attr
+ def target_id(cls):
+ return cls.__table__.c.get(
+ "target_id", Column(Integer, primary_key=True)
+ )
+
+ elif decl_type.use_existing_column:
+ target_id: Mapped[int] = mapped_column(
+ primary_key=True, use_existing_column=True
)
is_(
Manager.target_id.property.columns[0], Person.__table__.c.target_id
)
- def test_columns_single_inheritance_cascading_resolution_pk(self):
+ @testing.variation("decl_type", ["legacy", "use_existing_column"])
+ def test_columns_single_inheritance_cascading_resolution_pk(
+ self, decl_type
+ ):
"""An additional test for #4352 in terms of the requested use case."""
class TestBase(Base):
__abstract__ = True
- @declared_attr.cascading
- def id(cls):
- col_val = None
- if TestBase not in cls.__bases__:
- col_val = cls.__table__.c.get("id")
- if col_val is None:
- col_val = Column(Integer, primary_key=True)
- return col_val
+ if decl_type.legacy:
+
+ @declared_attr.cascading
+ def id(cls): # noqa: A001
+ col_val = None
+ if TestBase not in cls.__bases__:
+ col_val = cls.__table__.c.get("id")
+ if col_val is None:
+ col_val = Column(Integer, primary_key=True)
+ return col_val
+
+ elif decl_type.use_existing_column:
+ id: Mapped[int] = mapped_column( # noqa: A001
+ primary_key=True, use_existing_column=True
+ )
class Person(TestBase):
"""single table base class"""
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
