_configure_failed: Any = False
_ready_for_configure = False
- @util.deprecated_params(
- non_primary=(
- "1.3",
- "The :paramref:`.mapper.non_primary` parameter is deprecated, "
- "and will be removed in a future release. The functionality "
- "of non primary mappers is now better suited using the "
- ":class:`.AliasedClass` construct, which can also be used "
- "as the target of a :func:`_orm.relationship` in 1.3.",
- ),
- )
def __init__(
self,
class_: Type[_O],
local_table: Optional[FromClause] = None,
properties: Optional[Mapping[str, MapperProperty[Any]]] = None,
primary_key: Optional[Iterable[_ORMColumnExprArgument[Any]]] = None,
- non_primary: bool = False,
inherits: Optional[Union[Mapper[Any], Type[Any]]] = None,
inherit_condition: Optional[_ColumnExpressionArgument[bool]] = None,
inherit_foreign_keys: Optional[
See the change note and example at :ref:`legacy_is_orphan_addition`
for more detail on this change.
- :param non_primary: Specify that this :class:`_orm.Mapper`
- is in addition
- to the "primary" mapper, that is, the one used for persistence.
- The :class:`_orm.Mapper` created here may be used for ad-hoc
- mapping of the class to an alternate selectable, for loading
- only.
-
- .. seealso::
-
- :ref:`relationship_aliased_class` - the new pattern that removes
- the need for the :paramref:`_orm.Mapper.non_primary` flag.
-
:param passive_deletes: Indicates DELETE behavior of foreign key
columns when a joined-table inheritance entity is being deleted.
Defaults to ``False`` for a base mapper; for an inheriting mapper,
)
self._primary_key_argument = util.to_list(primary_key)
- self.non_primary = non_primary
self.always_refresh = always_refresh
"""
- non_primary: bool
- """Represent ``True`` if this :class:`_orm.Mapper` is a "non-primary"
- mapper, e.g. a mapper that is used only to select rows but not for
- persistence management.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- """
-
polymorphic_on: Optional[KeyedColumnElement[Any]]
"""The :class:`_schema.Column` or SQL expression specified as the
``polymorphic_on`` argument
self.dispatch._update(self.inherits.dispatch)
- if self.non_primary != self.inherits.non_primary:
- np = not self.non_primary and "primary" or "non-primary"
- raise sa_exc.ArgumentError(
- "Inheritance of %s mapper for class '%s' is "
- "only allowed from a %s mapper"
- % (np, self.class_.__name__, np)
- )
-
if self.single:
self.persist_selectable = self.inherits.persist_selectable
elif self.local_table is not self.inherits.local_table:
self._configure_polymorphic_setter(True)
def _configure_class_instrumentation(self):
- """If this mapper is to be a primary mapper (i.e. the
- non_primary flag is not set), associate this Mapper with the
+ """Associate this Mapper with the
given class and entity name.
Subsequent calls to ``class_mapper()`` for the ``class_`` / ``entity``
# this raises as of 2.0.
manager = attributes.opt_manager_of_class(self.class_)
- if self.non_primary:
- if not manager or not manager.is_mapped:
- raise sa_exc.InvalidRequestError(
- "Class %s has no primary mapper configured. Configure "
- "a primary mapper first before setting up a non primary "
- "Mapper." % self.class_
- )
- self.class_manager = manager
-
- assert manager.registry is not None
- self.registry = manager.registry
- self._identity_class = manager.mapper._identity_class
- manager.registry._add_non_primary_mapper(self)
- return
-
if manager is None or not manager.registry:
raise sa_exc.InvalidRequestError(
"The _mapper() function and Mapper() constructor may not be "
self._props[key] = prop
- if not self.non_primary:
- prop.instrument_class(self)
+ prop.instrument_class(self)
for mapper in self._inheriting_mappers:
mapper._adapt_inherited_property(key, prop, init)
and self.local_table.description
or str(self.local_table)
)
- + (self.non_primary and "|non-primary" or "")
+ ")"
)
return "<Mapper at 0x%x; %s>" % (id(self), self.class_.__name__)
def __str__(self) -> str:
- return "Mapper[%s%s(%s)]" % (
+ return "Mapper[%s(%s)]" % (
self.class_.__name__,
- self.non_primary and " (non-primary)" or "",
(
self.local_table.description
if self.local_table is not None
else:
reg._dispose_manager_and_mapper(manager)
- reg._non_primary_mappers.clear()
reg._dependents.clear()
for dep in reg._dependencies:
dep._dependents.discard(reg)
from sqlalchemy.orm import clear_mappers
from sqlalchemy.orm import collections
from sqlalchemy.orm import column_property
-from sqlalchemy.orm import configure_mappers
from sqlalchemy.orm import contains_alias
from sqlalchemy.orm import contains_eager
from sqlalchemy.orm import defaultload
from sqlalchemy.orm import subqueryload
from sqlalchemy.orm import synonym
from sqlalchemy.orm import undefer
-from sqlalchemy.orm import with_parent
from sqlalchemy.orm import with_polymorphic
from sqlalchemy.orm.collections import collection
from sqlalchemy.orm.strategy_options import lazyload
eq_(Sub._sa_converter(Sub(), 5), "sub_convert")
-class NonPrimaryRelationshipLoaderTest(_fixtures.FixtureTest):
- run_inserts = "once"
- run_deletes = None
-
- def test_selectload(self):
- """tests lazy loading with two relationships simultaneously,
- from the same table, using aliases."""
-
- users, orders, User, Address, Order, addresses = (
- self.tables.users,
- self.tables.orders,
- self.classes.User,
- self.classes.Address,
- self.classes.Order,
- self.tables.addresses,
- )
-
- openorders = sa.alias(orders, "openorders")
- closedorders = sa.alias(orders, "closedorders")
-
- self.mapper_registry.map_imperatively(Address, addresses)
-
- self.mapper_registry.map_imperatively(Order, orders)
-
- with testing.expect_deprecated(
- "The mapper.non_primary parameter is deprecated"
- ):
- open_mapper = self.mapper_registry.map_imperatively(
- Order, openorders, non_primary=True
- )
- closed_mapper = self.mapper_registry.map_imperatively(
- Order, closedorders, non_primary=True
- )
- self.mapper_registry.map_imperatively(
- User,
- users,
- properties=dict(
- addresses=relationship(Address, lazy=True),
- open_orders=relationship(
- open_mapper,
- primaryjoin=sa.and_(
- openorders.c.isopen == 1,
- users.c.id == openorders.c.user_id,
- ),
- lazy="select",
- ),
- closed_orders=relationship(
- closed_mapper,
- primaryjoin=sa.and_(
- closedorders.c.isopen == 0,
- users.c.id == closedorders.c.user_id,
- ),
- lazy="select",
- ),
- ),
- )
-
- self._run_double_test(10)
-
- def test_joinedload(self):
- """Eager loading with two relationships simultaneously,
- from the same table, using aliases."""
-
- users, orders, User, Address, Order, addresses = (
- self.tables.users,
- self.tables.orders,
- self.classes.User,
- self.classes.Address,
- self.classes.Order,
- self.tables.addresses,
- )
-
- openorders = sa.alias(orders, "openorders")
- closedorders = sa.alias(orders, "closedorders")
-
- self.mapper_registry.map_imperatively(Address, addresses)
- self.mapper_registry.map_imperatively(Order, orders)
-
- with testing.expect_deprecated(
- "The mapper.non_primary parameter is deprecated"
- ):
- open_mapper = self.mapper_registry.map_imperatively(
- Order, openorders, non_primary=True
- )
- closed_mapper = self.mapper_registry.map_imperatively(
- Order, closedorders, non_primary=True
- )
-
- self.mapper_registry.map_imperatively(
- User,
- users,
- properties=dict(
- addresses=relationship(
- Address, lazy="joined", order_by=addresses.c.id
- ),
- open_orders=relationship(
- open_mapper,
- primaryjoin=sa.and_(
- openorders.c.isopen == 1,
- users.c.id == openorders.c.user_id,
- ),
- lazy="joined",
- order_by=openorders.c.id,
- ),
- closed_orders=relationship(
- closed_mapper,
- primaryjoin=sa.and_(
- closedorders.c.isopen == 0,
- users.c.id == closedorders.c.user_id,
- ),
- lazy="joined",
- order_by=closedorders.c.id,
- ),
- ),
- )
- self._run_double_test(1)
-
- def test_selectin(self):
- users, orders, User, Address, Order, addresses = (
- self.tables.users,
- self.tables.orders,
- self.classes.User,
- self.classes.Address,
- self.classes.Order,
- self.tables.addresses,
- )
-
- openorders = sa.alias(orders, "openorders")
- closedorders = sa.alias(orders, "closedorders")
-
- self.mapper_registry.map_imperatively(Address, addresses)
- self.mapper_registry.map_imperatively(Order, orders)
-
- with testing.expect_deprecated(
- "The mapper.non_primary parameter is deprecated"
- ):
- open_mapper = self.mapper_registry.map_imperatively(
- Order, openorders, non_primary=True
- )
- closed_mapper = self.mapper_registry.map_imperatively(
- Order, closedorders, non_primary=True
- )
-
- self.mapper_registry.map_imperatively(
- User,
- users,
- properties=dict(
- addresses=relationship(
- Address, lazy="selectin", order_by=addresses.c.id
- ),
- open_orders=relationship(
- open_mapper,
- primaryjoin=sa.and_(
- openorders.c.isopen == 1,
- users.c.id == openorders.c.user_id,
- ),
- lazy="selectin",
- order_by=openorders.c.id,
- ),
- closed_orders=relationship(
- closed_mapper,
- primaryjoin=sa.and_(
- closedorders.c.isopen == 0,
- users.c.id == closedorders.c.user_id,
- ),
- lazy="selectin",
- order_by=closedorders.c.id,
- ),
- ),
- )
-
- self._run_double_test(4)
-
- def test_subqueryload(self):
- users, orders, User, Address, Order, addresses = (
- self.tables.users,
- self.tables.orders,
- self.classes.User,
- self.classes.Address,
- self.classes.Order,
- self.tables.addresses,
- )
-
- openorders = sa.alias(orders, "openorders")
- closedorders = sa.alias(orders, "closedorders")
-
- self.mapper_registry.map_imperatively(Address, addresses)
- self.mapper_registry.map_imperatively(Order, orders)
-
- with testing.expect_deprecated(
- "The mapper.non_primary parameter is deprecated"
- ):
- open_mapper = self.mapper_registry.map_imperatively(
- Order, openorders, non_primary=True
- )
- closed_mapper = self.mapper_registry.map_imperatively(
- Order, closedorders, non_primary=True
- )
-
- self.mapper_registry.map_imperatively(
- User,
- users,
- properties=dict(
- addresses=relationship(
- Address, lazy="subquery", order_by=addresses.c.id
- ),
- open_orders=relationship(
- open_mapper,
- primaryjoin=sa.and_(
- openorders.c.isopen == 1,
- users.c.id == openorders.c.user_id,
- ),
- lazy="subquery",
- order_by=openorders.c.id,
- ),
- closed_orders=relationship(
- closed_mapper,
- primaryjoin=sa.and_(
- closedorders.c.isopen == 0,
- users.c.id == closedorders.c.user_id,
- ),
- lazy="subquery",
- order_by=closedorders.c.id,
- ),
- ),
- )
-
- self._run_double_test(4)
-
- def _run_double_test(self, count):
- User, Address, Order, Item = self.classes(
- "User", "Address", "Order", "Item"
- )
- q = fixture_session().query(User).order_by(User.id)
-
- def go():
- eq_(
- [
- User(
- id=7,
- addresses=[Address(id=1)],
- open_orders=[Order(id=3)],
- closed_orders=[Order(id=1), Order(id=5)],
- ),
- User(
- id=8,
- addresses=[
- Address(id=2),
- Address(id=3),
- Address(id=4),
- ],
- open_orders=[],
- closed_orders=[],
- ),
- User(
- id=9,
- addresses=[Address(id=5)],
- open_orders=[Order(id=4)],
- closed_orders=[Order(id=2)],
- ),
- User(id=10),
- ],
- q.all(),
- )
-
- self.assert_sql_count(testing.db, go, count)
-
- sess = fixture_session()
- user = sess.get(User, 7)
-
- closed_mapper = User.closed_orders.entity
- open_mapper = User.open_orders.entity
- eq_(
- [Order(id=1), Order(id=5)],
- fixture_session()
- .query(closed_mapper)
- .filter(with_parent(user, User.closed_orders))
- .all(),
- )
- eq_(
- [Order(id=3)],
- fixture_session()
- .query(open_mapper)
- .filter(with_parent(user, User.open_orders))
- .all(),
- )
-
-
class ViewonlyFlagWarningTest(fixtures.MappedTest):
"""test for #4993.
eq_(getattr(rel, flag), value)
-class NonPrimaryMapperTest(_fixtures.FixtureTest, AssertsCompiledSQL):
- __dialect__ = "default"
-
- def teardown_test(self):
- clear_mappers()
-
- def test_non_primary_identity_class(self):
- User = self.classes.User
- users, addresses = self.tables.users, self.tables.addresses
-
- class AddressUser(User):
- pass
-
- self.mapper_registry.map_imperatively(
- User, users, polymorphic_identity="user"
- )
- m2 = self.mapper_registry.map_imperatively(
- AddressUser,
- addresses,
- inherits=User,
- polymorphic_identity="address",
- properties={"address_id": addresses.c.id},
- )
- with testing.expect_deprecated(
- "The mapper.non_primary parameter is deprecated"
- ):
- m3 = self.mapper_registry.map_imperatively(
- AddressUser, addresses, non_primary=True
- )
- assert m3._identity_class is m2._identity_class
- eq_(
- m2.identity_key_from_instance(AddressUser()),
- m3.identity_key_from_instance(AddressUser()),
- )
-
- def test_illegal_non_primary(self):
- users, Address, addresses, User = (
- self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User,
- )
-
- self.mapper_registry.map_imperatively(User, users)
- self.mapper_registry.map_imperatively(Address, addresses)
- with testing.expect_deprecated(
- "The mapper.non_primary parameter is deprecated"
- ):
- m = self.mapper_registry.map_imperatively( # noqa: F841
- User,
- users,
- non_primary=True,
- properties={"addresses": relationship(Address)},
- )
- assert_raises_message(
- sa.exc.ArgumentError,
- "Attempting to assign a new relationship 'addresses' "
- "to a non-primary mapper on class 'User'",
- configure_mappers,
- )
-
- def test_illegal_non_primary_2(self):
- User, users = self.classes.User, self.tables.users
-
- assert_raises_message(
- sa.exc.InvalidRequestError,
- "Configure a primary mapper first",
- self.mapper_registry.map_imperatively,
- User,
- users,
- non_primary=True,
- )
-
- def test_illegal_non_primary_3(self):
- users, addresses = self.tables.users, self.tables.addresses
-
- class Base:
- pass
-
- class Sub(Base):
- pass
-
- self.mapper_registry.map_imperatively(Base, users)
- assert_raises_message(
- sa.exc.InvalidRequestError,
- "Configure a primary mapper first",
- self.mapper_registry.map_imperatively,
- Sub,
- addresses,
- non_primary=True,
- )
-
- def test_illegal_non_primary_legacy(self, registry):
- users, Address, addresses, User = (
- self.tables.users,
- self.classes.Address,
- self.tables.addresses,
- self.classes.User,
- )
-
- registry.map_imperatively(User, users)
- registry.map_imperatively(Address, addresses)
- with testing.expect_deprecated(
- "The mapper.non_primary parameter is deprecated"
- ):
- m = registry.map_imperatively( # noqa: F841
- User,
- users,
- non_primary=True,
- properties={"addresses": relationship(Address)},
- )
- assert_raises_message(
- sa.exc.ArgumentError,
- "Attempting to assign a new relationship 'addresses' "
- "to a non-primary mapper on class 'User'",
- configure_mappers,
- )
-
- def test_illegal_non_primary_2_legacy(self, registry):
- User, users = self.classes.User, self.tables.users
-
- assert_raises_message(
- sa.exc.InvalidRequestError,
- "Configure a primary mapper first",
- registry.map_imperatively,
- User,
- users,
- non_primary=True,
- )
-
- def test_illegal_non_primary_3_legacy(self, registry):
- users, addresses = self.tables.users, self.tables.addresses
-
- class Base:
- pass
-
- class Sub(Base):
- pass
-
- registry.map_imperatively(Base, users)
-
- assert_raises_message(
- sa.exc.InvalidRequestError,
- "Configure a primary mapper first",
- registry.map_imperatively,
- Sub,
- addresses,
- non_primary=True,
- )
-
-
class InstancesTest(QueryTest, AssertsCompiledSQL):
@testing.fails(
"ORM refactor not allowing this yet, "
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
