--- /dev/null
+.. change::
+ :tags: usecase, orm
+ :tickets: 9220
+
+ Added new event hook :meth:`_orm.MapperEvents.after_mapper_constructed`,
+ which supplies an event hook to take place right as the
+ :class:`_orm.Mapper` object has been fully constructed, but before the
+ :meth:`_orm.registry.configure` call has been called. This allows code that
+ can create additional mappings and table structures based on the initial
+ configuration of a :class:`_orm.Mapper`, which also integrates within
+ Declarative configuration. Previously, when using Declarative, where the
+ :class:`_orm.Mapper` object is created within the class creation process,
+ there was no documented means of running code at this point. The change
+ is to immediately benefit custom mapping schemes such as that
+ of the :ref:`examples_versioned_history` example, which generate additional
+ mappers and tables in response to the creation of mapped classes.
+
+
+.. change::
+ :tags: usecase, orm
+ :tickets: 9220
+
+ The infrequently used :attr:`_orm.Mapper.iterate_properties` attribute and
+ :meth:`_orm.Mapper.get_property` method, which are primarily used
+ internally, no longer implicitly invoke the :meth:`_orm.registry.configure`
+ process. Public access to these methods is extremely rare and the only
+ benefit to having :meth:`_orm.registry.configure` would have been allowing
+ "backref" properties be present in these collections. In order to support
+ the new :meth:`_orm.MapperEvents.after_mapper_constructed` event, iteration
+ and access to the internal :class:`_orm.MapperProperty` objects is now
+ possible without triggering an implicit configure of the mapper itself.
+
+ The more-public facing route to iteration of all mapper attributes, the
+ :attr:`_orm.Mapper.attrs` collection and similar, will still implicitly
+ invoke the :meth:`_orm.registry.configure` step thus making backref
+ attributes available.
+
+ In all cases, the :meth:`_orm.registry.configure` is always available to
+ be called directly.
+
+.. change::
+ :tags: bug, examples
+ :tickets: 9220
+
+ Reworked the :ref:`examples_versioned_history` to work with
+ version 2.0, while at the same time improving the overall working of
+ this example to use newer APIs, including a newly added hook
+ :meth:`_orm.MapperEvents.after_mapper_constructed`.
+
+
--- /dev/null
+.. change::
+ :tags: bug, orm, regression
+ :tickets: 9232
+
+ Fixed obscure ORM inheritance issue caused by :ticket:`8705` where some
+ scenarios of inheriting mappers that indicated groups of columns from the
+ local table and the inheriting table together under a
+ :func:`_orm.column_property` would nonetheless warn that properties of the
+ same name were being combined implicitly.
as new rows in the same table, without using a separate history table.
Usage is illustrated via a unit test module ``test_versioning.py``, which can
-be run via ``pytest``::
+be run like any other module, using ``unittest`` internally::
- # assume SQLAlchemy is installed where pytest is
-
- cd examples/versioned_history
- pytest test_versioning.py
+ python -m examples.versioned_history.test_versioning
A fragment of example usage, using declarative::
from sqlalchemy import event
from sqlalchemy import ForeignKeyConstraint
from sqlalchemy import Integer
-from sqlalchemy import Table
+from sqlalchemy import PrimaryKeyConstraint
from sqlalchemy import util
-from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import attributes
-from sqlalchemy.orm import mapper
from sqlalchemy.orm import object_mapper
from sqlalchemy.orm.exc import UnmappedColumnError
from sqlalchemy.orm.relationships import RelationshipProperty
def _history_mapper(local_mapper):
+
cls = local_mapper.class_
- # set the "active_history" flag
- # on on column-mapped attributes so that the old version
- # of the info is always loaded (currently sets it on all attributes)
- for prop in local_mapper.iterate_properties:
- getattr(local_mapper.class_, prop.key).impl.active_history = True
+ if cls.__dict__.get("_history_mapper_configured", False):
+ return
- super_mapper = local_mapper.inherits
- super_history_mapper = getattr(cls, "__history_mapper__", None)
+ cls._history_mapper_configured = True
+ super_mapper = local_mapper.inherits
polymorphic_on = None
super_fks = []
+ properties = util.OrderedDict()
- def _col_copy(col):
- orig = col
- col = col.copy()
- orig.info["history_copy"] = col
- col.unique = False
- col.default = col.server_default = None
- col.autoincrement = False
- return col
+ if super_mapper:
+ super_history_mapper = super_mapper.class_.__history_mapper__
+ else:
+ super_history_mapper = None
- properties = util.OrderedDict()
if (
not super_mapper
or local_mapper.local_table is not super_mapper.local_table
):
- cols = []
version_meta = {"version_meta": True} # add column.info to identify
# columns specific to versioning
- for column in local_mapper.local_table.c:
- if _is_versioning_col(column):
- continue
+ history_table = local_mapper.local_table.to_metadata(
+ local_mapper.local_table.metadata,
+ name=local_mapper.local_table.name + "_history",
+ )
- col = _col_copy(column)
+ for orig_c, history_c in zip(
+ local_mapper.local_table.c, history_table.c
+ ):
+ orig_c.info["history_copy"] = history_c
+ history_c.unique = False
+ history_c.default = history_c.server_default = None
+ history_c.autoincrement = False
if super_mapper and col_references_table(
- column, super_mapper.local_table
+ orig_c, super_mapper.local_table
):
+ assert super_history_mapper is not None
super_fks.append(
(
- col.key,
+ history_c.key,
list(super_history_mapper.local_table.primary_key)[0],
)
)
+ if orig_c is local_mapper.polymorphic_on:
+ polymorphic_on = history_c
- cols.append(col)
-
- if column is local_mapper.polymorphic_on:
- polymorphic_on = col
-
- orig_prop = local_mapper.get_property_by_column(column)
+ orig_prop = local_mapper.get_property_by_column(orig_c)
# carry over column re-mappings
if (
len(orig_prop.columns) > 1
col.info["history_copy"] for col in orig_prop.columns
)
- if super_mapper:
- super_fks.append(
- ("version", super_history_mapper.local_table.c.version)
- )
+ for const in list(history_table.constraints):
+ if not isinstance(
+ const, (PrimaryKeyConstraint, ForeignKeyConstraint)
+ ):
+ history_table.constraints.discard(const)
# "version" stores the integer version id. This column is
# required.
- cols.append(
+ history_table.append_column(
Column(
"version",
Integer,
# "changed" column stores the UTC timestamp of when the
# history row was created.
# This column is optional and can be omitted.
- cols.append(
+ history_table.append_column(
Column(
"changed",
DateTime,
)
)
+ if super_mapper:
+ super_fks.append(
+ ("version", super_history_mapper.local_table.c.version)
+ )
+
if super_fks:
- cols.append(ForeignKeyConstraint(*zip(*super_fks)))
+ history_table.append_constraint(
+ ForeignKeyConstraint(*zip(*super_fks))
+ )
- table = Table(
- local_mapper.local_table.name + "_history",
- local_mapper.local_table.metadata,
- *cols,
- schema=local_mapper.local_table.schema,
- )
else:
+ history_table = None
+ super_history_table = super_mapper.local_table.metadata.tables[
+ super_mapper.local_table.name + "_history"
+ ]
+
# single table inheritance. take any additional columns that may have
# been added and add them to the history table.
for column in local_mapper.local_table.c:
- if column.key not in super_history_mapper.local_table.c:
- col = _col_copy(column)
- super_history_mapper.local_table.append_column(col)
- table = None
+ if column.key not in super_history_table.c:
+ col = Column(
+ column.name, column.type, nullable=column.nullable
+ )
+ super_history_table.append_column(col)
+
+ if not super_mapper:
+ local_mapper.local_table.append_column(
+ Column("version", Integer, default=1, nullable=False),
+ replace_existing=True,
+ )
+ local_mapper.add_property(
+ "version", local_mapper.local_table.c.version
+ )
+
+ if cls.use_mapper_versioning:
+ local_mapper.version_id_col = local_mapper.local_table.c.version
+
+ # set the "active_history" flag
+ # on on column-mapped attributes so that the old version
+ # of the info is always loaded (currently sets it on all attributes)
+ for prop in local_mapper.iterate_properties:
+ prop.active_history = True
+
+ super_mapper = local_mapper.inherits
if super_history_mapper:
bases = (super_history_mapper.class_,)
- if table is not None:
- properties["changed"] = (table.c.changed,) + tuple(
+ if history_table is not None:
+ properties["changed"] = (history_table.c.changed,) + tuple(
super_history_mapper.attrs.changed.columns
)
else:
bases = local_mapper.base_mapper.class_.__bases__
- versioned_cls = type.__new__(type, "%sHistory" % cls.__name__, bases, {})
- m = mapper(
+ versioned_cls = type.__new__(
+ type,
+ "%sHistory" % cls.__name__,
+ bases,
+ {"_history_mapper_configured": True},
+ )
+
+ m = local_mapper.registry.map_imperatively(
versioned_cls,
- table,
+ history_table,
inherits=super_history_mapper,
- polymorphic_on=polymorphic_on,
polymorphic_identity=local_mapper.polymorphic_identity,
+ polymorphic_on=polymorphic_on,
properties=properties,
)
cls.__history_mapper__ = m
- if not super_history_mapper:
- local_mapper.local_table.append_column(
- Column("version", Integer, default=1, nullable=False),
- replace_existing=True,
- )
- local_mapper.add_property(
- "version", local_mapper.local_table.c.version
- )
- if cls.use_mapper_versioning:
- local_mapper.version_id_col = local_mapper.local_table.c.version
-
class Versioned:
use_mapper_versioning = False
"""if True, also assign the version column to be tracked by the mapper"""
- @declared_attr
- def __mapper_cls__(cls):
- def map_(cls, *arg, **kw):
- mp = mapper(cls, *arg, **kw)
- _history_mapper(mp)
- return mp
-
- return map_
-
__table_args__ = {"sqlite_autoincrement": True}
"""Use sqlite_autoincrement, to ensure unique integer values
are used for new rows even for rows that have been deleted."""
+ def __init_subclass__(cls) -> None:
+ @event.listens_for(cls, "after_mapper_constructed")
+ def _mapper_constructed(mapper, class_):
+ _history_mapper(mapper)
+
def versioned_objects(iter_):
for obj in iter_:
"""Unit tests illustrating usage of the ``history_meta.py``
module functions."""
-from unittest import TestCase
+import unittest
import warnings
from sqlalchemy import Boolean
warnings.simplefilter("error")
-engine = None
-
-def setup_module():
- global engine
- engine = create_engine("sqlite://", echo=True)
-
-
-class TestVersioning(TestCase, AssertsCompiledSQL):
+class TestVersioning(AssertsCompiledSQL):
__dialect__ = "default"
def setUp(self):
+
+ self.engine = engine = create_engine("sqlite://")
self.session = Session(engine)
self.Base = declarative_base()
versioned_session(self.session)
def tearDown(self):
self.session.close()
clear_mappers()
- self.Base.metadata.drop_all(engine)
+ self.Base.metadata.drop_all(self.engine)
def create_tables(self):
- self.Base.metadata.create_all(engine)
+ self.Base.metadata.create_all(self.engine)
def test_plain(self):
class SomeClass(Versioned, self.Base, ComparableEntity):
self.assert_compile(
q,
"SELECT "
- "subsubtable_history.id AS subsubtable_history_id, "
- "subtable_history.id AS subtable_history_id, "
- "basetable_history.id AS basetable_history_id, "
- "subsubtable_history.changed AS subsubtable_history_changed, "
- "subtable_history.changed AS subtable_history_changed, "
- "basetable_history.changed AS basetable_history_changed, "
"basetable_history.name AS basetable_history_name, "
"basetable_history.type AS basetable_history_type, "
"subsubtable_history.version AS subsubtable_history_version, "
"basetable_history.version AS basetable_history_version, "
"subtable_history.base_id AS subtable_history_base_id, "
"subtable_history.subdata1 AS subtable_history_subdata1, "
+ "subsubtable_history.id AS subsubtable_history_id, "
+ "subtable_history.id AS subtable_history_id, "
+ "basetable_history.id AS basetable_history_id, "
+ "subsubtable_history.changed AS subsubtable_history_changed, "
+ "subtable_history.changed AS subtable_history_changed, "
+ "basetable_history.changed AS basetable_history_changed, "
"subsubtable_history.subdata2 AS subsubtable_history_subdata2 "
"FROM basetable_history "
"JOIN subtable_history "
DocumentHistory = Document.__history_mapper__.class_
v2 = self.session.query(Document).one()
v1 = self.session.query(DocumentHistory).one()
- self.assertEqual(v1.id, v2.id)
- self.assertEqual(v2.name, "Bar")
- self.assertEqual(v1.name, "Foo")
+ eq_(v1.id, v2.id)
+ eq_(v2.name, "Bar")
+ eq_(v1.name, "Foo")
def test_mutate_named_column(self):
class Document(self.Base, Versioned):
DocumentHistory = Document.__history_mapper__.class_
v2 = self.session.query(Document).one()
v1 = self.session.query(DocumentHistory).one()
- self.assertEqual(v1.id, v2.id)
- self.assertEqual(v2.description_, "Bar")
- self.assertEqual(v1.description_, "Foo")
+ eq_(v1.id, v2.id)
+ eq_(v2.description_, "Bar")
+ eq_(v1.description_, "Foo")
def test_unique_identifiers_across_deletes(self):
"""Ensure unique integer values are used for the primary table.
# If previous assertion fails, this will also fail:
sc2.name = "sc2 modified"
sess.commit()
+
+
+class TestVersioningUnittest(unittest.TestCase, TestVersioning):
+ pass
+
+
+if __name__ == "__main__":
+ unittest.main()
before instrumentation is applied to the mapped class.
This event is the earliest phase of mapper construction.
- Most attributes of the mapper are not yet initialized.
+ Most attributes of the mapper are not yet initialized. To
+ receive an event within initial mapper construction where basic
+ state is available such as the :attr:`_orm.Mapper.attrs` collection,
+ the :meth:`_orm.MapperEvents.after_mapper_constructed` event may
+ be a better choice.
This listener can either be applied to the :class:`_orm.Mapper`
class overall, or to any un-mapped class which serves as a base
of this event.
:param class\_: the mapped class.
+ .. seealso::
+
+ :meth:`_orm.MapperEvents.after_mapper_constructed`
+
+ """
+
+ def after_mapper_constructed(
+ self, mapper: Mapper[_O], class_: Type[_O]
+ ) -> None:
+ """Receive a class and mapper when the :class:`_orm.Mapper` has been
+ fully constructed.
+
+ This event is called after the initial constructor for
+ :class:`_orm.Mapper` completes. This occurs after the
+ :meth:`_orm.MapperEvents.instrument_class` event and after the
+ :class:`_orm.Mapper` has done an initial pass of its arguments
+ to generate its collection of :class:`_orm.MapperProperty` objects,
+ which are accessible via the :meth:`_orm.Mapper.get_property`
+ method and the :attr:`_orm.Mapper.iterate_properties` attribute.
+
+ This event differs from the
+ :meth:`_orm.MapperEvents.before_mapper_configured` event in that it
+ is invoked within the constructor for :class:`_orm.Mapper`, rather
+ than within the :meth:`_orm.registry.configure` process. Currently,
+ this event is the only one which is appropriate for handlers that
+ wish to create additional mapped classes in response to the
+ construction of this :class:`_orm.Mapper`, which will be part of the
+ same configure step when :meth:`_orm.registry.configure` next runs.
+
+ .. versionadded:: 2.0.2
+
+ .. seealso::
+
+ :ref:`examples_versioning` - an example which illustrates the use
+ of the :meth:`_orm.MapperEvents.before_mapper_configured`
+ event to create new mappers to record change-audit histories on
+ objects.
+
"""
def before_mapper_configured(
the configure step, by returning the :attr:`.orm.interfaces.EXT_SKIP`
symbol which indicates to the :func:`.configure_mappers` call that this
particular mapper (or hierarchy of mappers, if ``propagate=True`` is
- used) should be skipped in the current configuration run. When one or
+ used) should be skipped in the current configuration run. When one or
more mappers are skipped, the he "new mappers" flag will remain set,
meaning the :func:`.configure_mappers` function will continue to be
called when mappers are used, to continue to try to configure all
self._log("constructed")
self._expire_memoizations()
+ self.dispatch.after_mapper_constructed(self, self.class_)
+
def _prefer_eager_defaults(self, dialect, table):
if self.eager_defaults == "auto":
if not table.implicit_returning:
# that's given. For other properties, set them up in _props now.
if self._init_properties:
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
# 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.
+
+ _map_as_property_now = True
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
+ for given_col in possible_col_prop.columns:
+ if self.local_table.c.contains_column(given_col):
+ _map_as_property_now = False
+ explicit_col_props_by_key[key] = possible_col_prop
+ explicit_col_props_by_column[given_col] = (
+ key,
+ possible_col_prop,
+ )
- self._configure_property(key, possible_col_prop, init=False)
+ if _map_as_property_now:
+ self._configure_property(
+ key, possible_col_prop, init=False
+ )
# step 2: pull properties from the inherited mapper. reconcile
# columns with those which are explicit above. for properties that
incoming_prop=incoming_prop,
)
explicit_col_props_by_key[key] = new_prop
- explicit_col_props_by_column[incoming_prop.columns[0]] = (
- key,
- new_prop,
- )
+
+ for inc_col in incoming_prop.columns:
+ explicit_col_props_by_column[inc_col] = (
+ key,
+ new_prop,
+ )
elif key not in self._props:
self._adapt_inherited_property(key, inherited_prop, False)
# reconciliation against inherited columns occurs here also.
for column in self.persist_selectable.columns:
-
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
return key in self._props
def get_property(
- self, key: str, _configure_mappers: bool = True
+ self, key: str, _configure_mappers: bool = False
) -> MapperProperty[Any]:
"""return a MapperProperty associated with the given key."""
return self._props[key]
except KeyError as err:
raise sa_exc.InvalidRequestError(
- "Mapper '%s' has no property '%s'" % (self, key)
+ f"Mapper '{self}' has no property '{key}'. If this property "
+ "was indicated from other mappers or configure events, ensure "
+ "registry.configure() has been called."
) from err
def get_property_by_column(
def iterate_properties(self):
"""return an iterator of all MapperProperty objects."""
- self._check_configure()
return iter(self._props.values())
def _mappers_from_spec(
for mapper in reg._mappers_to_configure():
run_configure = None
+
for fn in mapper.dispatch.before_mapper_configured:
run_configure = fn(mapper, mapper.class_)
if run_configure is EXT_SKIP:
--- /dev/null
+import os
+import sys
+
+from sqlalchemy.testing import fixtures
+
+
+here = os.path.dirname(__file__)
+sqla_base = os.path.normpath(os.path.join(here, "..", ".."))
+
+
+sys.path.insert(0, sqla_base)
+
+test_versioning = __import__(
+ "examples.versioned_history.test_versioning"
+).versioned_history.test_versioning
+
+
+class VersionedRowsTest(
+ test_versioning.TestVersioning,
+ fixtures.RemoveORMEventsGlobally,
+ fixtures.TestBase,
+):
+ pass
from sqlalchemy.testing import mock
from sqlalchemy.testing.assertions import expect_raises_message
from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.fixtures import RemoveORMEventsGlobally
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
-from test.orm.test_events import _RemoveListeners
Base = None
class ConcreteInhTest(
- _RemoveListeners, DeclarativeTestBase, testing.AssertsCompiledSQL
+ RemoveORMEventsGlobally, DeclarativeTestBase, testing.AssertsCompiledSQL
):
def _roundtrip(
self,
class ConcreteExtensionConfigTest(
- _RemoveListeners, testing.AssertsCompiledSQL, DeclarativeTestBase
+ RemoveORMEventsGlobally, testing.AssertsCompiledSQL, DeclarativeTestBase
):
__dialect__ = "default"
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_subsubclass_groups_super_cols(self, decl_base):
+ """tested for #9220, which is a regression caused by #8705."""
+
+ class BaseClass(decl_base):
+ __tablename__ = "basetable"
+
+ id = Column(Integer, primary_key=True)
+ name = Column(String(50))
+ type = Column(String(20))
+
+ __mapper_args__ = {
+ "polymorphic_on": type,
+ "polymorphic_identity": "base",
+ }
+
+ class SubClass(BaseClass):
+ __tablename__ = "subtable"
+
+ id = column_property(
+ Column(Integer, primary_key=True), BaseClass.id
+ )
+ base_id = Column(Integer, ForeignKey("basetable.id"))
+ subdata1 = Column(String(50))
+
+ __mapper_args__ = {"polymorphic_identity": "sub"}
+
+ class SubSubClass(SubClass):
+ __tablename__ = "subsubtable"
+
+ id = column_property(
+ Column(Integer, ForeignKey("subtable.id"), primary_key=True),
+ SubClass.id,
+ BaseClass.id,
+ )
+ subdata2 = Column(String(50))
+
+ __mapper_args__ = {"polymorphic_identity": "subsub"}
+
+ is_(SubSubClass.id.property.columns[0], SubSubClass.__table__.c.id)
+ is_(
+ SubSubClass.id.property.columns[1]._deannotate(),
+ SubClass.__table__.c.id,
+ )
+ is_(
+ SubSubClass.id.property.columns[2]._deannotate(),
+ BaseClass.__table__.c.id,
+ )
+
def test_column_setup_sanity_check(self, decl_base):
class A(decl_base):
__tablename__ = "a"
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.entities import ComparableEntity
from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.fixtures import RemoveORMEventsGlobally
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
-from test.orm.test_events import _RemoveListeners
class ConcreteTest(AssertsCompiledSQL, fixtures.MappedTest):
eq_(sess.get(Office, 2).name, "office2")
-class AdaptOnNamesTest(_RemoveListeners, fixtures.DeclarativeMappedTest):
+class AdaptOnNamesTest(
+ RemoveORMEventsGlobally, fixtures.DeclarativeMappedTest
+):
"""test the full integration case for #7805"""
@classmethod
from sqlalchemy.testing import mock
from sqlalchemy.testing.fixtures import CacheKeyFixture
from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.fixtures import RemoveORMEventsGlobally
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from . import _fixtures
from .inheritance._poly_fixtures import Manager
from .inheritance._poly_fixtures import Person
from .test_deferred import InheritanceTest as _deferred_InheritanceTest
-from .test_events import _RemoveListeners
from .test_options import PathTest as OptionsPathTest
from .test_options import PathTest
from .test_options import QueryTest as OptionsQueryTest
self.assert_sql_count(testing.db, go, 1)
-class SessionEventsTest(_RemoveListeners, _fixtures.FixtureTest):
+class SessionEventsTest(RemoveORMEventsGlobally, _fixtures.FixtureTest):
run_inserts = None
def test_on_bulk_update_hook(self):
from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import declarative_base
from sqlalchemy.orm import deferred
-from sqlalchemy.orm import events
from sqlalchemy.orm import EXT_SKIP
from sqlalchemy.orm import instrumentation
from sqlalchemy.orm import joinedload
from sqlalchemy.testing.assertions import expect_raises_message
from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.fixtures import fixture_session
+from sqlalchemy.testing.fixtures import RemoveORMEventsGlobally
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
from sqlalchemy.testing.util import gc_collect
from test.orm import _fixtures
-class _RemoveListeners:
- @testing.fixture(autouse=True)
- def _remove_listeners(self):
- yield
- events.MapperEvents._clear()
- events.InstanceEvents._clear()
- events.SessionEvents._clear()
- events.InstrumentationEvents._clear()
- events.QueryEvents._clear()
-
-
-class ORMExecuteTest(_RemoveListeners, _fixtures.FixtureTest):
+class ORMExecuteTest(RemoveORMEventsGlobally, _fixtures.FixtureTest):
run_setup_mappers = "once"
run_inserts = "once"
run_deletes = None
eq_(m1.mock_calls, [])
-class MapperEventsTest(_RemoveListeners, _fixtures.FixtureTest):
+class MapperEventsTest(RemoveORMEventsGlobally, _fixtures.FixtureTest):
run_inserts = None
@classmethod
canary.mock_calls,
)
+ @testing.variation(
+ "listen_type",
+ ["listen_on_mapper", "listen_on_base", "listen_on_mixin"],
+ )
+ def test_mapper_config_sequence(self, decl_base, listen_type):
+
+ canary = Mock()
+
+ if listen_type.listen_on_mapper:
+ event.listen(Mapper, "instrument_class", canary.instrument_class)
+ event.listen(
+ Mapper,
+ "after_mapper_constructed",
+ canary.after_mapper_constructed,
+ )
+ elif listen_type.listen_on_base:
+ event.listen(
+ decl_base,
+ "instrument_class",
+ canary.instrument_class,
+ propagate=True,
+ )
+ event.listen(
+ decl_base,
+ "after_mapper_constructed",
+ canary.after_mapper_constructed,
+ propagate=True,
+ )
+ elif listen_type.listen_on_mixin:
+
+ class Mixin:
+ pass
+
+ event.listen(
+ Mixin,
+ "instrument_class",
+ canary.instrument_class,
+ propagate=True,
+ )
+ event.listen(
+ Mixin,
+ "after_mapper_constructed",
+ canary.after_mapper_constructed,
+ propagate=True,
+ )
+ else:
+ listen_type.fail()
+
+ event.listen(object, "class_instrument", canary.class_instrument)
+ event.listen(Mapper, "before_configured", canary.before_configured)
+ event.listen(
+ Mapper, "before_mapper_configured", canary.before_mapper_configured
+ )
+ event.listen(Mapper, "after_configured", canary.after_configured)
+
+ if listen_type.listen_on_mixin:
+
+ class Thing(Mixin, decl_base):
+ __tablename__ = "thing"
+
+ id = Column(Integer, primary_key=True)
+
+ else:
+
+ class Thing(decl_base):
+ __tablename__ = "thing"
+
+ id = Column(Integer, primary_key=True)
+
+ mp = inspect(Thing)
+ eq_(
+ canary.mock_calls,
+ [
+ call.instrument_class(mp, Thing),
+ call.class_instrument(Thing),
+ call.after_mapper_constructed(mp, Thing),
+ ],
+ )
+
+ decl_base.registry.configure()
+ eq_(
+ canary.mock_calls,
+ [
+ call.instrument_class(mp, Thing),
+ call.class_instrument(Thing),
+ call.after_mapper_constructed(mp, Thing),
+ call.before_configured(),
+ call.before_mapper_configured(mp, Thing),
+ call.after_configured(),
+ ],
+ )
+
@testing.combinations((True,), (False,), argnames="create_dependency")
@testing.combinations((True,), (False,), argnames="configure_at_once")
def test_before_mapper_configured_event(
class DeclarativeEventListenTest(
- _RemoveListeners, fixtures.DeclarativeMappedTest
+ RemoveORMEventsGlobally, fixtures.DeclarativeMappedTest
):
run_setup_classes = "each"
run_deletes = None
eq_(listen.mock_calls, [call(c1, "c"), call(b1, "b"), call(a1, "a")])
-class DeferredMapperEventsTest(_RemoveListeners, _fixtures.FixtureTest):
+class DeferredMapperEventsTest(RemoveORMEventsGlobally, _fixtures.FixtureTest):
""" "test event listeners against unmapped classes.
eq_(canary, [("refresh", None)])
-class SessionEventsTest(_RemoveListeners, _fixtures.FixtureTest):
+class SessionEventsTest(RemoveORMEventsGlobally, _fixtures.FixtureTest):
run_inserts = None
def test_class_listen(self):
assert "name" not in u1.__dict__
-class SessionLifecycleEventsTest(_RemoveListeners, _fixtures.FixtureTest):
+class SessionLifecycleEventsTest(
+ RemoveORMEventsGlobally, _fixtures.FixtureTest
+):
run_inserts = None
def _fixture(self, include_address=False):
class QueryEventsTest(
- _RemoveListeners,
+ RemoveORMEventsGlobally,
_fixtures.FixtureTest,
AssertsCompiledSQL,
testing.AssertsExecutionResults,
m = self.mapper(User, users)
assert not m.configured
assert list(m.iterate_properties)
- assert m.configured
+
+ # as of 2.0 #9220
+ # iterate properties doesn't do "configure" now, there is
+ # no reason for this
+ assert not m.configured
def test_configure_on_get_props_2(self):
User, users = self.classes.User, self.tables.users
m = self.mapper(User, users)
assert not m.configured
assert m.get_property("name")
- assert m.configured
+
+ # as of 2.0 #9220
+ # get_property() doesn't do "configure" now, there is
+ # no reason for this
+ assert not m.configured
def test_configure_on_get_props_3(self):
users, Address, addresses, User = (
addresses,
properties={"user": relationship(User, backref="addresses")},
)
- assert m.get_property("addresses")
+
+ # as of 2.0 #9220
+ # get_property() doesn't do "configure" now, there is
+ # no reason for this
+ with expect_raises_message(
+ sa.exc.InvalidRequestError, r".has no property 'addresses'"
+ ):
+ m.get_property("addresses")
+
+ configure_mappers()
+ is_(m.get_property("addresses"), m.attrs.addresses)
+
+ def test_backrefs_dont_automatically_configure(self):
+ """in #9220 for 2.0 we are changing an ancient behavior that
+ mapper.get_property() and mapper.iterate_properties() would call
+ configure_mappers(). The more modern public interface for this,
+ ``Mapper.attrs``, does.
+
+ """
+ users, Address, addresses, User = (
+ self.tables.users,
+ self.classes.Address,
+ self.tables.addresses,
+ self.classes.User,
+ )
+
+ m = self.mapper(User, users)
+ assert not m.configured
+ configure_mappers()
+
+ m2 = self.mapper(
+ Address,
+ addresses,
+ properties={"user": relationship(User, backref="addresses")},
+ )
+
+ with expect_raises_message(
+ AttributeError, r"no attribute 'addresses'"
+ ):
+ User.addresses
+
+ with expect_raises_message(
+ sa.exc.InvalidRequestError,
+ r"Mapper 'Mapper\[User\(users\)\]' has no property 'addresses'",
+ ):
+ User.__mapper__.get_property("addresses")
+
+ assert not m2.configured
+
+ # the more public-facing collection, mapper.attrs, *does* call
+ # configure still.
+
+ m.attrs.addresses
+ assert m2.configured
+ User.addresses
def test_info(self):
users = self.tables.users
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
