summaryrefslogtreecommitdiff
path: root/venv/lib/python3.11/site-packages/sqlalchemy/orm/mapper.py
diff options
context:
space:
mode:
authorcyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
committercyfraeviolae <cyfraeviolae>2024-04-03 03:17:55 -0400
commit12cf076118570eebbff08c6b3090e0d4798447a1 (patch)
tree3ba25e17e3c3a5e82316558ba3864b955919ff72 /venv/lib/python3.11/site-packages/sqlalchemy/orm/mapper.py
parentc45662ff3923b34614ddcc8feb9195541166dcc5 (diff)
no venv
Diffstat (limited to 'venv/lib/python3.11/site-packages/sqlalchemy/orm/mapper.py')
-rw-r--r--venv/lib/python3.11/site-packages/sqlalchemy/orm/mapper.py4420
1 files changed, 0 insertions, 4420 deletions
diff --git a/venv/lib/python3.11/site-packages/sqlalchemy/orm/mapper.py b/venv/lib/python3.11/site-packages/sqlalchemy/orm/mapper.py
deleted file mode 100644
index 0caed0e..0000000
--- a/venv/lib/python3.11/site-packages/sqlalchemy/orm/mapper.py
+++ /dev/null
@@ -1,4420 +0,0 @@
-# orm/mapper.py
-# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors
-# <see AUTHORS file>
-#
-# This module is part of SQLAlchemy and is released under
-# the MIT License: https://www.opensource.org/licenses/mit-license.php
-# mypy: allow-untyped-defs, allow-untyped-calls
-
-"""Logic to map Python classes to and from selectables.
-
-Defines the :class:`~sqlalchemy.orm.mapper.Mapper` class, the central
-configurational unit which associates a class with a database table.
-
-This is a semi-private module; the main configurational API of the ORM is
-available in :class:`~sqlalchemy.orm.`.
-
-"""
-from __future__ import annotations
-
-from collections import deque
-from functools import reduce
-from itertools import chain
-import sys
-import threading
-from typing import Any
-from typing import Callable
-from typing import cast
-from typing import Collection
-from typing import Deque
-from typing import Dict
-from typing import FrozenSet
-from typing import Generic
-from typing import Iterable
-from typing import Iterator
-from typing import List
-from typing import Mapping
-from typing import Optional
-from typing import Sequence
-from typing import Set
-from typing import Tuple
-from typing import Type
-from typing import TYPE_CHECKING
-from typing import TypeVar
-from typing import Union
-import weakref
-
-from . import attributes
-from . import exc as orm_exc
-from . import instrumentation
-from . import loading
-from . import properties
-from . import util as orm_util
-from ._typing import _O
-from .base import _class_to_mapper
-from .base import _parse_mapper_argument
-from .base import _state_mapper
-from .base import PassiveFlag
-from .base import state_str
-from .interfaces import _MappedAttribute
-from .interfaces import EXT_SKIP
-from .interfaces import InspectionAttr
-from .interfaces import MapperProperty
-from .interfaces import ORMEntityColumnsClauseRole
-from .interfaces import ORMFromClauseRole
-from .interfaces import StrategizedProperty
-from .path_registry import PathRegistry
-from .. import event
-from .. import exc as sa_exc
-from .. import inspection
-from .. import log
-from .. import schema
-from .. import sql
-from .. import util
-from ..event import dispatcher
-from ..event import EventTarget
-from ..sql import base as sql_base
-from ..sql import coercions
-from ..sql import expression
-from ..sql import operators
-from ..sql import roles
-from ..sql import TableClause
-from ..sql import util as sql_util
-from ..sql import visitors
-from ..sql.cache_key import MemoizedHasCacheKey
-from ..sql.elements import KeyedColumnElement
-from ..sql.schema import Column
-from ..sql.schema import Table
-from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
-from ..util import HasMemoized
-from ..util import HasMemoized_ro_memoized_attribute
-from ..util.typing import Literal
-
-if TYPE_CHECKING:
- from ._typing import _IdentityKeyType
- from ._typing import _InstanceDict
- from ._typing import _ORMColumnExprArgument
- from ._typing import _RegistryType
- from .decl_api import registry
- from .dependency import DependencyProcessor
- from .descriptor_props import CompositeProperty
- from .descriptor_props import SynonymProperty
- from .events import MapperEvents
- from .instrumentation import ClassManager
- from .path_registry import CachingEntityRegistry
- from .properties import ColumnProperty
- from .relationships import RelationshipProperty
- from .state import InstanceState
- from .util import ORMAdapter
- from ..engine import Row
- from ..engine import RowMapping
- from ..sql._typing import _ColumnExpressionArgument
- from ..sql._typing import _EquivalentColumnMap
- from ..sql.base import ReadOnlyColumnCollection
- from ..sql.elements import ColumnClause
- from ..sql.elements import ColumnElement
- from ..sql.selectable import FromClause
- from ..util import OrderedSet
-
-
-_T = TypeVar("_T", bound=Any)
-_MP = TypeVar("_MP", bound="MapperProperty[Any]")
-_Fn = TypeVar("_Fn", bound="Callable[..., Any]")
-
-
-_WithPolymorphicArg = Union[
- Literal["*"],
- Tuple[
- Union[Literal["*"], Sequence[Union["Mapper[Any]", Type[Any]]]],
- Optional["FromClause"],
- ],
- Sequence[Union["Mapper[Any]", Type[Any]]],
-]
-
-
-_mapper_registries: weakref.WeakKeyDictionary[_RegistryType, bool] = (
- weakref.WeakKeyDictionary()
-)
-
-
-def _all_registries() -> Set[registry]:
- with _CONFIGURE_MUTEX:
- return set(_mapper_registries)
-
-
-def _unconfigured_mappers() -> Iterator[Mapper[Any]]:
- for reg in _all_registries():
- yield from reg._mappers_to_configure()
-
-
-_already_compiling = False
-
-
-# a constant returned by _get_attr_by_column to indicate
-# this mapper is not handling an attribute for a particular
-# column
-NO_ATTRIBUTE = util.symbol("NO_ATTRIBUTE")
-
-# lock used to synchronize the "mapper configure" step
-_CONFIGURE_MUTEX = threading.RLock()
-
-
-@inspection._self_inspects
-@log.class_logger
-class Mapper(
- ORMFromClauseRole,
- ORMEntityColumnsClauseRole[_O],
- MemoizedHasCacheKey,
- InspectionAttr,
- log.Identified,
- inspection.Inspectable["Mapper[_O]"],
- EventTarget,
- Generic[_O],
-):
- """Defines an association between a Python class and a database table or
- other relational structure, so that ORM operations against the class may
- proceed.
-
- The :class:`_orm.Mapper` object is instantiated using mapping methods
- present on the :class:`_orm.registry` object. For information
- about instantiating new :class:`_orm.Mapper` objects, see
- :ref:`orm_mapping_classes_toplevel`.
-
- """
-
- dispatch: dispatcher[Mapper[_O]]
-
- _dispose_called = False
- _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[
- Sequence[_ORMColumnExprArgument[Any]]
- ] = None,
- always_refresh: bool = False,
- version_id_col: Optional[_ORMColumnExprArgument[Any]] = None,
- version_id_generator: Optional[
- Union[Literal[False], Callable[[Any], Any]]
- ] = None,
- polymorphic_on: Optional[
- Union[_ORMColumnExprArgument[Any], str, MapperProperty[Any]]
- ] = None,
- _polymorphic_map: Optional[Dict[Any, Mapper[Any]]] = None,
- polymorphic_identity: Optional[Any] = None,
- concrete: bool = False,
- with_polymorphic: Optional[_WithPolymorphicArg] = None,
- polymorphic_abstract: bool = False,
- polymorphic_load: Optional[Literal["selectin", "inline"]] = None,
- allow_partial_pks: bool = True,
- batch: bool = True,
- column_prefix: Optional[str] = None,
- include_properties: Optional[Sequence[str]] = None,
- exclude_properties: Optional[Sequence[str]] = None,
- passive_updates: bool = True,
- passive_deletes: bool = False,
- confirm_deleted_rows: bool = True,
- eager_defaults: Literal[True, False, "auto"] = "auto",
- legacy_is_orphan: bool = False,
- _compiled_cache_size: int = 100,
- ):
- r"""Direct constructor for a new :class:`_orm.Mapper` object.
-
- The :class:`_orm.Mapper` constructor is not called directly, and
- is normally invoked through the
- use of the :class:`_orm.registry` object through either the
- :ref:`Declarative <orm_declarative_mapping>` or
- :ref:`Imperative <orm_imperative_mapping>` mapping styles.
-
- .. versionchanged:: 2.0 The public facing ``mapper()`` function is
- removed; for a classical mapping configuration, use the
- :meth:`_orm.registry.map_imperatively` method.
-
- Parameters documented below may be passed to either the
- :meth:`_orm.registry.map_imperatively` method, or may be passed in the
- ``__mapper_args__`` declarative class attribute described at
- :ref:`orm_declarative_mapper_options`.
-
- :param class\_: The class to be mapped. When using Declarative,
- this argument is automatically passed as the declared class
- itself.
-
- :param local_table: The :class:`_schema.Table` or other
- :class:`_sql.FromClause` (i.e. selectable) to which the class is
- mapped. May be ``None`` if this mapper inherits from another mapper
- using single-table inheritance. When using Declarative, this
- argument is automatically passed by the extension, based on what is
- configured via the :attr:`_orm.DeclarativeBase.__table__` attribute
- or via the :class:`_schema.Table` produced as a result of
- the :attr:`_orm.DeclarativeBase.__tablename__` attribute being
- present.
-
- :param polymorphic_abstract: Indicates this class will be mapped in a
- polymorphic hierarchy, but not directly instantiated. The class is
- mapped normally, except that it has no requirement for a
- :paramref:`_orm.Mapper.polymorphic_identity` within an inheritance
- hierarchy. The class however must be part of a polymorphic
- inheritance scheme which uses
- :paramref:`_orm.Mapper.polymorphic_on` at the base.
-
- .. versionadded:: 2.0
-
- .. seealso::
-
- :ref:`orm_inheritance_abstract_poly`
-
- :param always_refresh: If True, all query operations for this mapped
- class will overwrite all data within object instances that already
- exist within the session, erasing any in-memory changes with
- whatever information was loaded from the database. Usage of this
- flag is highly discouraged; as an alternative, see the method
- :meth:`_query.Query.populate_existing`.
-
- :param allow_partial_pks: Defaults to True. Indicates that a
- composite primary key with some NULL values should be considered as
- possibly existing within the database. This affects whether a
- mapper will assign an incoming row to an existing identity, as well
- as if :meth:`.Session.merge` will check the database first for a
- particular primary key value. A "partial primary key" can occur if
- one has mapped to an OUTER JOIN, for example.
-
- :param batch: Defaults to ``True``, indicating that save operations
- of multiple entities can be batched together for efficiency.
- Setting to False indicates
- that an instance will be fully saved before saving the next
- instance. This is used in the extremely rare case that a
- :class:`.MapperEvents` listener requires being called
- in between individual row persistence operations.
-
- :param column_prefix: A string which will be prepended
- to the mapped attribute name when :class:`_schema.Column`
- objects are automatically assigned as attributes to the
- mapped class. Does not affect :class:`.Column` objects that
- are mapped explicitly in the :paramref:`.Mapper.properties`
- dictionary.
-
- This parameter is typically useful with imperative mappings
- that keep the :class:`.Table` object separate. Below, assuming
- the ``user_table`` :class:`.Table` object has columns named
- ``user_id``, ``user_name``, and ``password``::
-
- class User(Base):
- __table__ = user_table
- __mapper_args__ = {'column_prefix':'_'}
-
- The above mapping will assign the ``user_id``, ``user_name``, and
- ``password`` columns to attributes named ``_user_id``,
- ``_user_name``, and ``_password`` on the mapped ``User`` class.
-
- The :paramref:`.Mapper.column_prefix` parameter is uncommon in
- modern use. For dealing with reflected tables, a more flexible
- approach to automating a naming scheme is to intercept the
- :class:`.Column` objects as they are reflected; see the section
- :ref:`mapper_automated_reflection_schemes` for notes on this usage
- pattern.
-
- :param concrete: If True, indicates this mapper should use concrete
- table inheritance with its parent mapper.
-
- See the section :ref:`concrete_inheritance` for an example.
-
- :param confirm_deleted_rows: defaults to True; when a DELETE occurs
- of one more rows based on specific primary keys, a warning is
- emitted when the number of rows matched does not equal the number
- of rows expected. This parameter may be set to False to handle the
- case where database ON DELETE CASCADE rules may be deleting some of
- those rows automatically. The warning may be changed to an
- exception in a future release.
-
- :param eager_defaults: if True, the ORM will immediately fetch the
- value of server-generated default values after an INSERT or UPDATE,
- rather than leaving them as expired to be fetched on next access.
- This can be used for event schemes where the server-generated values
- are needed immediately before the flush completes.
-
- The fetch of values occurs either by using ``RETURNING`` inline
- with the ``INSERT`` or ``UPDATE`` statement, or by adding an
- additional ``SELECT`` statement subsequent to the ``INSERT`` or
- ``UPDATE``, if the backend does not support ``RETURNING``.
-
- The use of ``RETURNING`` is extremely performant in particular for
- ``INSERT`` statements where SQLAlchemy can take advantage of
- :ref:`insertmanyvalues <engine_insertmanyvalues>`, whereas the use of
- an additional ``SELECT`` is relatively poor performing, adding
- additional SQL round trips which would be unnecessary if these new
- attributes are not to be accessed in any case.
-
- For this reason, :paramref:`.Mapper.eager_defaults` defaults to the
- string value ``"auto"``, which indicates that server defaults for
- INSERT should be fetched using ``RETURNING`` if the backing database
- supports it and if the dialect in use supports "insertmanyreturning"
- for an INSERT statement. If the backing database does not support
- ``RETURNING`` or "insertmanyreturning" is not available, server
- defaults will not be fetched.
-
- .. versionchanged:: 2.0.0rc1 added the "auto" option for
- :paramref:`.Mapper.eager_defaults`
-
- .. seealso::
-
- :ref:`orm_server_defaults`
-
- .. versionchanged:: 2.0.0 RETURNING now works with multiple rows
- INSERTed at once using the
- :ref:`insertmanyvalues <engine_insertmanyvalues>` feature, which
- among other things allows the :paramref:`.Mapper.eager_defaults`
- feature to be very performant on supporting backends.
-
- :param exclude_properties: A list or set of string column names to
- be excluded from mapping.
-
- .. seealso::
-
- :ref:`include_exclude_cols`
-
- :param include_properties: An inclusive list or set of string column
- names to map.
-
- .. seealso::
-
- :ref:`include_exclude_cols`
-
- :param inherits: A mapped class or the corresponding
- :class:`_orm.Mapper`
- of one indicating a superclass to which this :class:`_orm.Mapper`
- should *inherit* from. The mapped class here must be a subclass
- of the other mapper's class. When using Declarative, this argument
- is passed automatically as a result of the natural class
- hierarchy of the declared classes.
-
- .. seealso::
-
- :ref:`inheritance_toplevel`
-
- :param inherit_condition: For joined table inheritance, a SQL
- expression which will
- define how the two tables are joined; defaults to a natural join
- between the two tables.
-
- :param inherit_foreign_keys: When ``inherit_condition`` is used and
- the columns present are missing a :class:`_schema.ForeignKey`
- configuration, this parameter can be used to specify which columns
- are "foreign". In most cases can be left as ``None``.
-
- :param legacy_is_orphan: Boolean, defaults to ``False``.
- When ``True``, specifies that "legacy" orphan consideration
- is to be applied to objects mapped by this mapper, which means
- that a pending (that is, not persistent) object is auto-expunged
- from an owning :class:`.Session` only when it is de-associated
- from *all* parents that specify a ``delete-orphan`` cascade towards
- this mapper. The new default behavior is that the object is
- auto-expunged when it is de-associated with *any* of its parents
- that specify ``delete-orphan`` cascade. This behavior is more
- consistent with that of a persistent object, and allows behavior to
- be consistent in more scenarios independently of whether or not an
- orphan object has been flushed yet or not.
-
- 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,
- defaults to ``False`` unless the value is set to ``True``
- on the superclass mapper.
-
- When ``True``, it is assumed that ON DELETE CASCADE is configured
- on the foreign key relationships that link this mapper's table
- to its superclass table, so that when the unit of work attempts
- to delete the entity, it need only emit a DELETE statement for the
- superclass table, and not this table.
-
- When ``False``, a DELETE statement is emitted for this mapper's
- table individually. If the primary key attributes local to this
- table are unloaded, then a SELECT must be emitted in order to
- validate these attributes; note that the primary key columns
- of a joined-table subclass are not part of the "primary key" of
- the object as a whole.
-
- Note that a value of ``True`` is **always** forced onto the
- subclass mappers; that is, it's not possible for a superclass
- to specify passive_deletes without this taking effect for
- all subclass mappers.
-
- .. seealso::
-
- :ref:`passive_deletes` - description of similar feature as
- used with :func:`_orm.relationship`
-
- :paramref:`.mapper.passive_updates` - supporting ON UPDATE
- CASCADE for joined-table inheritance mappers
-
- :param passive_updates: Indicates UPDATE behavior of foreign key
- columns when a primary key column changes on a joined-table
- inheritance mapping. Defaults to ``True``.
-
- When True, it is assumed that ON UPDATE CASCADE is configured on
- the foreign key in the database, and that the database will handle
- propagation of an UPDATE from a source column to dependent columns
- on joined-table rows.
-
- When False, it is assumed that the database does not enforce
- referential integrity and will not be issuing its own CASCADE
- operation for an update. The unit of work process will
- emit an UPDATE statement for the dependent columns during a
- primary key change.
-
- .. seealso::
-
- :ref:`passive_updates` - description of a similar feature as
- used with :func:`_orm.relationship`
-
- :paramref:`.mapper.passive_deletes` - supporting ON DELETE
- CASCADE for joined-table inheritance mappers
-
- :param polymorphic_load: Specifies "polymorphic loading" behavior
- for a subclass in an inheritance hierarchy (joined and single
- table inheritance only). Valid values are:
-
- * "'inline'" - specifies this class should be part of
- the "with_polymorphic" mappers, e.g. its columns will be included
- in a SELECT query against the base.
-
- * "'selectin'" - specifies that when instances of this class
- are loaded, an additional SELECT will be emitted to retrieve
- the columns specific to this subclass. The SELECT uses
- IN to fetch multiple subclasses at once.
-
- .. versionadded:: 1.2
-
- .. seealso::
-
- :ref:`with_polymorphic_mapper_config`
-
- :ref:`polymorphic_selectin`
-
- :param polymorphic_on: Specifies the column, attribute, or
- SQL expression used to determine the target class for an
- incoming row, when inheriting classes are present.
-
- May be specified as a string attribute name, or as a SQL
- expression such as a :class:`_schema.Column` or in a Declarative
- mapping a :func:`_orm.mapped_column` object. It is typically
- expected that the SQL expression corresponds to a column in the
- base-most mapped :class:`.Table`::
-
- class Employee(Base):
- __tablename__ = 'employee'
-
- id: Mapped[int] = mapped_column(primary_key=True)
- discriminator: Mapped[str] = mapped_column(String(50))
-
- __mapper_args__ = {
- "polymorphic_on":discriminator,
- "polymorphic_identity":"employee"
- }
-
- It may also be specified
- as a SQL expression, as in this example where we
- use the :func:`.case` construct to provide a conditional
- approach::
-
- class Employee(Base):
- __tablename__ = 'employee'
-
- id: Mapped[int] = mapped_column(primary_key=True)
- discriminator: Mapped[str] = mapped_column(String(50))
-
- __mapper_args__ = {
- "polymorphic_on":case(
- (discriminator == "EN", "engineer"),
- (discriminator == "MA", "manager"),
- else_="employee"),
- "polymorphic_identity":"employee"
- }
-
- It may also refer to any attribute using its string name,
- which is of particular use when using annotated column
- configurations::
-
- class Employee(Base):
- __tablename__ = 'employee'
-
- id: Mapped[int] = mapped_column(primary_key=True)
- discriminator: Mapped[str]
-
- __mapper_args__ = {
- "polymorphic_on": "discriminator",
- "polymorphic_identity": "employee"
- }
-
- When setting ``polymorphic_on`` to reference an
- attribute or expression that's not present in the
- locally mapped :class:`_schema.Table`, yet the value
- of the discriminator should be persisted to the database,
- the value of the
- discriminator is not automatically set on new
- instances; this must be handled by the user,
- either through manual means or via event listeners.
- A typical approach to establishing such a listener
- looks like::
-
- from sqlalchemy import event
- from sqlalchemy.orm import object_mapper
-
- @event.listens_for(Employee, "init", propagate=True)
- def set_identity(instance, *arg, **kw):
- mapper = object_mapper(instance)
- instance.discriminator = mapper.polymorphic_identity
-
- Where above, we assign the value of ``polymorphic_identity``
- for the mapped class to the ``discriminator`` attribute,
- thus persisting the value to the ``discriminator`` column
- in the database.
-
- .. warning::
-
- Currently, **only one discriminator column may be set**, typically
- on the base-most class in the hierarchy. "Cascading" polymorphic
- columns are not yet supported.
-
- .. seealso::
-
- :ref:`inheritance_toplevel`
-
- :param polymorphic_identity: Specifies the value which
- identifies this particular class as returned by the column expression
- referred to by the :paramref:`_orm.Mapper.polymorphic_on` setting. As
- rows are received, the value corresponding to the
- :paramref:`_orm.Mapper.polymorphic_on` column expression is compared
- to this value, indicating which subclass should be used for the newly
- reconstructed object.
-
- .. seealso::
-
- :ref:`inheritance_toplevel`
-
- :param properties: A dictionary mapping the string names of object
- attributes to :class:`.MapperProperty` instances, which define the
- persistence behavior of that attribute. Note that
- :class:`_schema.Column`
- objects present in
- the mapped :class:`_schema.Table` are automatically placed into
- ``ColumnProperty`` instances upon mapping, unless overridden.
- When using Declarative, this argument is passed automatically,
- based on all those :class:`.MapperProperty` instances declared
- in the declared class body.
-
- .. seealso::
-
- :ref:`orm_mapping_properties` - in the
- :ref:`orm_mapping_classes_toplevel`
-
- :param primary_key: A list of :class:`_schema.Column`
- objects, or alternatively string names of attribute names which
- refer to :class:`_schema.Column`, which define
- the primary key to be used against this mapper's selectable unit.
- This is normally simply the primary key of the ``local_table``, but
- can be overridden here.
-
- .. versionchanged:: 2.0.2 :paramref:`_orm.Mapper.primary_key`
- arguments may be indicated as string attribute names as well.
-
- .. seealso::
-
- :ref:`mapper_primary_key` - background and example use
-
- :param version_id_col: A :class:`_schema.Column`
- that will be used to keep a running version id of rows
- in the table. This is used to detect concurrent updates or
- the presence of stale data in a flush. The methodology is to
- detect if an UPDATE statement does not match the last known
- version id, a
- :class:`~sqlalchemy.orm.exc.StaleDataError` exception is
- thrown.
- By default, the column must be of :class:`.Integer` type,
- unless ``version_id_generator`` specifies an alternative version
- generator.
-
- .. seealso::
-
- :ref:`mapper_version_counter` - discussion of version counting
- and rationale.
-
- :param version_id_generator: Define how new version ids should
- be generated. Defaults to ``None``, which indicates that
- a simple integer counting scheme be employed. To provide a custom
- versioning scheme, provide a callable function of the form::
-
- def generate_version(version):
- return next_version
-
- Alternatively, server-side versioning functions such as triggers,
- or programmatic versioning schemes outside of the version id
- generator may be used, by specifying the value ``False``.
- Please see :ref:`server_side_version_counter` for a discussion
- of important points when using this option.
-
- .. seealso::
-
- :ref:`custom_version_counter`
-
- :ref:`server_side_version_counter`
-
-
- :param with_polymorphic: A tuple in the form ``(<classes>,
- <selectable>)`` indicating the default style of "polymorphic"
- loading, that is, which tables are queried at once. <classes> is
- any single or list of mappers and/or classes indicating the
- inherited classes that should be loaded at once. The special value
- ``'*'`` may be used to indicate all descending classes should be
- loaded immediately. The second tuple argument <selectable>
- indicates a selectable that will be used to query for multiple
- classes.
-
- The :paramref:`_orm.Mapper.polymorphic_load` parameter may be
- preferable over the use of :paramref:`_orm.Mapper.with_polymorphic`
- in modern mappings to indicate a per-subclass technique of
- indicating polymorphic loading styles.
-
- .. seealso::
-
- :ref:`with_polymorphic_mapper_config`
-
- """
- self.class_ = util.assert_arg_type(class_, type, "class_")
- self._sort_key = "%s.%s" % (
- self.class_.__module__,
- self.class_.__name__,
- )
-
- self._primary_key_argument = util.to_list(primary_key)
- self.non_primary = non_primary
-
- self.always_refresh = always_refresh
-
- if isinstance(version_id_col, MapperProperty):
- self.version_id_prop = version_id_col
- self.version_id_col = None
- else:
- self.version_id_col = (
- coercions.expect(
- roles.ColumnArgumentOrKeyRole,
- version_id_col,
- argname="version_id_col",
- )
- if version_id_col is not None
- else None
- )
-
- if version_id_generator is False:
- self.version_id_generator = False
- elif version_id_generator is None:
- self.version_id_generator = lambda x: (x or 0) + 1
- else:
- self.version_id_generator = version_id_generator
-
- self.concrete = concrete
- self.single = False
-
- if inherits is not None:
- self.inherits = _parse_mapper_argument(inherits)
- else:
- self.inherits = None
-
- if local_table is not None:
- self.local_table = coercions.expect(
- roles.StrictFromClauseRole,
- local_table,
- disable_inspection=True,
- argname="local_table",
- )
- elif self.inherits:
- # note this is a new flow as of 2.0 so that
- # .local_table need not be Optional
- self.local_table = self.inherits.local_table
- self.single = True
- else:
- raise sa_exc.ArgumentError(
- f"Mapper[{self.class_.__name__}(None)] has None for a "
- "primary table argument and does not specify 'inherits'"
- )
-
- if inherit_condition is not None:
- self.inherit_condition = coercions.expect(
- roles.OnClauseRole, inherit_condition
- )
- else:
- self.inherit_condition = None
-
- self.inherit_foreign_keys = inherit_foreign_keys
- self._init_properties = dict(properties) if properties else {}
- self._delete_orphans = []
- self.batch = batch
- self.eager_defaults = eager_defaults
- self.column_prefix = column_prefix
-
- # interim - polymorphic_on is further refined in
- # _configure_polymorphic_setter
- self.polymorphic_on = (
- coercions.expect( # type: ignore
- roles.ColumnArgumentOrKeyRole,
- polymorphic_on,
- argname="polymorphic_on",
- )
- if polymorphic_on is not None
- else None
- )
- self.polymorphic_abstract = polymorphic_abstract
- self._dependency_processors = []
- self.validators = util.EMPTY_DICT
- self.passive_updates = passive_updates
- self.passive_deletes = passive_deletes
- self.legacy_is_orphan = legacy_is_orphan
- self._clause_adapter = None
- self._requires_row_aliasing = False
- self._inherits_equated_pairs = None
- self._memoized_values = {}
- self._compiled_cache_size = _compiled_cache_size
- self._reconstructor = None
- self.allow_partial_pks = allow_partial_pks
-
- if self.inherits and not self.concrete:
- self.confirm_deleted_rows = False
- else:
- self.confirm_deleted_rows = confirm_deleted_rows
-
- self._set_with_polymorphic(with_polymorphic)
- self.polymorphic_load = polymorphic_load
-
- # our 'polymorphic identity', a string name that when located in a
- # result set row indicates this Mapper should be used to construct
- # the object instance for that row.
- self.polymorphic_identity = polymorphic_identity
-
- # a dictionary of 'polymorphic identity' names, associating those
- # names with Mappers that will be used to construct object instances
- # upon a select operation.
- if _polymorphic_map is None:
- self.polymorphic_map = {}
- else:
- self.polymorphic_map = _polymorphic_map
-
- if include_properties is not None:
- self.include_properties = util.to_set(include_properties)
- else:
- self.include_properties = None
- if exclude_properties:
- self.exclude_properties = util.to_set(exclude_properties)
- else:
- self.exclude_properties = None
-
- # prevent this mapper from being constructed
- # while a configure_mappers() is occurring (and defer a
- # configure_mappers() until construction succeeds)
- with _CONFIGURE_MUTEX:
- cast("MapperEvents", self.dispatch._events)._new_mapper_instance(
- class_, self
- )
- self._configure_inheritance()
- self._configure_class_instrumentation()
- self._configure_properties()
- self._configure_polymorphic_setter()
- self._configure_pks()
- self.registry._flag_new_mapper(self)
- 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:
- return False
-
- return (
- table in self._server_default_col_keys
- and dialect.insert_executemany_returning
- )
- else:
- return self.eager_defaults
-
- def _gen_cache_key(self, anon_map, bindparams):
- return (self,)
-
- # ### BEGIN
- # ATTRIBUTE DECLARATIONS START HERE
-
- is_mapper = True
- """Part of the inspection API."""
-
- represents_outer_join = False
-
- registry: _RegistryType
-
- @property
- def mapper(self) -> Mapper[_O]:
- """Part of the inspection API.
-
- Returns self.
-
- """
- return self
-
- @property
- def entity(self):
- r"""Part of the inspection API.
-
- Returns self.class\_.
-
- """
- return self.class_
-
- class_: Type[_O]
- """The class to which this :class:`_orm.Mapper` is mapped."""
-
- _identity_class: Type[_O]
-
- _delete_orphans: List[Tuple[str, Type[Any]]]
- _dependency_processors: List[DependencyProcessor]
- _memoized_values: Dict[Any, Callable[[], Any]]
- _inheriting_mappers: util.WeakSequence[Mapper[Any]]
- _all_tables: Set[TableClause]
- _polymorphic_attr_key: Optional[str]
-
- _pks_by_table: Dict[FromClause, OrderedSet[ColumnClause[Any]]]
- _cols_by_table: Dict[FromClause, OrderedSet[ColumnElement[Any]]]
-
- _props: util.OrderedDict[str, MapperProperty[Any]]
- _init_properties: Dict[str, MapperProperty[Any]]
-
- _columntoproperty: _ColumnMapping
-
- _set_polymorphic_identity: Optional[Callable[[InstanceState[_O]], None]]
- _validate_polymorphic_identity: Optional[
- Callable[[Mapper[_O], InstanceState[_O], _InstanceDict], None]
- ]
-
- tables: Sequence[TableClause]
- """A sequence containing the collection of :class:`_schema.Table`
- or :class:`_schema.TableClause` objects which this :class:`_orm.Mapper`
- is aware of.
-
- If the mapper is mapped to a :class:`_expression.Join`, or an
- :class:`_expression.Alias`
- representing a :class:`_expression.Select`, the individual
- :class:`_schema.Table`
- objects that comprise the full construct will be represented here.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- """
-
- validators: util.immutabledict[str, Tuple[str, Dict[str, Any]]]
- """An immutable dictionary of attributes which have been decorated
- using the :func:`_orm.validates` decorator.
-
- The dictionary contains string attribute names as keys
- mapped to the actual validation method.
-
- """
-
- always_refresh: bool
- allow_partial_pks: bool
- version_id_col: Optional[ColumnElement[Any]]
-
- with_polymorphic: Optional[
- Tuple[
- Union[Literal["*"], Sequence[Union[Mapper[Any], Type[Any]]]],
- Optional[FromClause],
- ]
- ]
-
- version_id_generator: Optional[Union[Literal[False], Callable[[Any], Any]]]
-
- local_table: FromClause
- """The immediate :class:`_expression.FromClause` to which this
- :class:`_orm.Mapper` refers.
-
- Typically is an instance of :class:`_schema.Table`, may be any
- :class:`.FromClause`.
-
- The "local" table is the
- selectable that the :class:`_orm.Mapper` is directly responsible for
- managing from an attribute access and flush perspective. For
- non-inheriting mappers, :attr:`.Mapper.local_table` will be the same
- as :attr:`.Mapper.persist_selectable`. For inheriting mappers,
- :attr:`.Mapper.local_table` refers to the specific portion of
- :attr:`.Mapper.persist_selectable` that includes the columns to which
- this :class:`.Mapper` is loading/persisting, such as a particular
- :class:`.Table` within a join.
-
- .. seealso::
-
- :attr:`_orm.Mapper.persist_selectable`.
-
- :attr:`_orm.Mapper.selectable`.
-
- """
-
- persist_selectable: FromClause
- """The :class:`_expression.FromClause` to which this :class:`_orm.Mapper`
- is mapped.
-
- Typically is an instance of :class:`_schema.Table`, may be any
- :class:`.FromClause`.
-
- The :attr:`_orm.Mapper.persist_selectable` is similar to
- :attr:`.Mapper.local_table`, but represents the :class:`.FromClause` that
- represents the inheriting class hierarchy overall in an inheritance
- scenario.
-
- :attr.`.Mapper.persist_selectable` is also separate from the
- :attr:`.Mapper.selectable` attribute, the latter of which may be an
- alternate subquery used for selecting columns.
- :attr.`.Mapper.persist_selectable` is oriented towards columns that
- will be written on a persist operation.
-
- .. seealso::
-
- :attr:`_orm.Mapper.selectable`.
-
- :attr:`_orm.Mapper.local_table`.
-
- """
-
- inherits: Optional[Mapper[Any]]
- """References the :class:`_orm.Mapper` which this :class:`_orm.Mapper`
- inherits from, if any.
-
- """
-
- inherit_condition: Optional[ColumnElement[bool]]
-
- configured: bool = False
- """Represent ``True`` if this :class:`_orm.Mapper` has been configured.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- .. seealso::
-
- :func:`.configure_mappers`.
-
- """
-
- concrete: bool
- """Represent ``True`` if this :class:`_orm.Mapper` is a concrete
- inheritance mapper.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- """
-
- primary_key: Tuple[Column[Any], ...]
- """An iterable containing the collection of :class:`_schema.Column`
- objects
- which comprise the 'primary key' of the mapped table, from the
- perspective of this :class:`_orm.Mapper`.
-
- This list is against the selectable in
- :attr:`_orm.Mapper.persist_selectable`.
- In the case of inheriting mappers, some columns may be managed by a
- superclass mapper. For example, in the case of a
- :class:`_expression.Join`, the
- primary key is determined by all of the primary key columns across all
- tables referenced by the :class:`_expression.Join`.
-
- The list is also not necessarily the same as the primary key column
- collection associated with the underlying tables; the :class:`_orm.Mapper`
- features a ``primary_key`` argument that can override what the
- :class:`_orm.Mapper` considers as primary key columns.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- """
-
- class_manager: ClassManager[_O]
- """The :class:`.ClassManager` which maintains event listeners
- and class-bound descriptors for this :class:`_orm.Mapper`.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- """
-
- single: bool
- """Represent ``True`` if this :class:`_orm.Mapper` is a single table
- inheritance mapper.
-
- :attr:`_orm.Mapper.local_table` will be ``None`` if this flag is set.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- """
-
- 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
- for this :class:`_orm.Mapper`, within an inheritance scenario.
-
- This attribute is normally a :class:`_schema.Column` instance but
- may also be an expression, such as one derived from
- :func:`.cast`.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- """
-
- polymorphic_map: Dict[Any, Mapper[Any]]
- """A mapping of "polymorphic identity" identifiers mapped to
- :class:`_orm.Mapper` instances, within an inheritance scenario.
-
- The identifiers can be of any type which is comparable to the
- type of column represented by :attr:`_orm.Mapper.polymorphic_on`.
-
- An inheritance chain of mappers will all reference the same
- polymorphic map object. The object is used to correlate incoming
- result rows to target mappers.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- """
-
- polymorphic_identity: Optional[Any]
- """Represent an identifier which is matched against the
- :attr:`_orm.Mapper.polymorphic_on` column during result row loading.
-
- Used only with inheritance, this object can be of any type which is
- comparable to the type of column represented by
- :attr:`_orm.Mapper.polymorphic_on`.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- """
-
- base_mapper: Mapper[Any]
- """The base-most :class:`_orm.Mapper` in an inheritance chain.
-
- In a non-inheriting scenario, this attribute will always be this
- :class:`_orm.Mapper`. In an inheritance scenario, it references
- the :class:`_orm.Mapper` which is parent to all other :class:`_orm.Mapper`
- objects in the inheritance chain.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- """
-
- columns: ReadOnlyColumnCollection[str, Column[Any]]
- """A collection of :class:`_schema.Column` or other scalar expression
- objects maintained by this :class:`_orm.Mapper`.
-
- The collection behaves the same as that of the ``c`` attribute on
- any :class:`_schema.Table` object,
- except that only those columns included in
- this mapping are present, and are keyed based on the attribute name
- defined in the mapping, not necessarily the ``key`` attribute of the
- :class:`_schema.Column` itself. Additionally, scalar expressions mapped
- by :func:`.column_property` are also present here.
-
- This is a *read only* attribute determined during mapper construction.
- Behavior is undefined if directly modified.
-
- """
-
- c: ReadOnlyColumnCollection[str, Column[Any]]
- """A synonym for :attr:`_orm.Mapper.columns`."""
-
- @util.non_memoized_property
- @util.deprecated("1.3", "Use .persist_selectable")
- def mapped_table(self):
- return self.persist_selectable
-
- @util.memoized_property
- def _path_registry(self) -> CachingEntityRegistry:
- return PathRegistry.per_mapper(self)
-
- def _configure_inheritance(self):
- """Configure settings related to inheriting and/or inherited mappers
- being present."""
-
- # a set of all mappers which inherit from this one.
- self._inheriting_mappers = util.WeakSequence()
-
- if self.inherits:
- if not issubclass(self.class_, self.inherits.class_):
- raise sa_exc.ArgumentError(
- "Class '%s' does not inherit from '%s'"
- % (self.class_.__name__, self.inherits.class_.__name__)
- )
-
- 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:
- if self.concrete:
- self.persist_selectable = self.local_table
- for mapper in self.iterate_to_root():
- if mapper.polymorphic_on is not None:
- mapper._requires_row_aliasing = True
- else:
- if self.inherit_condition is None:
- # figure out inherit condition from our table to the
- # immediate table of the inherited mapper, not its
- # full table which could pull in other stuff we don't
- # want (allows test/inheritance.InheritTest4 to pass)
- try:
- self.inherit_condition = sql_util.join_condition(
- self.inherits.local_table, self.local_table
- )
- except sa_exc.NoForeignKeysError as nfe:
- assert self.inherits.local_table is not None
- assert self.local_table is not None
- raise sa_exc.NoForeignKeysError(
- "Can't determine the inherit condition "
- "between inherited table '%s' and "
- "inheriting "
- "table '%s'; tables have no "
- "foreign key relationships established. "
- "Please ensure the inheriting table has "
- "a foreign key relationship to the "
- "inherited "
- "table, or provide an "
- "'on clause' using "
- "the 'inherit_condition' mapper argument."
- % (
- self.inherits.local_table.description,
- self.local_table.description,
- )
- ) from nfe
- except sa_exc.AmbiguousForeignKeysError as afe:
- assert self.inherits.local_table is not None
- assert self.local_table is not None
- raise sa_exc.AmbiguousForeignKeysError(
- "Can't determine the inherit condition "
- "between inherited table '%s' and "
- "inheriting "
- "table '%s'; tables have more than one "
- "foreign key relationship established. "
- "Please specify the 'on clause' using "
- "the 'inherit_condition' mapper argument."
- % (
- self.inherits.local_table.description,
- self.local_table.description,
- )
- ) from afe
- assert self.inherits.persist_selectable is not None
- self.persist_selectable = sql.join(
- self.inherits.persist_selectable,
- self.local_table,
- self.inherit_condition,
- )
-
- fks = util.to_set(self.inherit_foreign_keys)
- self._inherits_equated_pairs = sql_util.criterion_as_pairs(
- self.persist_selectable.onclause,
- consider_as_foreign_keys=fks,
- )
- else:
- self.persist_selectable = self.local_table
-
- if self.polymorphic_identity is None:
- self._identity_class = self.class_
-
- if (
- not self.polymorphic_abstract
- and self.inherits.base_mapper.polymorphic_on is not None
- ):
- util.warn(
- f"{self} does not indicate a 'polymorphic_identity', "
- "yet is part of an inheritance hierarchy that has a "
- f"'polymorphic_on' column of "
- f"'{self.inherits.base_mapper.polymorphic_on}'. "
- "If this is an intermediary class that should not be "
- "instantiated, the class may either be left unmapped, "
- "or may include the 'polymorphic_abstract=True' "
- "parameter in its Mapper arguments. To leave the "
- "class unmapped when using Declarative, set the "
- "'__abstract__ = True' attribute on the class."
- )
- elif self.concrete:
- self._identity_class = self.class_
- else:
- self._identity_class = self.inherits._identity_class
-
- if self.version_id_col is None:
- self.version_id_col = self.inherits.version_id_col
- self.version_id_generator = self.inherits.version_id_generator
- elif (
- self.inherits.version_id_col is not None
- and self.version_id_col is not self.inherits.version_id_col
- ):
- util.warn(
- "Inheriting version_id_col '%s' does not match inherited "
- "version_id_col '%s' and will not automatically populate "
- "the inherited versioning column. "
- "version_id_col should only be specified on "
- "the base-most mapper that includes versioning."
- % (
- self.version_id_col.description,
- self.inherits.version_id_col.description,
- )
- )
-
- self.polymorphic_map = self.inherits.polymorphic_map
- self.batch = self.inherits.batch
- self.inherits._inheriting_mappers.append(self)
- self.base_mapper = self.inherits.base_mapper
- self.passive_updates = self.inherits.passive_updates
- self.passive_deletes = (
- self.inherits.passive_deletes or self.passive_deletes
- )
- self._all_tables = self.inherits._all_tables
-
- if self.polymorphic_identity is not None:
- if self.polymorphic_identity in self.polymorphic_map:
- util.warn(
- "Reassigning polymorphic association for identity %r "
- "from %r to %r: Check for duplicate use of %r as "
- "value for polymorphic_identity."
- % (
- self.polymorphic_identity,
- self.polymorphic_map[self.polymorphic_identity],
- self,
- self.polymorphic_identity,
- )
- )
- self.polymorphic_map[self.polymorphic_identity] = self
-
- if self.polymorphic_load and self.concrete:
- raise sa_exc.ArgumentError(
- "polymorphic_load is not currently supported "
- "with concrete table inheritance"
- )
- if self.polymorphic_load == "inline":
- self.inherits._add_with_polymorphic_subclass(self)
- elif self.polymorphic_load == "selectin":
- pass
- elif self.polymorphic_load is not None:
- raise sa_exc.ArgumentError(
- "unknown argument for polymorphic_load: %r"
- % self.polymorphic_load
- )
-
- else:
- self._all_tables = set()
- self.base_mapper = self
- assert self.local_table is not None
- self.persist_selectable = self.local_table
- if self.polymorphic_identity is not None:
- self.polymorphic_map[self.polymorphic_identity] = self
- self._identity_class = self.class_
-
- if self.persist_selectable is None:
- raise sa_exc.ArgumentError(
- "Mapper '%s' does not have a persist_selectable specified."
- % self
- )
-
- def _set_with_polymorphic(
- self, with_polymorphic: Optional[_WithPolymorphicArg]
- ) -> None:
- if with_polymorphic == "*":
- self.with_polymorphic = ("*", None)
- elif isinstance(with_polymorphic, (tuple, list)):
- if isinstance(with_polymorphic[0], (str, tuple, list)):
- self.with_polymorphic = cast(
- """Tuple[
- Union[
- Literal["*"],
- Sequence[Union["Mapper[Any]", Type[Any]]],
- ],
- Optional["FromClause"],
- ]""",
- with_polymorphic,
- )
- else:
- self.with_polymorphic = (with_polymorphic, None)
- elif with_polymorphic is not None:
- raise sa_exc.ArgumentError(
- f"Invalid setting for with_polymorphic: {with_polymorphic!r}"
- )
- else:
- self.with_polymorphic = None
-
- if self.with_polymorphic and self.with_polymorphic[1] is not None:
- self.with_polymorphic = (
- self.with_polymorphic[0],
- coercions.expect(
- roles.StrictFromClauseRole,
- self.with_polymorphic[1],
- allow_select=True,
- ),
- )
-
- if self.configured:
- self._expire_memoizations()
-
- def _add_with_polymorphic_subclass(self, mapper):
- subcl = mapper.class_
- if self.with_polymorphic is None:
- self._set_with_polymorphic((subcl,))
- elif self.with_polymorphic[0] != "*":
- assert isinstance(self.with_polymorphic[0], tuple)
- self._set_with_polymorphic(
- (self.with_polymorphic[0] + (subcl,), self.with_polymorphic[1])
- )
-
- def _set_concrete_base(self, mapper):
- """Set the given :class:`_orm.Mapper` as the 'inherits' for this
- :class:`_orm.Mapper`, assuming this :class:`_orm.Mapper` is concrete
- and does not already have an inherits."""
-
- assert self.concrete
- assert not self.inherits
- assert isinstance(mapper, Mapper)
- self.inherits = mapper
- self.inherits.polymorphic_map.update(self.polymorphic_map)
- self.polymorphic_map = self.inherits.polymorphic_map
- for mapper in self.iterate_to_root():
- if mapper.polymorphic_on is not None:
- mapper._requires_row_aliasing = True
- self.batch = self.inherits.batch
- for mp in self.self_and_descendants:
- mp.base_mapper = self.inherits.base_mapper
- self.inherits._inheriting_mappers.append(self)
- self.passive_updates = self.inherits.passive_updates
- self._all_tables = self.inherits._all_tables
-
- for key, prop in mapper._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)
-
- def _set_polymorphic_on(self, polymorphic_on):
- self.polymorphic_on = polymorphic_on
- 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
- given class and entity name.
-
- Subsequent calls to ``class_mapper()`` for the ``class_`` / ``entity``
- name combination will return this mapper. Also decorate the
- `__init__` method on the mapped class to include optional
- auto-session attachment logic.
-
- """
-
- # we expect that declarative has applied the class manager
- # already and set up a registry. if this is None,
- # 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 "
- "invoked directly outside of a declarative registry."
- " Please use the sqlalchemy.orm.registry.map_imperatively() "
- "function for a classical mapping."
- )
-
- self.dispatch.instrument_class(self, self.class_)
-
- # this invokes the class_instrument event and sets up
- # the __init__ method. documented behavior is that this must
- # occur after the instrument_class event above.
- # yes two events with the same two words reversed and different APIs.
- # :(
-
- manager = instrumentation.register_class(
- self.class_,
- mapper=self,
- expired_attribute_loader=util.partial(
- loading.load_scalar_attributes, self
- ),
- # finalize flag means instrument the __init__ method
- # and call the class_instrument event
- finalize=True,
- )
-
- self.class_manager = manager
-
- assert manager.registry is not None
- self.registry = manager.registry
-
- # The remaining members can be added by any mapper,
- # e_name None or not.
- if manager.mapper is None:
- return
-
- event.listen(manager, "init", _event_on_init, raw=True)
-
- for key, method in util.iterate_attributes(self.class_):
- if key == "__init__" and hasattr(method, "_sa_original_init"):
- method = method._sa_original_init
- if hasattr(method, "__func__"):
- method = method.__func__
- if callable(method):
- if hasattr(method, "__sa_reconstructor__"):
- self._reconstructor = method
- event.listen(manager, "load", _event_on_load, raw=True)
- elif hasattr(method, "__sa_validators__"):
- validation_opts = method.__sa_validation_opts__
- for name in method.__sa_validators__:
- if name in self.validators:
- raise sa_exc.InvalidRequestError(
- "A validation function for mapped "
- "attribute %r on mapper %s already exists."
- % (name, self)
- )
- self.validators = self.validators.union(
- {name: (method, validation_opts)}
- )
-
- def _set_dispose_flags(self) -> None:
- self.configured = True
- self._ready_for_configure = True
- self._dispose_called = True
-
- self.__dict__.pop("_configure_failed", None)
-
- def _str_arg_to_mapped_col(self, argname: str, key: str) -> Column[Any]:
- try:
- prop = self._props[key]
- except KeyError as err:
- raise sa_exc.ArgumentError(
- f"Can't determine {argname} column '{key}' - "
- "no attribute is mapped to this name."
- ) from err
- try:
- expr = prop.expression
- except AttributeError as ae:
- raise sa_exc.ArgumentError(
- f"Can't determine {argname} column '{key}'; "
- "property does not refer to a single mapped Column"
- ) from ae
- if not isinstance(expr, Column):
- raise sa_exc.ArgumentError(
- f"Can't determine {argname} column '{key}'; "
- "property does not refer to a single "
- "mapped Column"
- )
- return expr
-
- def _configure_pks(self) -> None:
- self.tables = sql_util.find_tables(self.persist_selectable)
-
- self._all_tables.update(t for t in self.tables)
-
- self._pks_by_table = {}
- self._cols_by_table = {}
-
- all_cols = util.column_set(
- chain(*[col.proxy_set for col in self._columntoproperty])
- )
-
- pk_cols = util.column_set(c for c in all_cols if c.primary_key)
-
- # identify primary key columns which are also mapped by this mapper.
- for fc in set(self.tables).union([self.persist_selectable]):
- if fc.primary_key and pk_cols.issuperset(fc.primary_key):
- # ordering is important since it determines the ordering of
- # mapper.primary_key (and therefore query.get())
- self._pks_by_table[fc] = util.ordered_column_set( # type: ignore # noqa: E501
- fc.primary_key
- ).intersection(
- pk_cols
- )
- self._cols_by_table[fc] = util.ordered_column_set(fc.c).intersection( # type: ignore # noqa: E501
- all_cols
- )
-
- if self._primary_key_argument:
- coerced_pk_arg = [
- (
- self._str_arg_to_mapped_col("primary_key", c)
- if isinstance(c, str)
- else c
- )
- for c in (
- coercions.expect(
- roles.DDLConstraintColumnRole,
- coerce_pk,
- argname="primary_key",
- )
- for coerce_pk in self._primary_key_argument
- )
- ]
- else:
- coerced_pk_arg = None
-
- # if explicit PK argument sent, add those columns to the
- # primary key mappings
- if coerced_pk_arg:
- for k in coerced_pk_arg:
- if k.table not in self._pks_by_table:
- self._pks_by_table[k.table] = util.OrderedSet()
- self._pks_by_table[k.table].add(k)
-
- # otherwise, see that we got a full PK for the mapped table
- elif (
- self.persist_selectable not in self._pks_by_table
- or len(self._pks_by_table[self.persist_selectable]) == 0
- ):
- raise sa_exc.ArgumentError(
- "Mapper %s could not assemble any primary "
- "key columns for mapped table '%s'"
- % (self, self.persist_selectable.description)
- )
- elif self.local_table not in self._pks_by_table and isinstance(
- self.local_table, schema.Table
- ):
- util.warn(
- "Could not assemble any primary "
- "keys for locally mapped table '%s' - "
- "no rows will be persisted in this Table."
- % self.local_table.description
- )
-
- if (
- self.inherits
- and not self.concrete
- and not self._primary_key_argument
- ):
- # if inheriting, the "primary key" for this mapper is
- # that of the inheriting (unless concrete or explicit)
- self.primary_key = self.inherits.primary_key
- else:
- # determine primary key from argument or persist_selectable pks
- primary_key: Collection[ColumnElement[Any]]
-
- if coerced_pk_arg:
- primary_key = [
- cc if cc is not None else c
- for cc, c in (
- (self.persist_selectable.corresponding_column(c), c)
- for c in coerced_pk_arg
- )
- ]
- else:
- # if heuristically determined PKs, reduce to the minimal set
- # of columns by eliminating FK->PK pairs for a multi-table
- # expression. May over-reduce for some kinds of UNIONs
- # / CTEs; use explicit PK argument for these special cases
- primary_key = sql_util.reduce_columns(
- self._pks_by_table[self.persist_selectable],
- ignore_nonexistent_tables=True,
- )
-
- if len(primary_key) == 0:
- raise sa_exc.ArgumentError(
- "Mapper %s could not assemble any primary "
- "key columns for mapped table '%s'"
- % (self, self.persist_selectable.description)
- )
-
- self.primary_key = tuple(primary_key)
- self._log("Identified primary key columns: %s", primary_key)
-
- # determine cols that aren't expressed within our tables; mark these
- # as "read only" properties which are refreshed upon INSERT/UPDATE
- self._readonly_props = {
- self._columntoproperty[col]
- for col in self._columntoproperty
- if self._columntoproperty[col] not in self._identity_key_props
- and (
- not hasattr(col, "table")
- or col.table not in self._cols_by_table
- )
- }
-
- def _configure_properties(self) -> None:
- self.columns = self.c = sql_base.ColumnCollection() # type: ignore
-
- # object attribute names mapped to MapperProperty objects
- self._props = util.OrderedDict()
-
- # table columns mapped to MapperProperty
- self._columntoproperty = _ColumnMapping(self)
-
- 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_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.
-
- _map_as_property_now = True
- if isinstance(possible_col_prop, properties.ColumnProperty):
- 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,
- )
-
- 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
- # are only in the inheriting mapper, set them up as local props
- if self.inherits:
- 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
-
- 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)
-
- # 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.
-
- 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
- # Table / selectable
- key, prop = explicit_col_props_by_column[column]
- self._configure_property(key, prop, init=False)
- continue
-
- elif column in self._columntoproperty:
- continue
-
- column_key = (self.column_prefix or "") + column.key
- if self._should_exclude(
- column.key,
- column_key,
- local=self.local_table.c.contains_column(column),
- column=column,
- ):
- continue
-
- # adjust the "key" used for this column to that
- # of the inheriting mapper
- for mapper in self.iterate_to_root():
- if column in mapper._columntoproperty:
- column_key = mapper._columntoproperty[column].key
-
- self._configure_property(
- column_key,
- column,
- init=False,
- setparent=True,
- )
-
- def _configure_polymorphic_setter(self, init=False):
- """Configure an attribute on the mapper representing the
- 'polymorphic_on' column, if applicable, and not
- already generated by _configure_properties (which is typical).
-
- Also create a setter function which will assign this
- attribute to the value of the 'polymorphic_identity'
- upon instance construction, also if applicable. This
- routine will run when an instance is created.
-
- """
- setter = False
- polymorphic_key: Optional[str] = None
-
- if self.polymorphic_on is not None:
- setter = True
-
- if isinstance(self.polymorphic_on, str):
- # polymorphic_on specified as a string - link
- # it to mapped ColumnProperty
- try:
- self.polymorphic_on = self._props[self.polymorphic_on]
- except KeyError as err:
- raise sa_exc.ArgumentError(
- "Can't determine polymorphic_on "
- "value '%s' - no attribute is "
- "mapped to this name." % self.polymorphic_on
- ) from err
-
- if self.polymorphic_on in self._columntoproperty:
- # polymorphic_on is a column that is already mapped
- # to a ColumnProperty
- prop = self._columntoproperty[self.polymorphic_on]
- elif isinstance(self.polymorphic_on, MapperProperty):
- # polymorphic_on is directly a MapperProperty,
- # ensure it's a ColumnProperty
- if not isinstance(
- self.polymorphic_on, properties.ColumnProperty
- ):
- raise sa_exc.ArgumentError(
- "Only direct column-mapped "
- "property or SQL expression "
- "can be passed for polymorphic_on"
- )
- prop = self.polymorphic_on
- else:
- # polymorphic_on is a Column or SQL expression and
- # doesn't appear to be mapped. this means it can be 1.
- # only present in the with_polymorphic selectable or
- # 2. a totally standalone SQL expression which we'd
- # hope is compatible with this mapper's persist_selectable
- col = self.persist_selectable.corresponding_column(
- self.polymorphic_on
- )
- if col is None:
- # polymorphic_on doesn't derive from any
- # column/expression isn't present in the mapped
- # table. we will make a "hidden" ColumnProperty
- # for it. Just check that if it's directly a
- # schema.Column and we have with_polymorphic, it's
- # likely a user error if the schema.Column isn't
- # represented somehow in either persist_selectable or
- # with_polymorphic. Otherwise as of 0.7.4 we
- # just go with it and assume the user wants it
- # that way (i.e. a CASE statement)
- setter = False
- instrument = False
- col = self.polymorphic_on
- if isinstance(col, schema.Column) and (
- self.with_polymorphic is None
- or self.with_polymorphic[1] is None
- or self.with_polymorphic[1].corresponding_column(col)
- is None
- ):
- raise sa_exc.InvalidRequestError(
- "Could not map polymorphic_on column "
- "'%s' to the mapped table - polymorphic "
- "loads will not function properly"
- % col.description
- )
- else:
- # column/expression that polymorphic_on derives from
- # is present in our mapped table
- # and is probably mapped, but polymorphic_on itself
- # is not. This happens when
- # the polymorphic_on is only directly present in the
- # with_polymorphic selectable, as when use
- # polymorphic_union.
- # we'll make a separate ColumnProperty for it.
- instrument = True
- key = getattr(col, "key", None)
- if key:
- if self._should_exclude(key, key, False, col):
- raise sa_exc.InvalidRequestError(
- "Cannot exclude or override the "
- "discriminator column %r" % key
- )
- else:
- self.polymorphic_on = col = col.label("_sa_polymorphic_on")
- key = col.key
-
- prop = properties.ColumnProperty(col, _instrument=instrument)
- self._configure_property(key, prop, init=init, setparent=True)
-
- # the actual polymorphic_on should be the first public-facing
- # column in the property
- self.polymorphic_on = prop.columns[0]
- polymorphic_key = prop.key
- else:
- # no polymorphic_on was set.
- # check inheriting mappers for one.
- for mapper in self.iterate_to_root():
- # determine if polymorphic_on of the parent
- # should be propagated here. If the col
- # is present in our mapped table, or if our mapped
- # table is the same as the parent (i.e. single table
- # inheritance), we can use it
- if mapper.polymorphic_on is not None:
- if self.persist_selectable is mapper.persist_selectable:
- self.polymorphic_on = mapper.polymorphic_on
- else:
- self.polymorphic_on = (
- self.persist_selectable
- ).corresponding_column(mapper.polymorphic_on)
- # we can use the parent mapper's _set_polymorphic_identity
- # directly; it ensures the polymorphic_identity of the
- # instance's mapper is used so is portable to subclasses.
- if self.polymorphic_on is not None:
- self._set_polymorphic_identity = (
- mapper._set_polymorphic_identity
- )
- self._polymorphic_attr_key = (
- mapper._polymorphic_attr_key
- )
- self._validate_polymorphic_identity = (
- mapper._validate_polymorphic_identity
- )
- else:
- self._set_polymorphic_identity = None
- self._polymorphic_attr_key = None
- return
-
- if self.polymorphic_abstract and self.polymorphic_on is None:
- raise sa_exc.InvalidRequestError(
- "The Mapper.polymorphic_abstract parameter may only be used "
- "on a mapper hierarchy which includes the "
- "Mapper.polymorphic_on parameter at the base of the hierarchy."
- )
-
- if setter:
-
- def _set_polymorphic_identity(state):
- dict_ = state.dict
- # TODO: what happens if polymorphic_on column attribute name
- # does not match .key?
-
- polymorphic_identity = (
- state.manager.mapper.polymorphic_identity
- )
- if (
- polymorphic_identity is None
- and state.manager.mapper.polymorphic_abstract
- ):
- raise sa_exc.InvalidRequestError(
- f"Can't instantiate class for {state.manager.mapper}; "
- "mapper is marked polymorphic_abstract=True"
- )
-
- state.get_impl(polymorphic_key).set(
- state,
- dict_,
- polymorphic_identity,
- None,
- )
-
- self._polymorphic_attr_key = polymorphic_key
-
- def _validate_polymorphic_identity(mapper, state, dict_):
- if (
- polymorphic_key in dict_
- and dict_[polymorphic_key]
- not in mapper._acceptable_polymorphic_identities
- ):
- util.warn_limited(
- "Flushing object %s with "
- "incompatible polymorphic identity %r; the "
- "object may not refresh and/or load correctly",
- (state_str(state), dict_[polymorphic_key]),
- )
-
- self._set_polymorphic_identity = _set_polymorphic_identity
- self._validate_polymorphic_identity = (
- _validate_polymorphic_identity
- )
- else:
- self._polymorphic_attr_key = None
- self._set_polymorphic_identity = None
-
- _validate_polymorphic_identity = None
-
- @HasMemoized.memoized_attribute
- def _version_id_prop(self):
- if self.version_id_col is not None:
- return self._columntoproperty[self.version_id_col]
- else:
- return None
-
- @HasMemoized.memoized_attribute
- def _acceptable_polymorphic_identities(self):
- identities = set()
-
- stack = deque([self])
- while stack:
- item = stack.popleft()
- if item.persist_selectable is self.persist_selectable:
- identities.add(item.polymorphic_identity)
- stack.extend(item._inheriting_mappers)
-
- return identities
-
- @HasMemoized.memoized_attribute
- def _prop_set(self):
- return frozenset(self._props.values())
-
- @util.preload_module("sqlalchemy.orm.descriptor_props")
- def _adapt_inherited_property(self, key, prop, init):
- descriptor_props = util.preloaded.orm_descriptor_props
-
- if not self.concrete:
- self._configure_property(key, prop, init=False, setparent=False)
- elif key not in self._props:
- # determine if the class implements this attribute; if not,
- # or if it is implemented by the attribute that is handling the
- # given superclass-mapped property, then we need to report that we
- # can't use this at the instance level since we are a concrete
- # mapper and we don't map this. don't trip user-defined
- # descriptors that might have side effects when invoked.
- implementing_attribute = self.class_manager._get_class_attr_mro(
- key, prop
- )
- if implementing_attribute is prop or (
- isinstance(
- implementing_attribute, attributes.InstrumentedAttribute
- )
- and implementing_attribute._parententity is prop.parent
- ):
- self._configure_property(
- key,
- descriptor_props.ConcreteInheritedProperty(),
- init=init,
- setparent=True,
- )
-
- @util.preload_module("sqlalchemy.orm.descriptor_props")
- def _configure_property(
- self,
- key: str,
- prop_arg: Union[KeyedColumnElement[Any], MapperProperty[Any]],
- *,
- init: bool = True,
- setparent: bool = True,
- warn_for_existing: bool = False,
- ) -> MapperProperty[Any]:
- descriptor_props = util.preloaded.orm_descriptor_props
- self._log(
- "_configure_property(%s, %s)", key, prop_arg.__class__.__name__
- )
-
- if not isinstance(prop_arg, MapperProperty):
- prop: MapperProperty[Any] = self._property_from_column(
- key, prop_arg
- )
- else:
- prop = prop_arg
-
- if isinstance(prop, properties.ColumnProperty):
- col = self.persist_selectable.corresponding_column(prop.columns[0])
-
- # if the column is not present in the mapped table,
- # test if a column has been added after the fact to the
- # parent table (or their parent, etc.) [ticket:1570]
- if col is None and self.inherits:
- path = [self]
- for m in self.inherits.iterate_to_root():
- col = m.local_table.corresponding_column(prop.columns[0])
- if col is not None:
- for m2 in path:
- m2.persist_selectable._refresh_for_new_column(col)
- col = self.persist_selectable.corresponding_column(
- prop.columns[0]
- )
- break
- path.append(m)
-
- # subquery expression, column not present in the mapped
- # selectable.
- if col is None:
- col = prop.columns[0]
-
- # column is coming in after _readonly_props was
- # initialized; check for 'readonly'
- if hasattr(self, "_readonly_props") and (
- not hasattr(col, "table")
- or col.table not in self._cols_by_table
- ):
- self._readonly_props.add(prop)
-
- else:
- # if column is coming in after _cols_by_table was
- # initialized, ensure the col is in the right set
- if (
- hasattr(self, "_cols_by_table")
- and col.table in self._cols_by_table
- and col not in self._cols_by_table[col.table]
- ):
- self._cols_by_table[col.table].add(col)
-
- # if this properties.ColumnProperty represents the "polymorphic
- # discriminator" column, mark it. We'll need this when rendering
- # columns in SELECT statements.
- if not hasattr(prop, "_is_polymorphic_discriminator"):
- prop._is_polymorphic_discriminator = (
- col is self.polymorphic_on
- or prop.columns[0] is self.polymorphic_on
- )
-
- if isinstance(col, expression.Label):
- # new in 1.4, get column property against expressions
- # to be addressable in subqueries
- col.key = col._tq_key_label = key
-
- self.columns.add(col, key)
-
- for col in prop.columns:
- for proxy_col in col.proxy_set:
- self._columntoproperty[proxy_col] = prop
-
- if getattr(prop, "key", key) != key:
- util.warn(
- f"ORM mapped property {self.class_.__name__}.{prop.key} being "
- "assigned to attribute "
- f"{key!r} is already associated with "
- f"attribute {prop.key!r}. The attribute will be de-associated "
- f"from {prop.key!r}."
- )
-
- prop.key = key
-
- if setparent:
- prop.set_parent(self, init)
-
- if key in self._props and getattr(
- self._props[key], "_mapped_by_synonym", False
- ):
- syn = self._props[key]._mapped_by_synonym
- raise sa_exc.ArgumentError(
- "Can't call map_column=True for synonym %r=%r, "
- "a ColumnProperty already exists keyed to the name "
- "%r for column %r" % (syn, key, key, syn)
- )
-
- # replacement cases
-
- # case one: prop is replacing a prop that we have mapped. this is
- # independent of whatever might be in the actual class dictionary
- if (
- key in self._props
- and not isinstance(
- self._props[key], descriptor_props.ConcreteInheritedProperty
- )
- and not isinstance(prop, descriptor_props.SynonymProperty)
- ):
- if warn_for_existing:
- util.warn_deprecated(
- f"User-placed attribute {self.class_.__name__}.{key} on "
- f"{self} is replacing an existing ORM-mapped attribute. "
- "Behavior is not fully defined in this case. This "
- "use is deprecated and will raise an error in a future "
- "release",
- "2.0",
- )
- oldprop = self._props[key]
- self._path_registry.pop(oldprop, None)
-
- # case two: prop is replacing an attribute on the class of some kind.
- # we have to be more careful here since it's normal when using
- # Declarative that all the "declared attributes" on the class
- # get replaced.
- elif (
- warn_for_existing
- and self.class_.__dict__.get(key, None) is not None
- and not isinstance(prop, descriptor_props.SynonymProperty)
- and not isinstance(
- self._props.get(key, None),
- descriptor_props.ConcreteInheritedProperty,
- )
- ):
- util.warn_deprecated(
- f"User-placed attribute {self.class_.__name__}.{key} on "
- f"{self} is replacing an existing class-bound "
- "attribute of the same name. "
- "Behavior is not fully defined in this case. This "
- "use is deprecated and will raise an error in a future "
- "release",
- "2.0",
- )
-
- self._props[key] = prop
-
- if not self.non_primary:
- prop.instrument_class(self)
-
- for mapper in self._inheriting_mappers:
- mapper._adapt_inherited_property(key, prop, init)
-
- if init:
- prop.init()
- prop.post_instrument_class(self)
-
- if self.configured:
- self._expire_memoizations()
-
- 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,
- column: KeyedColumnElement[Any],
- ) -> ColumnProperty[Any]:
- """generate/update a :class:`.ColumnProperty` given a
- :class:`_schema.Column` or other SQL expression object."""
-
- descriptor_props = util.preloaded.orm_descriptor_props
-
- prop = self._props.get(key)
-
- if isinstance(prop, properties.ColumnProperty):
- return self._reconcile_prop_with_incoming_columns(
- key,
- prop,
- single_column=column,
- warn_only=prop.parent is not self,
- )
- elif prop is None or isinstance(
- prop, descriptor_props.ConcreteInheritedProperty
- ):
- return self._make_prop_from_column(key, column)
- else:
- raise sa_exc.ArgumentError(
- "WARNING: when configuring property '%s' on %s, "
- "column '%s' conflicts with property '%r'. "
- "To resolve this, map the column to the class under a "
- "different name in the 'properties' dictionary. Or, "
- "to remove all awareness of the column entirely "
- "(including its availability as a foreign key), "
- "use the 'include_properties' or 'exclude_properties' "
- "mapper arguments to control specifically which table "
- "columns get mapped." % (key, self, column.key, prop)
- )
-
- @util.langhelpers.tag_method_for_warnings(
- "This warning originated from the `configure_mappers()` process, "
- "which was invoked automatically in response to a user-initiated "
- "operation.",
- sa_exc.SAWarning,
- )
- def _check_configure(self) -> None:
- if self.registry._new_mappers:
- _configure_registries({self.registry}, cascade=True)
-
- def _post_configure_properties(self) -> None:
- """Call the ``init()`` method on all ``MapperProperties``
- attached to this mapper.
-
- This is a deferred configuration step which is intended
- to execute once all mappers have been constructed.
-
- """
-
- self._log("_post_configure_properties() started")
- l = [(key, prop) for key, prop in self._props.items()]
- for key, prop in l:
- self._log("initialize prop %s", key)
-
- if prop.parent is self and not prop._configure_started:
- prop.init()
-
- if prop._configure_finished:
- prop.post_instrument_class(self)
-
- self._log("_post_configure_properties() complete")
- self.configured = True
-
- def add_properties(self, dict_of_properties):
- """Add the given dictionary of properties to this mapper,
- using `add_property`.
-
- """
- for key, value in dict_of_properties.items():
- self.add_property(key, value)
-
- def add_property(
- self, key: str, prop: Union[Column[Any], MapperProperty[Any]]
- ) -> None:
- """Add an individual MapperProperty to this mapper.
-
- If the mapper has not been configured yet, just adds the
- property to the initial properties dictionary sent to the
- constructor. If this Mapper has already been configured, then
- the given MapperProperty is configured immediately.
-
- """
- prop = self._configure_property(
- key, prop, init=self.configured, warn_for_existing=True
- )
- assert isinstance(prop, MapperProperty)
- self._init_properties[key] = prop
-
- def _expire_memoizations(self) -> None:
- for mapper in self.iterate_to_root():
- mapper._reset_memoizations()
-
- @property
- def _log_desc(self) -> str:
- return (
- "("
- + self.class_.__name__
- + "|"
- + (
- self.local_table is not None
- and self.local_table.description
- or str(self.local_table)
- )
- + (self.non_primary and "|non-primary" or "")
- + ")"
- )
-
- def _log(self, msg: str, *args: Any) -> None:
- self.logger.info("%s " + msg, *((self._log_desc,) + args))
-
- def _log_debug(self, msg: str, *args: Any) -> None:
- self.logger.debug("%s " + msg, *((self._log_desc,) + args))
-
- def __repr__(self) -> str:
- return "<Mapper at 0x%x; %s>" % (id(self), self.class_.__name__)
-
- def __str__(self) -> str:
- return "Mapper[%s%s(%s)]" % (
- self.class_.__name__,
- self.non_primary and " (non-primary)" or "",
- (
- self.local_table.description
- if self.local_table is not None
- else self.persist_selectable.description
- ),
- )
-
- def _is_orphan(self, state: InstanceState[_O]) -> bool:
- orphan_possible = False
- for mapper in self.iterate_to_root():
- for key, cls in mapper._delete_orphans:
- orphan_possible = True
-
- has_parent = attributes.manager_of_class(cls).has_parent(
- state, key, optimistic=state.has_identity
- )
-
- if self.legacy_is_orphan and has_parent:
- return False
- elif not self.legacy_is_orphan and not has_parent:
- return True
-
- if self.legacy_is_orphan:
- return orphan_possible
- else:
- return False
-
- def has_property(self, key: str) -> bool:
- return key in self._props
-
- def get_property(
- self, key: str, _configure_mappers: bool = False
- ) -> MapperProperty[Any]:
- """return a MapperProperty associated with the given key."""
-
- if _configure_mappers:
- self._check_configure()
-
- try:
- return self._props[key]
- except KeyError as err:
- raise sa_exc.InvalidRequestError(
- 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(
- self, column: ColumnElement[_T]
- ) -> MapperProperty[_T]:
- """Given a :class:`_schema.Column` object, return the
- :class:`.MapperProperty` which maps this column."""
-
- return self._columntoproperty[column]
-
- @property
- def iterate_properties(self):
- """return an iterator of all MapperProperty objects."""
-
- return iter(self._props.values())
-
- def _mappers_from_spec(
- self, spec: Any, selectable: Optional[FromClause]
- ) -> Sequence[Mapper[Any]]:
- """given a with_polymorphic() argument, return the set of mappers it
- represents.
-
- Trims the list of mappers to just those represented within the given
- selectable, if present. This helps some more legacy-ish mappings.
-
- """
- if spec == "*":
- mappers = list(self.self_and_descendants)
- elif spec:
- mapper_set = set()
- for m in util.to_list(spec):
- m = _class_to_mapper(m)
- if not m.isa(self):
- raise sa_exc.InvalidRequestError(
- "%r does not inherit from %r" % (m, self)
- )
-
- if selectable is None:
- mapper_set.update(m.iterate_to_root())
- else:
- mapper_set.add(m)
- mappers = [m for m in self.self_and_descendants if m in mapper_set]
- else:
- mappers = []
-
- if selectable is not None:
- tables = set(
- sql_util.find_tables(selectable, include_aliases=True)
- )
- mappers = [m for m in mappers if m.local_table in tables]
- return mappers
-
- def _selectable_from_mappers(
- self, mappers: Iterable[Mapper[Any]], innerjoin: bool
- ) -> FromClause:
- """given a list of mappers (assumed to be within this mapper's
- inheritance hierarchy), construct an outerjoin amongst those mapper's
- mapped tables.
-
- """
- from_obj = self.persist_selectable
- for m in mappers:
- if m is self:
- continue
- if m.concrete:
- raise sa_exc.InvalidRequestError(
- "'with_polymorphic()' requires 'selectable' argument "
- "when concrete-inheriting mappers are used."
- )
- elif not m.single:
- if innerjoin:
- from_obj = from_obj.join(
- m.local_table, m.inherit_condition
- )
- else:
- from_obj = from_obj.outerjoin(
- m.local_table, m.inherit_condition
- )
-
- return from_obj
-
- @HasMemoized.memoized_attribute
- def _version_id_has_server_side_value(self) -> bool:
- vid_col = self.version_id_col
-
- if vid_col is None:
- return False
-
- elif not isinstance(vid_col, Column):
- return True
- else:
- return vid_col.server_default is not None or (
- vid_col.default is not None
- and (
- not vid_col.default.is_scalar
- and not vid_col.default.is_callable
- )
- )
-
- @HasMemoized.memoized_attribute
- def _single_table_criterion(self):
- if self.single and self.inherits and self.polymorphic_on is not None:
- return self.polymorphic_on._annotate(
- {"parententity": self, "parentmapper": self}
- ).in_(
- [
- m.polymorphic_identity
- for m in self.self_and_descendants
- if not m.polymorphic_abstract
- ]
- )
- else:
- return None
-
- @HasMemoized.memoized_attribute
- def _has_aliased_polymorphic_fromclause(self):
- """return True if with_polymorphic[1] is an aliased fromclause,
- like a subquery.
-
- As of #8168, polymorphic adaption with ORMAdapter is used only
- if this is present.
-
- """
- return self.with_polymorphic and isinstance(
- self.with_polymorphic[1],
- expression.AliasedReturnsRows,
- )
-
- @HasMemoized.memoized_attribute
- def _should_select_with_poly_adapter(self):
- """determine if _MapperEntity or _ORMColumnEntity will need to use
- polymorphic adaption when setting up a SELECT as well as fetching
- rows for mapped classes and subclasses against this Mapper.
-
- moved here from context.py for #8456 to generalize the ruleset
- for this condition.
-
- """
-
- # this has been simplified as of #8456.
- # rule is: if we have a with_polymorphic or a concrete-style
- # polymorphic selectable, *or* if the base mapper has either of those,
- # we turn on the adaption thing. if not, we do *no* adaption.
- #
- # (UPDATE for #8168: the above comment was not accurate, as we were
- # still saying "do polymorphic" if we were using an auto-generated
- # flattened JOIN for with_polymorphic.)
- #
- # this splits the behavior among the "regular" joined inheritance
- # and single inheritance mappers, vs. the "weird / difficult"
- # concrete and joined inh mappings that use a with_polymorphic of
- # some kind or polymorphic_union.
- #
- # note we have some tests in test_polymorphic_rel that query against
- # a subclass, then refer to the superclass that has a with_polymorphic
- # on it (such as test_join_from_polymorphic_explicit_aliased_three).
- # these tests actually adapt the polymorphic selectable (like, the
- # UNION or the SELECT subquery with JOIN in it) to be just the simple
- # subclass table. Hence even if we are a "plain" inheriting mapper
- # but our base has a wpoly on it, we turn on adaption. This is a
- # legacy case we should probably disable.
- #
- #
- # UPDATE: simplified way more as of #8168. polymorphic adaption
- # is turned off even if with_polymorphic is set, as long as there
- # is no user-defined aliased selectable / subquery configured.
- # this scales back the use of polymorphic adaption in practice
- # to basically no cases except for concrete inheritance with a
- # polymorphic base class.
- #
- return (
- self._has_aliased_polymorphic_fromclause
- or self._requires_row_aliasing
- or (self.base_mapper._has_aliased_polymorphic_fromclause)
- or self.base_mapper._requires_row_aliasing
- )
-
- @HasMemoized.memoized_attribute
- def _with_polymorphic_mappers(self) -> Sequence[Mapper[Any]]:
- self._check_configure()
-
- if not self.with_polymorphic:
- return []
- return self._mappers_from_spec(*self.with_polymorphic)
-
- @HasMemoized.memoized_attribute
- def _post_inspect(self):
- """This hook is invoked by attribute inspection.
-
- E.g. when Query calls:
-
- coercions.expect(roles.ColumnsClauseRole, ent, keep_inspect=True)
-
- This allows the inspection process run a configure mappers hook.
-
- """
- self._check_configure()
-
- @HasMemoized_ro_memoized_attribute
- def _with_polymorphic_selectable(self) -> FromClause:
- if not self.with_polymorphic:
- return self.persist_selectable
-
- spec, selectable = self.with_polymorphic
- if selectable is not None:
- return selectable
- else:
- return self._selectable_from_mappers(
- self._mappers_from_spec(spec, selectable), False
- )
-
- with_polymorphic_mappers = _with_polymorphic_mappers
- """The list of :class:`_orm.Mapper` objects included in the
- default "polymorphic" query.
-
- """
-
- @HasMemoized_ro_memoized_attribute
- def _insert_cols_evaluating_none(self):
- return {
- table: frozenset(
- col for col in columns if col.type.should_evaluate_none
- )
- for table, columns in self._cols_by_table.items()
- }
-
- @HasMemoized.memoized_attribute
- def _insert_cols_as_none(self):
- return {
- table: frozenset(
- col.key
- for col in columns
- if not col.primary_key
- and not col.server_default
- and not col.default
- and not col.type.should_evaluate_none
- )
- for table, columns in self._cols_by_table.items()
- }
-
- @HasMemoized.memoized_attribute
- def _propkey_to_col(self):
- return {
- table: {self._columntoproperty[col].key: col for col in columns}
- for table, columns in self._cols_by_table.items()
- }
-
- @HasMemoized.memoized_attribute
- def _pk_keys_by_table(self):
- return {
- table: frozenset([col.key for col in pks])
- for table, pks in self._pks_by_table.items()
- }
-
- @HasMemoized.memoized_attribute
- def _pk_attr_keys_by_table(self):
- return {
- table: frozenset([self._columntoproperty[col].key for col in pks])
- for table, pks in self._pks_by_table.items()
- }
-
- @HasMemoized.memoized_attribute
- def _server_default_cols(
- self,
- ) -> Mapping[FromClause, FrozenSet[Column[Any]]]:
- return {
- table: frozenset(
- [
- col
- for col in cast("Iterable[Column[Any]]", columns)
- if col.server_default is not None
- or (
- col.default is not None
- and col.default.is_clause_element
- )
- ]
- )
- for table, columns in self._cols_by_table.items()
- }
-
- @HasMemoized.memoized_attribute
- def _server_onupdate_default_cols(
- self,
- ) -> Mapping[FromClause, FrozenSet[Column[Any]]]:
- return {
- table: frozenset(
- [
- col
- for col in cast("Iterable[Column[Any]]", columns)
- if col.server_onupdate is not None
- or (
- col.onupdate is not None
- and col.onupdate.is_clause_element
- )
- ]
- )
- for table, columns in self._cols_by_table.items()
- }
-
- @HasMemoized.memoized_attribute
- def _server_default_col_keys(self) -> Mapping[FromClause, FrozenSet[str]]:
- return {
- table: frozenset(col.key for col in cols if col.key is not None)
- for table, cols in self._server_default_cols.items()
- }
-
- @HasMemoized.memoized_attribute
- def _server_onupdate_default_col_keys(
- self,
- ) -> Mapping[FromClause, FrozenSet[str]]:
- return {
- table: frozenset(col.key for col in cols if col.key is not None)
- for table, cols in self._server_onupdate_default_cols.items()
- }
-
- @HasMemoized.memoized_attribute
- def _server_default_plus_onupdate_propkeys(self) -> Set[str]:
- result: Set[str] = set()
-
- col_to_property = self._columntoproperty
- for table, columns in self._server_default_cols.items():
- result.update(
- col_to_property[col].key
- for col in columns.intersection(col_to_property)
- )
- for table, columns in self._server_onupdate_default_cols.items():
- result.update(
- col_to_property[col].key
- for col in columns.intersection(col_to_property)
- )
- return result
-
- @HasMemoized.memoized_instancemethod
- def __clause_element__(self):
- annotations: Dict[str, Any] = {
- "entity_namespace": self,
- "parententity": self,
- "parentmapper": self,
- }
- if self.persist_selectable is not self.local_table:
- # joined table inheritance, with polymorphic selectable,
- # etc.
- annotations["dml_table"] = self.local_table._annotate(
- {
- "entity_namespace": self,
- "parententity": self,
- "parentmapper": self,
- }
- )._set_propagate_attrs(
- {"compile_state_plugin": "orm", "plugin_subject": self}
- )
-
- return self.selectable._annotate(annotations)._set_propagate_attrs(
- {"compile_state_plugin": "orm", "plugin_subject": self}
- )
-
- @util.memoized_property
- def select_identity_token(self):
- return (
- expression.null()
- ._annotate(
- {
- "entity_namespace": self,
- "parententity": self,
- "parentmapper": self,
- "identity_token": True,
- }
- )
- ._set_propagate_attrs(
- {"compile_state_plugin": "orm", "plugin_subject": self}
- )
- )
-
- @property
- def selectable(self) -> FromClause:
- """The :class:`_schema.FromClause` construct this
- :class:`_orm.Mapper` selects from by default.
-
- Normally, this is equivalent to :attr:`.persist_selectable`, unless
- the ``with_polymorphic`` feature is in use, in which case the
- full "polymorphic" selectable is returned.
-
- """
- return self._with_polymorphic_selectable
-
- def _with_polymorphic_args(
- self,
- spec: Any = None,
- selectable: Union[Literal[False, None], FromClause] = False,
- innerjoin: bool = False,
- ) -> Tuple[Sequence[Mapper[Any]], FromClause]:
- if selectable not in (None, False):
- selectable = coercions.expect(
- roles.StrictFromClauseRole, selectable, allow_select=True
- )
-
- if self.with_polymorphic:
- if not spec:
- spec = self.with_polymorphic[0]
- if selectable is False:
- selectable = self.with_polymorphic[1]
- elif selectable is False:
- selectable = None
- mappers = self._mappers_from_spec(spec, selectable)
- if selectable is not None:
- return mappers, selectable
- else:
- return mappers, self._selectable_from_mappers(mappers, innerjoin)
-
- @HasMemoized.memoized_attribute
- def _polymorphic_properties(self):
- return list(
- self._iterate_polymorphic_properties(
- self._with_polymorphic_mappers
- )
- )
-
- @property
- def _all_column_expressions(self):
- poly_properties = self._polymorphic_properties
- adapter = self._polymorphic_adapter
-
- return [
- adapter.columns[c] if adapter else c
- for prop in poly_properties
- if isinstance(prop, properties.ColumnProperty)
- and prop._renders_in_subqueries
- for c in prop.columns
- ]
-
- def _columns_plus_keys(self, polymorphic_mappers=()):
- if polymorphic_mappers:
- poly_properties = self._iterate_polymorphic_properties(
- polymorphic_mappers
- )
- else:
- poly_properties = self._polymorphic_properties
-
- return [
- (prop.key, prop.columns[0])
- for prop in poly_properties
- if isinstance(prop, properties.ColumnProperty)
- ]
-
- @HasMemoized.memoized_attribute
- def _polymorphic_adapter(self) -> Optional[orm_util.ORMAdapter]:
- if self._has_aliased_polymorphic_fromclause:
- return orm_util.ORMAdapter(
- orm_util._TraceAdaptRole.MAPPER_POLYMORPHIC_ADAPTER,
- self,
- selectable=self.selectable,
- equivalents=self._equivalent_columns,
- limit_on_entity=False,
- )
- else:
- return None
-
- def _iterate_polymorphic_properties(self, mappers=None):
- """Return an iterator of MapperProperty objects which will render into
- a SELECT."""
- if mappers is None:
- mappers = self._with_polymorphic_mappers
-
- if not mappers:
- for c in self.iterate_properties:
- yield c
- else:
- # in the polymorphic case, filter out discriminator columns
- # from other mappers, as these are sometimes dependent on that
- # mapper's polymorphic selectable (which we don't want rendered)
- for c in util.unique_list(
- chain(
- *[
- list(mapper.iterate_properties)
- for mapper in [self] + mappers
- ]
- )
- ):
- if getattr(c, "_is_polymorphic_discriminator", False) and (
- self.polymorphic_on is None
- or c.columns[0] is not self.polymorphic_on
- ):
- continue
- yield c
-
- @HasMemoized.memoized_attribute
- def attrs(self) -> util.ReadOnlyProperties[MapperProperty[Any]]:
- """A namespace of all :class:`.MapperProperty` objects
- associated this mapper.
-
- This is an object that provides each property based on
- its key name. For instance, the mapper for a
- ``User`` class which has ``User.name`` attribute would
- provide ``mapper.attrs.name``, which would be the
- :class:`.ColumnProperty` representing the ``name``
- column. The namespace object can also be iterated,
- which would yield each :class:`.MapperProperty`.
-
- :class:`_orm.Mapper` has several pre-filtered views
- of this attribute which limit the types of properties
- returned, including :attr:`.synonyms`, :attr:`.column_attrs`,
- :attr:`.relationships`, and :attr:`.composites`.
-
- .. warning::
-
- The :attr:`_orm.Mapper.attrs` accessor namespace is an
- instance of :class:`.OrderedProperties`. This is
- a dictionary-like object which includes a small number of
- named methods such as :meth:`.OrderedProperties.items`
- and :meth:`.OrderedProperties.values`. When
- accessing attributes dynamically, favor using the dict-access
- scheme, e.g. ``mapper.attrs[somename]`` over
- ``getattr(mapper.attrs, somename)`` to avoid name collisions.
-
- .. seealso::
-
- :attr:`_orm.Mapper.all_orm_descriptors`
-
- """
-
- self._check_configure()
- return util.ReadOnlyProperties(self._props)
-
- @HasMemoized.memoized_attribute
- def all_orm_descriptors(self) -> util.ReadOnlyProperties[InspectionAttr]:
- """A namespace of all :class:`.InspectionAttr` attributes associated
- with the mapped class.
-
- These attributes are in all cases Python :term:`descriptors`
- associated with the mapped class or its superclasses.
-
- This namespace includes attributes that are mapped to the class
- as well as attributes declared by extension modules.
- It includes any Python descriptor type that inherits from
- :class:`.InspectionAttr`. This includes
- :class:`.QueryableAttribute`, as well as extension types such as
- :class:`.hybrid_property`, :class:`.hybrid_method` and
- :class:`.AssociationProxy`.
-
- To distinguish between mapped attributes and extension attributes,
- the attribute :attr:`.InspectionAttr.extension_type` will refer
- to a constant that distinguishes between different extension types.
-
- The sorting of the attributes is based on the following rules:
-
- 1. Iterate through the class and its superclasses in order from
- subclass to superclass (i.e. iterate through ``cls.__mro__``)
-
- 2. For each class, yield the attributes in the order in which they
- appear in ``__dict__``, with the exception of those in step
- 3 below. In Python 3.6 and above this ordering will be the
- same as that of the class' construction, with the exception
- of attributes that were added after the fact by the application
- or the mapper.
-
- 3. If a certain attribute key is also in the superclass ``__dict__``,
- then it's included in the iteration for that class, and not the
- class in which it first appeared.
-
- The above process produces an ordering that is deterministic in terms
- of the order in which attributes were assigned to the class.
-
- .. versionchanged:: 1.3.19 ensured deterministic ordering for
- :meth:`_orm.Mapper.all_orm_descriptors`.
-
- When dealing with a :class:`.QueryableAttribute`, the
- :attr:`.QueryableAttribute.property` attribute refers to the
- :class:`.MapperProperty` property, which is what you get when
- referring to the collection of mapped properties via
- :attr:`_orm.Mapper.attrs`.
-
- .. warning::
-
- The :attr:`_orm.Mapper.all_orm_descriptors`
- accessor namespace is an
- instance of :class:`.OrderedProperties`. This is
- a dictionary-like object which includes a small number of
- named methods such as :meth:`.OrderedProperties.items`
- and :meth:`.OrderedProperties.values`. When
- accessing attributes dynamically, favor using the dict-access
- scheme, e.g. ``mapper.all_orm_descriptors[somename]`` over
- ``getattr(mapper.all_orm_descriptors, somename)`` to avoid name
- collisions.
-
- .. seealso::
-
- :attr:`_orm.Mapper.attrs`
-
- """
- return util.ReadOnlyProperties(
- dict(self.class_manager._all_sqla_attributes())
- )
-
- @HasMemoized.memoized_attribute
- @util.preload_module("sqlalchemy.orm.descriptor_props")
- def _pk_synonyms(self) -> Dict[str, str]:
- """return a dictionary of {syn_attribute_name: pk_attr_name} for
- all synonyms that refer to primary key columns
-
- """
- descriptor_props = util.preloaded.orm_descriptor_props
-
- pk_keys = {prop.key for prop in self._identity_key_props}
-
- return {
- syn.key: syn.name
- for k, syn in self._props.items()
- if isinstance(syn, descriptor_props.SynonymProperty)
- and syn.name in pk_keys
- }
-
- @HasMemoized.memoized_attribute
- @util.preload_module("sqlalchemy.orm.descriptor_props")
- def synonyms(self) -> util.ReadOnlyProperties[SynonymProperty[Any]]:
- """Return a namespace of all :class:`.Synonym`
- properties maintained by this :class:`_orm.Mapper`.
-
- .. seealso::
-
- :attr:`_orm.Mapper.attrs` - namespace of all
- :class:`.MapperProperty`
- objects.
-
- """
- descriptor_props = util.preloaded.orm_descriptor_props
-
- return self._filter_properties(descriptor_props.SynonymProperty)
-
- @property
- def entity_namespace(self):
- return self.class_
-
- @HasMemoized.memoized_attribute
- def column_attrs(self) -> util.ReadOnlyProperties[ColumnProperty[Any]]:
- """Return a namespace of all :class:`.ColumnProperty`
- properties maintained by this :class:`_orm.Mapper`.
-
- .. seealso::
-
- :attr:`_orm.Mapper.attrs` - namespace of all
- :class:`.MapperProperty`
- objects.
-
- """
- return self._filter_properties(properties.ColumnProperty)
-
- @HasMemoized.memoized_attribute
- @util.preload_module("sqlalchemy.orm.relationships")
- def relationships(
- self,
- ) -> util.ReadOnlyProperties[RelationshipProperty[Any]]:
- """A namespace of all :class:`.Relationship` properties
- maintained by this :class:`_orm.Mapper`.
-
- .. warning::
-
- the :attr:`_orm.Mapper.relationships` accessor namespace is an
- instance of :class:`.OrderedProperties`. This is
- a dictionary-like object which includes a small number of
- named methods such as :meth:`.OrderedProperties.items`
- and :meth:`.OrderedProperties.values`. When
- accessing attributes dynamically, favor using the dict-access
- scheme, e.g. ``mapper.relationships[somename]`` over
- ``getattr(mapper.relationships, somename)`` to avoid name
- collisions.
-
- .. seealso::
-
- :attr:`_orm.Mapper.attrs` - namespace of all
- :class:`.MapperProperty`
- objects.
-
- """
- return self._filter_properties(
- util.preloaded.orm_relationships.RelationshipProperty
- )
-
- @HasMemoized.memoized_attribute
- @util.preload_module("sqlalchemy.orm.descriptor_props")
- def composites(self) -> util.ReadOnlyProperties[CompositeProperty[Any]]:
- """Return a namespace of all :class:`.Composite`
- properties maintained by this :class:`_orm.Mapper`.
-
- .. seealso::
-
- :attr:`_orm.Mapper.attrs` - namespace of all
- :class:`.MapperProperty`
- objects.
-
- """
- return self._filter_properties(
- util.preloaded.orm_descriptor_props.CompositeProperty
- )
-
- def _filter_properties(
- self, type_: Type[_MP]
- ) -> util.ReadOnlyProperties[_MP]:
- self._check_configure()
- return util.ReadOnlyProperties(
- util.OrderedDict(
- (k, v) for k, v in self._props.items() if isinstance(v, type_)
- )
- )
-
- @HasMemoized.memoized_attribute
- def _get_clause(self):
- """create a "get clause" based on the primary key. this is used
- by query.get() and many-to-one lazyloads to load this item
- by primary key.
-
- """
- params = [
- (
- primary_key,
- sql.bindparam("pk_%d" % idx, type_=primary_key.type),
- )
- for idx, primary_key in enumerate(self.primary_key, 1)
- ]
- return (
- sql.and_(*[k == v for (k, v) in params]),
- util.column_dict(params),
- )
-
- @HasMemoized.memoized_attribute
- def _equivalent_columns(self) -> _EquivalentColumnMap:
- """Create a map of all equivalent columns, based on
- the determination of column pairs that are equated to
- one another based on inherit condition. This is designed
- to work with the queries that util.polymorphic_union
- comes up with, which often don't include the columns from
- the base table directly (including the subclass table columns
- only).
-
- The resulting structure is a dictionary of columns mapped
- to lists of equivalent columns, e.g.::
-
- {
- tablea.col1:
- {tableb.col1, tablec.col1},
- tablea.col2:
- {tabled.col2}
- }
-
- """
- result: _EquivalentColumnMap = {}
-
- def visit_binary(binary):
- if binary.operator == operators.eq:
- if binary.left in result:
- result[binary.left].add(binary.right)
- else:
- result[binary.left] = {binary.right}
- if binary.right in result:
- result[binary.right].add(binary.left)
- else:
- result[binary.right] = {binary.left}
-
- for mapper in self.base_mapper.self_and_descendants:
- if mapper.inherit_condition is not None:
- visitors.traverse(
- mapper.inherit_condition, {}, {"binary": visit_binary}
- )
-
- return result
-
- def _is_userland_descriptor(self, assigned_name: str, obj: Any) -> bool:
- if isinstance(
- obj,
- (
- _MappedAttribute,
- instrumentation.ClassManager,
- expression.ColumnElement,
- ),
- ):
- return False
- else:
- return assigned_name not in self._dataclass_fields
-
- @HasMemoized.memoized_attribute
- def _dataclass_fields(self):
- return [f.name for f in util.dataclass_fields(self.class_)]
-
- def _should_exclude(self, name, assigned_name, local, column):
- """determine whether a particular property should be implicitly
- present on the class.
-
- This occurs when properties are propagated from an inherited class, or
- are applied from the columns present in the mapped table.
-
- """
-
- if column is not None and sql_base._never_select_column(column):
- return True
-
- # check for class-bound attributes and/or descriptors,
- # either local or from an inherited class
- # ignore dataclass field default values
- if local:
- if self.class_.__dict__.get(
- assigned_name, None
- ) is not None and self._is_userland_descriptor(
- assigned_name, self.class_.__dict__[assigned_name]
- ):
- return True
- else:
- attr = self.class_manager._get_class_attr_mro(assigned_name, None)
- if attr is not None and self._is_userland_descriptor(
- assigned_name, attr
- ):
- return True
-
- if (
- self.include_properties is not None
- and name not in self.include_properties
- and (column is None or column not in self.include_properties)
- ):
- self._log("not including property %s" % (name))
- return True
-
- if self.exclude_properties is not None and (
- name in self.exclude_properties
- or (column is not None and column in self.exclude_properties)
- ):
- self._log("excluding property %s" % (name))
- return True
-
- return False
-
- def common_parent(self, other: Mapper[Any]) -> bool:
- """Return true if the given mapper shares a
- common inherited parent as this mapper."""
-
- return self.base_mapper is other.base_mapper
-
- def is_sibling(self, other: Mapper[Any]) -> bool:
- """return true if the other mapper is an inheriting sibling to this
- one. common parent but different branch
-
- """
- return (
- self.base_mapper is other.base_mapper
- and not self.isa(other)
- and not other.isa(self)
- )
-
- def _canload(
- self, state: InstanceState[Any], allow_subtypes: bool
- ) -> bool:
- s = self.primary_mapper()
- if self.polymorphic_on is not None or allow_subtypes:
- return _state_mapper(state).isa(s)
- else:
- return _state_mapper(state) is s
-
- def isa(self, other: Mapper[Any]) -> bool:
- """Return True if the this mapper inherits from the given mapper."""
-
- m: Optional[Mapper[Any]] = self
- while m and m is not other:
- m = m.inherits
- return bool(m)
-
- def iterate_to_root(self) -> Iterator[Mapper[Any]]:
- m: Optional[Mapper[Any]] = self
- while m:
- yield m
- m = m.inherits
-
- @HasMemoized.memoized_attribute
- def self_and_descendants(self) -> Sequence[Mapper[Any]]:
- """The collection including this mapper and all descendant mappers.
-
- This includes not just the immediately inheriting mappers but
- all their inheriting mappers as well.
-
- """
- descendants = []
- stack = deque([self])
- while stack:
- item = stack.popleft()
- descendants.append(item)
- stack.extend(item._inheriting_mappers)
- return util.WeakSequence(descendants)
-
- def polymorphic_iterator(self) -> Iterator[Mapper[Any]]:
- """Iterate through the collection including this mapper and
- all descendant mappers.
-
- This includes not just the immediately inheriting mappers but
- all their inheriting mappers as well.
-
- To iterate through an entire hierarchy, use
- ``mapper.base_mapper.polymorphic_iterator()``.
-
- """
- return iter(self.self_and_descendants)
-
- def primary_mapper(self) -> Mapper[Any]:
- """Return the primary mapper corresponding to this mapper's class key
- (class)."""
-
- return self.class_manager.mapper
-
- @property
- def primary_base_mapper(self) -> Mapper[Any]:
- return self.class_manager.mapper.base_mapper
-
- def _result_has_identity_key(self, result, adapter=None):
- pk_cols: Sequence[ColumnClause[Any]] = self.primary_key
- if adapter:
- pk_cols = [adapter.columns[c] for c in pk_cols]
- rk = result.keys()
- for col in pk_cols:
- if col not in rk:
- return False
- else:
- return True
-
- def identity_key_from_row(
- self,
- row: Optional[Union[Row[Any], RowMapping]],
- identity_token: Optional[Any] = None,
- adapter: Optional[ORMAdapter] = None,
- ) -> _IdentityKeyType[_O]:
- """Return an identity-map key for use in storing/retrieving an
- item from the identity map.
-
- :param row: A :class:`.Row` or :class:`.RowMapping` produced from a
- result set that selected from the ORM mapped primary key columns.
-
- .. versionchanged:: 2.0
- :class:`.Row` or :class:`.RowMapping` are accepted
- for the "row" argument
-
- """
- pk_cols: Sequence[ColumnClause[Any]] = self.primary_key
- if adapter:
- pk_cols = [adapter.columns[c] for c in pk_cols]
-
- if hasattr(row, "_mapping"):
- mapping = row._mapping # type: ignore
- else:
- mapping = cast("Mapping[Any, Any]", row)
-
- return (
- self._identity_class,
- tuple(mapping[column] for column in pk_cols), # type: ignore
- identity_token,
- )
-
- def identity_key_from_primary_key(
- self,
- primary_key: Tuple[Any, ...],
- identity_token: Optional[Any] = None,
- ) -> _IdentityKeyType[_O]:
- """Return an identity-map key for use in storing/retrieving an
- item from an identity map.
-
- :param primary_key: A list of values indicating the identifier.
-
- """
- return (
- self._identity_class,
- tuple(primary_key),
- identity_token,
- )
-
- def identity_key_from_instance(self, instance: _O) -> _IdentityKeyType[_O]:
- """Return the identity key for the given instance, based on
- its primary key attributes.
-
- If the instance's state is expired, calling this method
- will result in a database check to see if the object has been deleted.
- If the row no longer exists,
- :class:`~sqlalchemy.orm.exc.ObjectDeletedError` is raised.
-
- This value is typically also found on the instance state under the
- attribute name `key`.
-
- """
- state = attributes.instance_state(instance)
- return self._identity_key_from_state(state, PassiveFlag.PASSIVE_OFF)
-
- def _identity_key_from_state(
- self,
- state: InstanceState[_O],
- passive: PassiveFlag = PassiveFlag.PASSIVE_RETURN_NO_VALUE,
- ) -> _IdentityKeyType[_O]:
- dict_ = state.dict
- manager = state.manager
- return (
- self._identity_class,
- tuple(
- [
- manager[prop.key].impl.get(state, dict_, passive)
- for prop in self._identity_key_props
- ]
- ),
- state.identity_token,
- )
-
- def primary_key_from_instance(self, instance: _O) -> Tuple[Any, ...]:
- """Return the list of primary key values for the given
- instance.
-
- If the instance's state is expired, calling this method
- will result in a database check to see if the object has been deleted.
- If the row no longer exists,
- :class:`~sqlalchemy.orm.exc.ObjectDeletedError` is raised.
-
- """
- state = attributes.instance_state(instance)
- identity_key = self._identity_key_from_state(
- state, PassiveFlag.PASSIVE_OFF
- )
- return identity_key[1]
-
- @HasMemoized.memoized_attribute
- def _persistent_sortkey_fn(self):
- key_fns = [col.type.sort_key_function for col in self.primary_key]
-
- if set(key_fns).difference([None]):
-
- def key(state):
- return tuple(
- key_fn(val) if key_fn is not None else val
- for key_fn, val in zip(key_fns, state.key[1])
- )
-
- else:
-
- def key(state):
- return state.key[1]
-
- return key
-
- @HasMemoized.memoized_attribute
- def _identity_key_props(self):
- return [self._columntoproperty[col] for col in self.primary_key]
-
- @HasMemoized.memoized_attribute
- def _all_pk_cols(self):
- collection: Set[ColumnClause[Any]] = set()
- for table in self.tables:
- collection.update(self._pks_by_table[table])
- return collection
-
- @HasMemoized.memoized_attribute
- def _should_undefer_in_wildcard(self):
- cols: Set[ColumnElement[Any]] = set(self.primary_key)
- if self.polymorphic_on is not None:
- cols.add(self.polymorphic_on)
- return cols
-
- @HasMemoized.memoized_attribute
- def _primary_key_propkeys(self):
- return {self._columntoproperty[col].key for col in self._all_pk_cols}
-
- def _get_state_attr_by_column(
- self,
- state: InstanceState[_O],
- dict_: _InstanceDict,
- column: ColumnElement[Any],
- passive: PassiveFlag = PassiveFlag.PASSIVE_RETURN_NO_VALUE,
- ) -> Any:
- prop = self._columntoproperty[column]
- return state.manager[prop.key].impl.get(state, dict_, passive=passive)
-
- def _set_committed_state_attr_by_column(self, state, dict_, column, value):
- prop = self._columntoproperty[column]
- state.manager[prop.key].impl.set_committed_value(state, dict_, value)
-
- def _set_state_attr_by_column(self, state, dict_, column, value):
- prop = self._columntoproperty[column]
- state.manager[prop.key].impl.set(state, dict_, value, None)
-
- def _get_committed_attr_by_column(self, obj, column):
- state = attributes.instance_state(obj)
- dict_ = attributes.instance_dict(obj)
- return self._get_committed_state_attr_by_column(
- state, dict_, column, passive=PassiveFlag.PASSIVE_OFF
- )
-
- def _get_committed_state_attr_by_column(
- self, state, dict_, column, passive=PassiveFlag.PASSIVE_RETURN_NO_VALUE
- ):
- prop = self._columntoproperty[column]
- return state.manager[prop.key].impl.get_committed_value(
- state, dict_, passive=passive
- )
-
- def _optimized_get_statement(self, state, attribute_names):
- """assemble a WHERE clause which retrieves a given state by primary
- key, using a minimized set of tables.
-
- Applies to a joined-table inheritance mapper where the
- requested attribute names are only present on joined tables,
- not the base table. The WHERE clause attempts to include
- only those tables to minimize joins.
-
- """
- props = self._props
-
- col_attribute_names = set(attribute_names).intersection(
- state.mapper.column_attrs.keys()
- )
- tables: Set[FromClause] = set(
- chain(
- *[
- sql_util.find_tables(c, check_columns=True)
- for key in col_attribute_names
- for c in props[key].columns
- ]
- )
- )
-
- if self.base_mapper.local_table in tables:
- return None
-
- def visit_binary(binary):
- leftcol = binary.left
- rightcol = binary.right
- if leftcol is None or rightcol is None:
- return
-
- if leftcol.table not in tables:
- leftval = self._get_committed_state_attr_by_column(
- state,
- state.dict,
- leftcol,
- passive=PassiveFlag.PASSIVE_NO_INITIALIZE,
- )
- if leftval in orm_util._none_set:
- raise _OptGetColumnsNotAvailable()
- binary.left = sql.bindparam(
- None, leftval, type_=binary.right.type
- )
- elif rightcol.table not in tables:
- rightval = self._get_committed_state_attr_by_column(
- state,
- state.dict,
- rightcol,
- passive=PassiveFlag.PASSIVE_NO_INITIALIZE,
- )
- if rightval in orm_util._none_set:
- raise _OptGetColumnsNotAvailable()
- binary.right = sql.bindparam(
- None, rightval, type_=binary.right.type
- )
-
- allconds: List[ColumnElement[bool]] = []
-
- start = False
-
- # as of #7507, from the lowest base table on upwards,
- # we include all intermediary tables.
-
- for mapper in reversed(list(self.iterate_to_root())):
- if mapper.local_table in tables:
- start = True
- elif not isinstance(mapper.local_table, expression.TableClause):
- return None
- if start and not mapper.single:
- assert mapper.inherits
- assert not mapper.concrete
- assert mapper.inherit_condition is not None
- allconds.append(mapper.inherit_condition)
- tables.add(mapper.local_table)
-
- # only the bottom table needs its criteria to be altered to fit
- # the primary key ident - the rest of the tables upwards to the
- # descendant-most class should all be present and joined to each
- # other.
- try:
- _traversed = visitors.cloned_traverse(
- allconds[0], {}, {"binary": visit_binary}
- )
- except _OptGetColumnsNotAvailable:
- return None
- else:
- allconds[0] = _traversed
-
- cond = sql.and_(*allconds)
-
- cols = []
- for key in col_attribute_names:
- cols.extend(props[key].columns)
- return (
- sql.select(*cols)
- .where(cond)
- .set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
- )
-
- def _iterate_to_target_viawpoly(self, mapper):
- if self.isa(mapper):
- prev = self
- for m in self.iterate_to_root():
- yield m
-
- if m is not prev and prev not in m._with_polymorphic_mappers:
- break
-
- prev = m
- if m is mapper:
- break
-
- @HasMemoized.memoized_attribute
- def _would_selectinload_combinations_cache(self):
- return {}
-
- def _would_selectin_load_only_from_given_mapper(self, super_mapper):
- """return True if this mapper would "selectin" polymorphic load based
- on the given super mapper, and not from a setting from a subclass.
-
- given::
-
- class A:
- ...
-
- class B(A):
- __mapper_args__ = {"polymorphic_load": "selectin"}
-
- class C(B):
- ...
-
- class D(B):
- __mapper_args__ = {"polymorphic_load": "selectin"}
-
- ``inspect(C)._would_selectin_load_only_from_given_mapper(inspect(B))``
- returns True, because C does selectin loading because of B's setting.
-
- OTOH, ``inspect(D)
- ._would_selectin_load_only_from_given_mapper(inspect(B))``
- returns False, because D does selectin loading because of its own
- setting; when we are doing a selectin poly load from B, we want to
- filter out D because it would already have its own selectin poly load
- set up separately.
-
- Added as part of #9373.
-
- """
- cache = self._would_selectinload_combinations_cache
-
- try:
- return cache[super_mapper]
- except KeyError:
- pass
-
- # assert that given object is a supermapper, meaning we already
- # strong reference it directly or indirectly. this allows us
- # to not worry that we are creating new strongrefs to unrelated
- # mappers or other objects.
- assert self.isa(super_mapper)
-
- mapper = super_mapper
- for m in self._iterate_to_target_viawpoly(mapper):
- if m.polymorphic_load == "selectin":
- retval = m is super_mapper
- break
- else:
- retval = False
-
- cache[super_mapper] = retval
- return retval
-
- def _should_selectin_load(self, enabled_via_opt, polymorphic_from):
- if not enabled_via_opt:
- # common case, takes place for all polymorphic loads
- mapper = polymorphic_from
- for m in self._iterate_to_target_viawpoly(mapper):
- if m.polymorphic_load == "selectin":
- return m
- else:
- # uncommon case, selectin load options were used
- enabled_via_opt = set(enabled_via_opt)
- enabled_via_opt_mappers = {e.mapper: e for e in enabled_via_opt}
- for entity in enabled_via_opt.union([polymorphic_from]):
- mapper = entity.mapper
- for m in self._iterate_to_target_viawpoly(mapper):
- if (
- m.polymorphic_load == "selectin"
- or m in enabled_via_opt_mappers
- ):
- return enabled_via_opt_mappers.get(m, m)
-
- return None
-
- @util.preload_module("sqlalchemy.orm.strategy_options")
- def _subclass_load_via_in(self, entity, polymorphic_from):
- """Assemble a that can load the columns local to
- this subclass as a SELECT with IN.
-
- """
- strategy_options = util.preloaded.orm_strategy_options
-
- assert self.inherits
-
- if self.polymorphic_on is not None:
- polymorphic_prop = self._columntoproperty[self.polymorphic_on]
- keep_props = set([polymorphic_prop] + self._identity_key_props)
- else:
- keep_props = set(self._identity_key_props)
-
- disable_opt = strategy_options.Load(entity)
- enable_opt = strategy_options.Load(entity)
-
- classes_to_include = {self}
- m: Optional[Mapper[Any]] = self.inherits
- while (
- m is not None
- and m is not polymorphic_from
- and m.polymorphic_load == "selectin"
- ):
- classes_to_include.add(m)
- m = m.inherits
-
- for prop in self.attrs:
- # skip prop keys that are not instrumented on the mapped class.
- # this is primarily the "_sa_polymorphic_on" property that gets
- # created for an ad-hoc polymorphic_on SQL expression, issue #8704
- if prop.key not in self.class_manager:
- continue
-
- if prop.parent in classes_to_include or prop in keep_props:
- # "enable" options, to turn on the properties that we want to
- # load by default (subject to options from the query)
- if not isinstance(prop, StrategizedProperty):
- continue
-
- enable_opt = enable_opt._set_generic_strategy(
- # convert string name to an attribute before passing
- # to loader strategy. note this must be in terms
- # of given entity, such as AliasedClass, etc.
- (getattr(entity.entity_namespace, prop.key),),
- dict(prop.strategy_key),
- _reconcile_to_other=True,
- )
- else:
- # "disable" options, to turn off the properties from the
- # superclass that we *don't* want to load, applied after
- # the options from the query to override them
- disable_opt = disable_opt._set_generic_strategy(
- # convert string name to an attribute before passing
- # to loader strategy. note this must be in terms
- # of given entity, such as AliasedClass, etc.
- (getattr(entity.entity_namespace, prop.key),),
- {"do_nothing": True},
- _reconcile_to_other=False,
- )
-
- primary_key = [
- sql_util._deep_annotate(pk, {"_orm_adapt": True})
- for pk in self.primary_key
- ]
-
- in_expr: ColumnElement[Any]
-
- if len(primary_key) > 1:
- in_expr = sql.tuple_(*primary_key)
- else:
- in_expr = primary_key[0]
-
- if entity.is_aliased_class:
- assert entity.mapper is self
-
- q = sql.select(entity).set_label_style(
- LABEL_STYLE_TABLENAME_PLUS_COL
- )
-
- in_expr = entity._adapter.traverse(in_expr)
- primary_key = [entity._adapter.traverse(k) for k in primary_key]
- q = q.where(
- in_expr.in_(sql.bindparam("primary_keys", expanding=True))
- ).order_by(*primary_key)
- else:
- q = sql.select(self).set_label_style(
- LABEL_STYLE_TABLENAME_PLUS_COL
- )
- q = q.where(
- in_expr.in_(sql.bindparam("primary_keys", expanding=True))
- ).order_by(*primary_key)
-
- return q, enable_opt, disable_opt
-
- @HasMemoized.memoized_attribute
- def _subclass_load_via_in_mapper(self):
- # the default is loading this mapper against the basemost mapper
- return self._subclass_load_via_in(self, self.base_mapper)
-
- def cascade_iterator(
- self,
- type_: str,
- state: InstanceState[_O],
- halt_on: Optional[Callable[[InstanceState[Any]], bool]] = None,
- ) -> Iterator[
- Tuple[object, Mapper[Any], InstanceState[Any], _InstanceDict]
- ]:
- r"""Iterate each element and its mapper in an object graph,
- for all relationships that meet the given cascade rule.
-
- :param type\_:
- The name of the cascade rule (i.e. ``"save-update"``, ``"delete"``,
- etc.).
-
- .. note:: the ``"all"`` cascade is not accepted here. For a generic
- object traversal function, see :ref:`faq_walk_objects`.
-
- :param state:
- The lead InstanceState. child items will be processed per
- the relationships defined for this object's mapper.
-
- :return: the method yields individual object instances.
-
- .. seealso::
-
- :ref:`unitofwork_cascades`
-
- :ref:`faq_walk_objects` - illustrates a generic function to
- traverse all objects without relying on cascades.
-
- """
- visited_states: Set[InstanceState[Any]] = set()
- prp, mpp = object(), object()
-
- assert state.mapper.isa(self)
-
- # this is actually a recursive structure, fully typing it seems
- # a little too difficult for what it's worth here
- visitables: Deque[
- Tuple[
- Deque[Any],
- object,
- Optional[InstanceState[Any]],
- Optional[_InstanceDict],
- ]
- ]
-
- visitables = deque(
- [(deque(state.mapper._props.values()), prp, state, state.dict)]
- )
-
- while visitables:
- iterator, item_type, parent_state, parent_dict = visitables[-1]
- if not iterator:
- visitables.pop()
- continue
-
- if item_type is prp:
- prop = iterator.popleft()
- if not prop.cascade or type_ not in prop.cascade:
- continue
- assert parent_state is not None
- assert parent_dict is not None
- queue = deque(
- prop.cascade_iterator(
- type_,
- parent_state,
- parent_dict,
- visited_states,
- halt_on,
- )
- )
- if queue:
- visitables.append((queue, mpp, None, None))
- elif item_type is mpp:
- (
- instance,
- instance_mapper,
- corresponding_state,
- corresponding_dict,
- ) = iterator.popleft()
- yield (
- instance,
- instance_mapper,
- corresponding_state,
- corresponding_dict,
- )
- visitables.append(
- (
- deque(instance_mapper._props.values()),
- prp,
- corresponding_state,
- corresponding_dict,
- )
- )
-
- @HasMemoized.memoized_attribute
- def _compiled_cache(self):
- return util.LRUCache(self._compiled_cache_size)
-
- @HasMemoized.memoized_attribute
- def _multiple_persistence_tables(self):
- return len(self.tables) > 1
-
- @HasMemoized.memoized_attribute
- def _sorted_tables(self):
- table_to_mapper: Dict[TableClause, Mapper[Any]] = {}
-
- for mapper in self.base_mapper.self_and_descendants:
- for t in mapper.tables:
- table_to_mapper.setdefault(t, mapper)
-
- extra_dependencies = []
- for table, mapper in table_to_mapper.items():
- super_ = mapper.inherits
- if super_:
- extra_dependencies.extend(
- [(super_table, table) for super_table in super_.tables]
- )
-
- def skip(fk):
- # attempt to skip dependencies that are not
- # significant to the inheritance chain
- # for two tables that are related by inheritance.
- # while that dependency may be important, it's technically
- # not what we mean to sort on here.
- parent = table_to_mapper.get(fk.parent.table)
- dep = table_to_mapper.get(fk.column.table)
- if (
- parent is not None
- and dep is not None
- and dep is not parent
- and dep.inherit_condition is not None
- ):
- cols = set(sql_util._find_columns(dep.inherit_condition))
- if parent.inherit_condition is not None:
- cols = cols.union(
- sql_util._find_columns(parent.inherit_condition)
- )
- return fk.parent not in cols and fk.column not in cols
- else:
- return fk.parent not in cols
- return False
-
- sorted_ = sql_util.sort_tables(
- table_to_mapper,
- skip_fn=skip,
- extra_dependencies=extra_dependencies,
- )
-
- ret = util.OrderedDict()
- for t in sorted_:
- ret[t] = table_to_mapper[t]
- return ret
-
- def _memo(self, key: Any, callable_: Callable[[], _T]) -> _T:
- if key in self._memoized_values:
- return cast(_T, self._memoized_values[key])
- else:
- self._memoized_values[key] = value = callable_()
- return value
-
- @util.memoized_property
- def _table_to_equated(self):
- """memoized map of tables to collections of columns to be
- synchronized upwards to the base mapper."""
-
- result: util.defaultdict[
- Table,
- List[
- Tuple[
- Mapper[Any],
- List[Tuple[ColumnElement[Any], ColumnElement[Any]]],
- ]
- ],
- ] = util.defaultdict(list)
-
- def set_union(x, y):
- return x.union(y)
-
- for table in self._sorted_tables:
- cols = set(table.c)
-
- for m in self.iterate_to_root():
- if m._inherits_equated_pairs and cols.intersection(
- reduce(
- set_union,
- [l.proxy_set for l, r in m._inherits_equated_pairs],
- )
- ):
- result[table].append((m, m._inherits_equated_pairs))
-
- return result
-
-
-class _OptGetColumnsNotAvailable(Exception):
- pass
-
-
-def configure_mappers() -> None:
- """Initialize the inter-mapper relationships of all mappers that
- have been constructed thus far across all :class:`_orm.registry`
- collections.
-
- The configure step is used to reconcile and initialize the
- :func:`_orm.relationship` linkages between mapped classes, as well as to
- invoke configuration events such as the
- :meth:`_orm.MapperEvents.before_configured` and
- :meth:`_orm.MapperEvents.after_configured`, which may be used by ORM
- extensions or user-defined extension hooks.
-
- Mapper configuration is normally invoked automatically, the first time
- mappings from a particular :class:`_orm.registry` are used, as well as
- whenever mappings are used and additional not-yet-configured mappers have
- been constructed. The automatic configuration process however is local only
- to the :class:`_orm.registry` involving the target mapper and any related
- :class:`_orm.registry` objects which it may depend on; this is
- equivalent to invoking the :meth:`_orm.registry.configure` method
- on a particular :class:`_orm.registry`.
-
- By contrast, the :func:`_orm.configure_mappers` function will invoke the
- configuration process on all :class:`_orm.registry` objects that
- exist in memory, and may be useful for scenarios where many individual
- :class:`_orm.registry` objects that are nonetheless interrelated are
- in use.
-
- .. versionchanged:: 1.4
-
- As of SQLAlchemy 1.4.0b2, this function works on a
- per-:class:`_orm.registry` basis, locating all :class:`_orm.registry`
- objects present and invoking the :meth:`_orm.registry.configure` method
- on each. The :meth:`_orm.registry.configure` method may be preferred to
- limit the configuration of mappers to those local to a particular
- :class:`_orm.registry` and/or declarative base class.
-
- Points at which automatic configuration is invoked include when a mapped
- class is instantiated into an instance, as well as when ORM queries
- are emitted using :meth:`.Session.query` or :meth:`_orm.Session.execute`
- with an ORM-enabled statement.
-
- The mapper configure process, whether invoked by
- :func:`_orm.configure_mappers` or from :meth:`_orm.registry.configure`,
- provides several event hooks that can be used to augment the mapper
- configuration step. These hooks include:
-
- * :meth:`.MapperEvents.before_configured` - called once before
- :func:`.configure_mappers` or :meth:`_orm.registry.configure` does any
- work; this can be used to establish additional options, properties, or
- related mappings before the operation proceeds.
-
- * :meth:`.MapperEvents.mapper_configured` - called as each individual
- :class:`_orm.Mapper` is configured within the process; will include all
- mapper state except for backrefs set up by other mappers that are still
- to be configured.
-
- * :meth:`.MapperEvents.after_configured` - called once after
- :func:`.configure_mappers` or :meth:`_orm.registry.configure` is
- complete; at this stage, all :class:`_orm.Mapper` objects that fall
- within the scope of the configuration operation will be fully configured.
- Note that the calling application may still have other mappings that
- haven't been produced yet, such as if they are in modules as yet
- unimported, and may also have mappings that are still to be configured,
- if they are in other :class:`_orm.registry` collections not part of the
- current scope of configuration.
-
- """
-
- _configure_registries(_all_registries(), cascade=True)
-
-
-def _configure_registries(
- registries: Set[_RegistryType], cascade: bool
-) -> None:
- for reg in registries:
- if reg._new_mappers:
- break
- else:
- return
-
- with _CONFIGURE_MUTEX:
- global _already_compiling
- if _already_compiling:
- return
- _already_compiling = True
- try:
- # double-check inside mutex
- for reg in registries:
- if reg._new_mappers:
- break
- else:
- return
-
- Mapper.dispatch._for_class(Mapper).before_configured() # type: ignore # noqa: E501
- # initialize properties on all mappers
- # note that _mapper_registry is unordered, which
- # may randomly conceal/reveal issues related to
- # the order of mapper compilation
-
- _do_configure_registries(registries, cascade)
- finally:
- _already_compiling = False
- Mapper.dispatch._for_class(Mapper).after_configured() # type: ignore
-
-
-@util.preload_module("sqlalchemy.orm.decl_api")
-def _do_configure_registries(
- registries: Set[_RegistryType], cascade: bool
-) -> None:
- registry = util.preloaded.orm_decl_api.registry
-
- orig = set(registries)
-
- for reg in registry._recurse_with_dependencies(registries):
- has_skip = False
-
- 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:
- has_skip = True
- break
- if run_configure is EXT_SKIP:
- continue
-
- if getattr(mapper, "_configure_failed", False):
- e = sa_exc.InvalidRequestError(
- "One or more mappers failed to initialize - "
- "can't proceed with initialization of other "
- "mappers. Triggering mapper: '%s'. "
- "Original exception was: %s"
- % (mapper, mapper._configure_failed)
- )
- e._configure_failed = mapper._configure_failed # type: ignore
- raise e
-
- if not mapper.configured:
- try:
- mapper._post_configure_properties()
- mapper._expire_memoizations()
- mapper.dispatch.mapper_configured(mapper, mapper.class_)
- except Exception:
- exc = sys.exc_info()[1]
- if not hasattr(exc, "_configure_failed"):
- mapper._configure_failed = exc
- raise
- if not has_skip:
- reg._new_mappers = False
-
- if not cascade and reg._dependencies.difference(orig):
- raise sa_exc.InvalidRequestError(
- "configure was called with cascade=False but "
- "additional registries remain"
- )
-
-
-@util.preload_module("sqlalchemy.orm.decl_api")
-def _dispose_registries(registries: Set[_RegistryType], cascade: bool) -> None:
- registry = util.preloaded.orm_decl_api.registry
-
- orig = set(registries)
-
- for reg in registry._recurse_with_dependents(registries):
- if not cascade and reg._dependents.difference(orig):
- raise sa_exc.InvalidRequestError(
- "Registry has dependent registries that are not disposed; "
- "pass cascade=True to clear these also"
- )
-
- while reg._managers:
- try:
- manager, _ = reg._managers.popitem()
- except KeyError:
- # guard against race between while and popitem
- pass
- else:
- reg._dispose_manager_and_mapper(manager)
-
- reg._non_primary_mappers.clear()
- reg._dependents.clear()
- for dep in reg._dependencies:
- dep._dependents.discard(reg)
- reg._dependencies.clear()
- # this wasn't done in the 1.3 clear_mappers() and in fact it
- # was a bug, as it could cause configure_mappers() to invoke
- # the "before_configured" event even though mappers had all been
- # disposed.
- reg._new_mappers = False
-
-
-def reconstructor(fn):
- """Decorate a method as the 'reconstructor' hook.
-
- Designates a single method as the "reconstructor", an ``__init__``-like
- method that will be called by the ORM after the instance has been
- loaded from the database or otherwise reconstituted.
-
- .. tip::
-
- The :func:`_orm.reconstructor` decorator makes use of the
- :meth:`_orm.InstanceEvents.load` event hook, which can be
- used directly.
-
- The reconstructor will be invoked with no arguments. Scalar
- (non-collection) database-mapped attributes of the instance will
- be available for use within the function. Eagerly-loaded
- collections are generally not yet available and will usually only
- contain the first element. ORM state changes made to objects at
- this stage will not be recorded for the next flush() operation, so
- the activity within a reconstructor should be conservative.
-
- .. seealso::
-
- :meth:`.InstanceEvents.load`
-
- """
- fn.__sa_reconstructor__ = True
- return fn
-
-
-def validates(
- *names: str, include_removes: bool = False, include_backrefs: bool = True
-) -> Callable[[_Fn], _Fn]:
- r"""Decorate a method as a 'validator' for one or more named properties.
-
- Designates a method as a validator, a method which receives the
- name of the attribute as well as a value to be assigned, or in the
- case of a collection, the value to be added to the collection.
- The function can then raise validation exceptions to halt the
- process from continuing (where Python's built-in ``ValueError``
- and ``AssertionError`` exceptions are reasonable choices), or can
- modify or replace the value before proceeding. The function should
- otherwise return the given value.
-
- Note that a validator for a collection **cannot** issue a load of that
- collection within the validation routine - this usage raises
- an assertion to avoid recursion overflows. This is a reentrant
- condition which is not supported.
-
- :param \*names: list of attribute names to be validated.
- :param include_removes: if True, "remove" events will be
- sent as well - the validation function must accept an additional
- argument "is_remove" which will be a boolean.
-
- :param include_backrefs: defaults to ``True``; if ``False``, the
- validation function will not emit if the originator is an attribute
- event related via a backref. This can be used for bi-directional
- :func:`.validates` usage where only one validator should emit per
- attribute operation.
-
- .. versionchanged:: 2.0.16 This paramter inadvertently defaulted to
- ``False`` for releases 2.0.0 through 2.0.15. Its correct default
- of ``True`` is restored in 2.0.16.
-
- .. seealso::
-
- :ref:`simple_validators` - usage examples for :func:`.validates`
-
- """
-
- def wrap(fn: _Fn) -> _Fn:
- fn.__sa_validators__ = names # type: ignore[attr-defined]
- fn.__sa_validation_opts__ = { # type: ignore[attr-defined]
- "include_removes": include_removes,
- "include_backrefs": include_backrefs,
- }
- return fn
-
- return wrap
-
-
-def _event_on_load(state, ctx):
- instrumenting_mapper = state.manager.mapper
-
- if instrumenting_mapper._reconstructor:
- instrumenting_mapper._reconstructor(state.obj())
-
-
-def _event_on_init(state, args, kwargs):
- """Run init_instance hooks.
-
- This also includes mapper compilation, normally not needed
- here but helps with some piecemeal configuration
- scenarios (such as in the ORM tutorial).
-
- """
-
- instrumenting_mapper = state.manager.mapper
- if instrumenting_mapper:
- instrumenting_mapper._check_configure()
- if instrumenting_mapper._set_polymorphic_identity:
- instrumenting_mapper._set_polymorphic_identity(state)
-
-
-class _ColumnMapping(Dict["ColumnElement[Any]", "MapperProperty[Any]"]):
- """Error reporting helper for mapper._columntoproperty."""
-
- __slots__ = ("mapper",)
-
- def __init__(self, mapper):
- # TODO: weakref would be a good idea here
- self.mapper = mapper
-
- def __missing__(self, column):
- prop = self.mapper._props.get(column)
- if prop:
- raise orm_exc.UnmappedColumnError(
- "Column '%s.%s' is not available, due to "
- "conflicting property '%s':%r"
- % (column.table.name, column.name, column.key, prop)
- )
- raise orm_exc.UnmappedColumnError(
- "No column %s is configured on mapper %s..."
- % (column, self.mapper)
- )