Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/util.py: 32%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# orm/util.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: allow-untyped-defs, allow-untyped-calls
9from __future__ import annotations
11import enum
12import functools
13import itertools
14import re
15import types
16import typing
17from typing import AbstractSet
18from typing import Any
19from typing import Callable
20from typing import cast
21from typing import Dict
22from typing import FrozenSet
23from typing import Generic
24from typing import get_origin
25from typing import Iterable
26from typing import Iterator
27from typing import List
28from typing import Literal
29from typing import Match
30from typing import Optional
31from typing import Protocol
32from typing import Sequence
33from typing import Tuple
34from typing import Type
35from typing import TYPE_CHECKING
36from typing import TypeVar
37from typing import Union
38import weakref
40from . import attributes # noqa
41from . import exc as orm_exc
42from ._typing import _O
43from ._typing import insp_is_aliased_class
44from ._typing import insp_is_mapper
45from ._typing import prop_is_relationship
46from .base import _class_to_mapper as _class_to_mapper
47from .base import _MappedAnnotationBase
48from .base import _never_set as _never_set # noqa: F401
49from .base import _none_only_set as _none_only_set # noqa: F401
50from .base import _none_set as _none_set # noqa: F401
51from .base import attribute_str as attribute_str # noqa: F401
52from .base import class_mapper as class_mapper
53from .base import DynamicMapped
54from .base import InspectionAttr as InspectionAttr
55from .base import instance_str as instance_str # noqa: F401
56from .base import Mapped
57from .base import object_mapper as object_mapper
58from .base import object_state as object_state # noqa: F401
59from .base import opt_manager_of_class
60from .base import ORMDescriptor
61from .base import state_attribute_str as state_attribute_str # noqa: F401
62from .base import state_class_str as state_class_str # noqa: F401
63from .base import state_str as state_str # noqa: F401
64from .base import WriteOnlyMapped
65from .interfaces import CriteriaOption
66from .interfaces import MapperProperty as MapperProperty
67from .interfaces import ORMColumnsClauseRole
68from .interfaces import ORMEntityColumnsClauseRole
69from .interfaces import ORMFromClauseRole
70from .path_registry import PathRegistry as PathRegistry
71from .. import event
72from .. import exc as sa_exc
73from .. import inspection
74from .. import sql
75from .. import util
76from ..engine.result import result_tuple
77from ..sql import coercions
78from ..sql import expression
79from ..sql import lambdas
80from ..sql import roles
81from ..sql import util as sql_util
82from ..sql import visitors
83from ..sql._typing import is_selectable
84from ..sql.annotation import SupportsCloneAnnotations
85from ..sql.base import WriteableColumnCollection
86from ..sql.cache_key import HasCacheKey
87from ..sql.cache_key import MemoizedHasCacheKey
88from ..sql.elements import ColumnElement
89from ..sql.elements import KeyedColumnElement
90from ..sql.schema import MetaData
91from ..sql.selectable import FromClause
92from ..sql.selectable import GenerativeSelect
93from ..util.langhelpers import MemoizedSlots
94from ..util.typing import de_stringify_annotation as _de_stringify_annotation
95from ..util.typing import eval_name_only as _eval_name_only
96from ..util.typing import fixup_container_fwd_refs
97from ..util.typing import GenericProtocol
98from ..util.typing import is_origin_of_cls
99from ..util.typing import TupleAny
100from ..util.typing import Unpack
102if typing.TYPE_CHECKING:
103 from ._typing import _EntityType
104 from ._typing import _IdentityKeyType
105 from ._typing import _InternalEntityType
106 from ._typing import _ORMCOLEXPR
107 from .context import _MapperEntity
108 from .context import _ORMCompileState
109 from .decl_api import RegistryType
110 from .mapper import Mapper
111 from .path_registry import _AbstractEntityRegistry
112 from .query import Query
113 from .relationships import RelationshipProperty
114 from ..engine import Row
115 from ..engine import RowMapping
116 from ..sql._typing import _CE
117 from ..sql._typing import _ColumnExpressionArgument
118 from ..sql._typing import _EquivalentColumnMap
119 from ..sql._typing import _FromClauseArgument
120 from ..sql._typing import _OnClauseArgument
121 from ..sql._typing import _PropagateAttrsType
122 from ..sql.annotation import _SA
123 from ..sql.base import ReadOnlyColumnCollection
124 from ..sql.elements import BindParameter
125 from ..sql.selectable import _ColumnsClauseElement
126 from ..sql.selectable import Select
127 from ..sql.selectable import Selectable
128 from ..sql.visitors import anon_map
129 from ..util.typing import _AnnotationScanType
130 from ..util.typing import _MatchedOnType
132_T = TypeVar("_T", bound=Any)
134all_cascades = frozenset(
135 (
136 "delete",
137 "delete-orphan",
138 "all",
139 "merge",
140 "expunge",
141 "save-update",
142 "refresh-expire",
143 "none",
144 )
145)
147_de_stringify_partial = functools.partial(
148 functools.partial,
149 locals_=util.immutabledict(
150 {
151 "Mapped": Mapped,
152 "WriteOnlyMapped": WriteOnlyMapped,
153 "DynamicMapped": DynamicMapped,
154 }
155 ),
156)
158# partial is practically useless as we have to write out the whole
159# function and maintain the signature anyway
162class _DeStringifyAnnotation(Protocol):
163 def __call__(
164 self,
165 cls: Type[Any],
166 annotation: _AnnotationScanType,
167 originating_module: str,
168 *,
169 str_cleanup_fn: Optional[Callable[[str, str], str]] = None,
170 include_generic: bool = False,
171 ) -> _MatchedOnType: ...
174de_stringify_annotation = cast(
175 _DeStringifyAnnotation, _de_stringify_partial(_de_stringify_annotation)
176)
179class _EvalNameOnly(Protocol):
180 def __call__(self, name: str, module_name: str) -> Any: ...
183eval_name_only = cast(_EvalNameOnly, _de_stringify_partial(_eval_name_only))
186class CascadeOptions(FrozenSet[str]):
187 """Keeps track of the options sent to
188 :paramref:`.relationship.cascade`"""
190 _add_w_all_cascades = all_cascades.difference(
191 ["all", "none", "delete-orphan"]
192 )
193 _allowed_cascades = all_cascades
195 _viewonly_cascades = ["expunge", "all", "none", "refresh-expire", "merge"]
197 __slots__ = (
198 "save_update",
199 "delete",
200 "refresh_expire",
201 "merge",
202 "expunge",
203 "delete_orphan",
204 )
206 save_update: bool
207 delete: bool
208 refresh_expire: bool
209 merge: bool
210 expunge: bool
211 delete_orphan: bool
213 def __new__(
214 cls, value_list: Optional[Union[Iterable[str], str]]
215 ) -> CascadeOptions:
216 if isinstance(value_list, str) or value_list is None:
217 return cls.from_string(value_list) # type: ignore[no-any-return]
218 values = set(value_list)
219 if values.difference(cls._allowed_cascades):
220 raise sa_exc.ArgumentError(
221 "Invalid cascade option(s): %s"
222 % ", ".join(
223 [
224 repr(x)
225 for x in sorted(
226 values.difference(cls._allowed_cascades)
227 )
228 ]
229 )
230 )
232 if "all" in values:
233 values.update(cls._add_w_all_cascades)
234 if "none" in values:
235 values.clear()
236 values.discard("all")
238 self = super().__new__(cls, values)
239 self.save_update = "save-update" in values
240 self.delete = "delete" in values
241 self.refresh_expire = "refresh-expire" in values
242 self.merge = "merge" in values
243 self.expunge = "expunge" in values
244 self.delete_orphan = "delete-orphan" in values
246 if self.delete_orphan and not self.delete:
247 util.warn("The 'delete-orphan' cascade option requires 'delete'.")
248 return self
250 def __repr__(self):
251 return "CascadeOptions(%r)" % (",".join([x for x in sorted(self)]))
253 @classmethod
254 def from_string(cls, arg):
255 values = [c for c in re.split(r"\s*,\s*", arg or "") if c]
256 return cls(values)
259def _metadata_for_cls(cls: Type[Any], registry: RegistryType) -> MetaData:
260 meta = getattr(cls, "metadata", None)
261 if meta is not None and isinstance(meta, MetaData):
262 return meta
263 return registry.metadata
266def _validator_events(desc, key, validator, include_removes, include_backrefs):
267 """Runs a validation method on an attribute value to be set or
268 appended.
269 """
271 if not include_backrefs:
273 def detect_is_backref(state, initiator):
274 impl = state.manager[key].impl
275 return initiator.impl is not impl
277 if include_removes:
279 def append(state, value, initiator):
280 if initiator.op is not attributes.OP_BULK_REPLACE and (
281 include_backrefs or not detect_is_backref(state, initiator)
282 ):
283 return validator(state.obj(), key, value, False)
284 else:
285 return value
287 def bulk_set(state, values, initiator):
288 if include_backrefs or not detect_is_backref(state, initiator):
289 obj = state.obj()
290 values[:] = [
291 validator(obj, key, value, False) for value in values
292 ]
294 def set_(state, value, oldvalue, initiator):
295 if include_backrefs or not detect_is_backref(state, initiator):
296 return validator(state.obj(), key, value, False)
297 else:
298 return value
300 def remove(state, value, initiator):
301 if include_backrefs or not detect_is_backref(state, initiator):
302 validator(state.obj(), key, value, True)
304 else:
306 def append(state, value, initiator):
307 if initiator.op is not attributes.OP_BULK_REPLACE and (
308 include_backrefs or not detect_is_backref(state, initiator)
309 ):
310 return validator(state.obj(), key, value)
311 else:
312 return value
314 def bulk_set(state, values, initiator):
315 if include_backrefs or not detect_is_backref(state, initiator):
316 obj = state.obj()
317 values[:] = [validator(obj, key, value) for value in values]
319 def set_(state, value, oldvalue, initiator):
320 if include_backrefs or not detect_is_backref(state, initiator):
321 return validator(state.obj(), key, value)
322 else:
323 return value
325 event.listen(desc, "append", append, raw=True, retval=True)
326 event.listen(desc, "bulk_replace", bulk_set, raw=True)
327 event.listen(desc, "set", set_, raw=True, retval=True)
328 if include_removes:
329 event.listen(desc, "remove", remove, raw=True, retval=True)
332def polymorphic_union(
333 table_map, typecolname, aliasname="p_union", cast_nulls=True
334):
335 """Create a ``UNION`` statement used by a polymorphic mapper.
337 See :ref:`concrete_inheritance` for an example of how
338 this is used.
340 :param table_map: mapping of polymorphic identities to
341 :class:`_schema.Table` objects.
342 :param typecolname: string name of a "discriminator" column, which will be
343 derived from the query, producing the polymorphic identity for
344 each row. If ``None``, no polymorphic discriminator is generated.
345 :param aliasname: name of the :func:`~sqlalchemy.sql.expression.alias()`
346 construct generated.
347 :param cast_nulls: if True, non-existent columns, which are represented
348 as labeled NULLs, will be passed into CAST. This is a legacy behavior
349 that is problematic on some backends such as Oracle - in which case it
350 can be set to False.
352 """
354 colnames: util.OrderedSet[str] = util.OrderedSet()
355 colnamemaps = {}
356 types = {}
357 for key in table_map:
358 table = table_map[key]
360 table = coercions.expect(roles.FromClauseRole, table)
361 table_map[key] = table
363 m = {}
364 for c in table.c:
365 if c.key == typecolname:
366 raise sa_exc.InvalidRequestError(
367 "Polymorphic union can't use '%s' as the discriminator "
368 "column due to mapped column %r; please apply the "
369 "'typecolname' "
370 "argument; this is available on "
371 "ConcreteBase as '_concrete_discriminator_name'"
372 % (typecolname, c)
373 )
374 colnames.add(c.key)
375 m[c.key] = c
376 types[c.key] = c.type
377 colnamemaps[table] = m
379 def col(name, table):
380 try:
381 return colnamemaps[table][name]
382 except KeyError:
383 if cast_nulls:
384 return sql.cast(sql.null(), types[name]).label(name)
385 else:
386 return sql.type_coerce(sql.null(), types[name]).label(name)
388 result = []
389 for type_, table in table_map.items():
390 if typecolname is not None:
391 result.append(
392 sql.select(
393 *(
394 [col(name, table) for name in colnames]
395 + [
396 sql.literal_column(
397 sql_util._quote_ddl_expr(type_)
398 ).label(typecolname)
399 ]
400 )
401 ).select_from(table)
402 )
403 else:
404 result.append(
405 sql.select(
406 *[col(name, table) for name in colnames]
407 ).select_from(table)
408 )
409 return sql.union_all(*result).alias(aliasname)
412def identity_key(
413 class_: Optional[Type[_T]] = None,
414 ident: Union[Any, Tuple[Any, ...]] = None,
415 *,
416 instance: Optional[_T] = None,
417 row: Optional[Union[Row[Unpack[TupleAny]], RowMapping]] = None,
418 identity_token: Optional[Any] = None,
419) -> _IdentityKeyType[_T]:
420 r"""Generate "identity key" tuples, as are used as keys in the
421 :attr:`.Session.identity_map` dictionary.
423 This function has several call styles:
425 * ``identity_key(class, ident, identity_token=token)``
427 This form receives a mapped class and a primary key scalar or
428 tuple as an argument.
430 E.g.::
432 >>> identity_key(MyClass, (1, 2))
433 (<class '__main__.MyClass'>, (1, 2), None)
435 :param class: mapped class (must be a positional argument)
436 :param ident: primary key, may be a scalar or tuple argument.
437 :param identity_token: optional identity token
439 * ``identity_key(instance=instance)``
441 This form will produce the identity key for a given instance. The
442 instance need not be persistent, only that its primary key attributes
443 are populated (else the key will contain ``None`` for those missing
444 values).
446 E.g.::
448 >>> instance = MyClass(1, 2)
449 >>> identity_key(instance=instance)
450 (<class '__main__.MyClass'>, (1, 2), None)
452 In this form, the given instance is ultimately run though
453 :meth:`_orm.Mapper.identity_key_from_instance`, which will have the
454 effect of performing a database check for the corresponding row
455 if the object is expired.
457 :param instance: object instance (must be given as a keyword arg)
459 * ``identity_key(class, row=row, identity_token=token)``
461 This form is similar to the class/tuple form, except is passed a
462 database result row as a :class:`.Row` or :class:`.RowMapping` object.
464 E.g.::
466 >>> row = engine.execute(text("select * from table where a=1 and b=2")).first()
467 >>> identity_key(MyClass, row=row)
468 (<class '__main__.MyClass'>, (1, 2), None)
470 :param class: mapped class (must be a positional argument)
471 :param row: :class:`.Row` row returned by a :class:`_engine.CursorResult`
472 (must be given as a keyword arg)
473 :param identity_token: optional identity token
475 """ # noqa: E501
476 if class_ is not None:
477 mapper = class_mapper(class_)
478 if row is None:
479 if ident is None:
480 raise sa_exc.ArgumentError("ident or row is required")
481 return mapper.identity_key_from_primary_key(
482 tuple(util.to_list(ident)), identity_token=identity_token
483 )
484 else:
485 return mapper.identity_key_from_row(
486 row, identity_token=identity_token
487 )
488 elif instance is not None:
489 mapper = object_mapper(instance)
490 return mapper.identity_key_from_instance(instance)
491 else:
492 raise sa_exc.ArgumentError("class or instance is required")
495class _TraceAdaptRole(enum.Enum):
496 """Enumeration of all the use cases for ORMAdapter.
498 ORMAdapter remains one of the most complicated aspects of the ORM, as it is
499 used for in-place adaption of column expressions to be applied to a SELECT,
500 replacing :class:`.Table` and other objects that are mapped to classes with
501 aliases of those tables in the case of joined eager loading, or in the case
502 of polymorphic loading as used with concrete mappings or other custom "with
503 polymorphic" parameters, with whole user-defined subqueries. The
504 enumerations provide an overview of all the use cases used by ORMAdapter, a
505 layer of formality as to the introduction of new ORMAdapter use cases (of
506 which none are anticipated), as well as a means to trace the origins of a
507 particular ORMAdapter within runtime debugging.
509 SQLAlchemy 2.0 has greatly scaled back ORM features which relied heavily on
510 open-ended statement adaption, including the ``Query.with_polymorphic()``
511 method and the ``Query.select_from_entity()`` methods, favoring
512 user-explicit aliasing schemes using the ``aliased()`` and
513 ``with_polymorphic()`` standalone constructs; these still use adaption,
514 however the adaption is applied in a narrower scope.
516 """
518 # aliased() use that is used to adapt individual attributes at query
519 # construction time
520 ALIASED_INSP = enum.auto()
522 # joinedload cases; typically adapt an ON clause of a relationship
523 # join
524 JOINEDLOAD_USER_DEFINED_ALIAS = enum.auto()
525 JOINEDLOAD_PATH_WITH_POLYMORPHIC = enum.auto()
526 JOINEDLOAD_MEMOIZED_ADAPTER = enum.auto()
528 # polymorphic cases - these are complex ones that replace FROM
529 # clauses, replacing tables with subqueries
530 MAPPER_POLYMORPHIC_ADAPTER = enum.auto()
531 WITH_POLYMORPHIC_ADAPTER = enum.auto()
532 WITH_POLYMORPHIC_ADAPTER_RIGHT_JOIN = enum.auto()
533 DEPRECATED_JOIN_ADAPT_RIGHT_SIDE = enum.auto()
535 # the from_statement() case, used only to adapt individual attributes
536 # from a given statement to local ORM attributes at result fetching
537 # time. assigned to ORMCompileState._from_obj_alias
538 ADAPT_FROM_STATEMENT = enum.auto()
540 # the joinedload for queries that have LIMIT/OFFSET/DISTINCT case;
541 # the query is placed inside of a subquery with the LIMIT/OFFSET/etc.,
542 # joinedloads are then placed on the outside.
543 # assigned to ORMCompileState.compound_eager_adapter
544 COMPOUND_EAGER_STATEMENT = enum.auto()
546 # the legacy Query._set_select_from() case.
547 # this is needed for Query's set operations (i.e. UNION, etc. )
548 # as well as "legacy from_self()", which while removed from 2.0 as
549 # public API, is used for the Query.count() method. this one
550 # still does full statement traversal
551 # assigned to ORMCompileState._from_obj_alias
552 LEGACY_SELECT_FROM_ALIAS = enum.auto()
555class ORMStatementAdapter(sql_util.ColumnAdapter):
556 """ColumnAdapter which includes a role attribute."""
558 __slots__ = ("role",)
560 def __init__(
561 self,
562 role: _TraceAdaptRole,
563 selectable: Selectable,
564 *,
565 equivalents: Optional[_EquivalentColumnMap] = None,
566 adapt_required: bool = False,
567 allow_label_resolve: bool = True,
568 anonymize_labels: bool = False,
569 adapt_on_names: bool = False,
570 adapt_from_selectables: Optional[AbstractSet[FromClause]] = None,
571 ):
572 self.role = role
573 super().__init__(
574 selectable,
575 equivalents=equivalents,
576 adapt_required=adapt_required,
577 allow_label_resolve=allow_label_resolve,
578 anonymize_labels=anonymize_labels,
579 adapt_on_names=adapt_on_names,
580 adapt_from_selectables=adapt_from_selectables,
581 )
584class ORMAdapter(sql_util.ColumnAdapter):
585 """ColumnAdapter subclass which excludes adaptation of entities from
586 non-matching mappers.
588 """
590 __slots__ = ("role", "mapper", "is_aliased_class", "aliased_insp")
592 is_aliased_class: bool
593 aliased_insp: Optional[AliasedInsp[Any]]
595 def __init__(
596 self,
597 role: _TraceAdaptRole,
598 entity: _InternalEntityType[Any],
599 *,
600 equivalents: Optional[_EquivalentColumnMap] = None,
601 adapt_required: bool = False,
602 allow_label_resolve: bool = True,
603 anonymize_labels: bool = False,
604 selectable: Optional[Selectable] = None,
605 limit_on_entity: bool = True,
606 adapt_on_names: bool = False,
607 adapt_from_selectables: Optional[AbstractSet[FromClause]] = None,
608 ):
609 self.role = role
610 self.mapper = entity.mapper
611 if selectable is None:
612 selectable = entity.selectable
613 if insp_is_aliased_class(entity):
614 self.is_aliased_class = True
615 self.aliased_insp = entity
616 else:
617 self.is_aliased_class = False
618 self.aliased_insp = None
620 super().__init__(
621 selectable,
622 equivalents,
623 adapt_required=adapt_required,
624 allow_label_resolve=allow_label_resolve,
625 anonymize_labels=anonymize_labels,
626 include_fn=self._include_fn if limit_on_entity else None,
627 adapt_on_names=adapt_on_names,
628 adapt_from_selectables=adapt_from_selectables,
629 )
631 def _include_fn(self, elem):
632 entity = elem._annotations.get("parentmapper", None)
634 return not entity or entity.isa(self.mapper) or self.mapper.isa(entity)
637def _is_alias_of_selectable(
638 selectable: FromClause, target: FromClause
639) -> bool:
640 """Return True if ``selectable`` is ``target``, or a plain alias of it,
641 including a flat alias of a join of the same structure and join types."""
643 if selectable is target or (
644 isinstance(selectable, expression.Alias)
645 and selectable.element is target
646 ):
647 return True
649 return all(
650 elem is tgt
651 or (isinstance(elem, expression.Alias) and elem.element is tgt)
652 or (
653 # zip_longest() pads the shorter of the two with None
654 elem is not None
655 and tgt is not None
656 and elem._is_join
657 and tgt._is_join
658 and elem.isouter == tgt.isouter
659 and elem.full == tgt.full
660 )
661 or (
662 isinstance(elem, expression.FromGrouping)
663 and isinstance(tgt, expression.FromGrouping)
664 )
665 for elem, tgt in itertools.zip_longest(
666 sql_util.surface_selectables(selectable),
667 sql_util.surface_selectables(target),
668 )
669 )
672class AliasedClass(
673 inspection.Inspectable["AliasedInsp[_O]"], ORMColumnsClauseRole[_O]
674):
675 r"""Represents an "aliased" form of a mapped class for usage with Query.
677 The ORM equivalent of a :func:`~sqlalchemy.sql.expression.alias`
678 construct, this object mimics the mapped class using a
679 ``__getattr__`` scheme and maintains a reference to a
680 real :class:`~sqlalchemy.sql.expression.Alias` object.
682 A primary purpose of :class:`.AliasedClass` is to serve as an alternate
683 within a SQL statement generated by the ORM, such that an existing
684 mapped entity can be used in multiple contexts. A simple example::
686 # find all pairs of users with the same name
687 user_alias = aliased(User)
688 session.query(User, user_alias).join(
689 (user_alias, User.id > user_alias.id)
690 ).filter(User.name == user_alias.name)
692 :class:`.AliasedClass` is also capable of mapping an existing mapped
693 class to an entirely new selectable, provided this selectable is column-
694 compatible with the existing mapped selectable, and it can also be
695 configured in a mapping as the target of a :func:`_orm.relationship`.
696 See the links below for examples.
698 The :class:`.AliasedClass` object is constructed typically using the
699 :func:`_orm.aliased` function. It also is produced with additional
700 configuration when using the :func:`_orm.with_polymorphic` function.
702 The resulting object is an instance of :class:`.AliasedClass`.
703 This object implements an attribute scheme which produces the
704 same attribute and method interface as the original mapped
705 class, allowing :class:`.AliasedClass` to be compatible
706 with any attribute technique which works on the original class,
707 including hybrid attributes (see :ref:`hybrids_toplevel`).
709 The :class:`.AliasedClass` can be inspected for its underlying
710 :class:`_orm.Mapper`, aliased selectable, and other information
711 using :func:`_sa.inspect`::
713 from sqlalchemy import inspect
715 my_alias = aliased(MyClass)
716 insp = inspect(my_alias)
718 The resulting inspection object is an instance of :class:`.AliasedInsp`.
721 .. seealso::
723 :func:`.aliased`
725 :func:`.with_polymorphic`
727 :ref:`relationship_aliased_class`
729 :ref:`relationship_to_window_function`
732 """
734 __name__: str
736 def __init__(
737 self,
738 mapped_class_or_ac: _EntityType[_O],
739 alias: Optional[FromClause] = None,
740 name: Optional[str] = None,
741 flat: bool = False,
742 adapt_on_names: bool = False,
743 with_polymorphic_mappers: Optional[Sequence[Mapper[Any]]] = None,
744 with_polymorphic_discriminator: Optional[ColumnElement[Any]] = None,
745 base_alias: Optional[AliasedInsp[Any]] = None,
746 use_mapper_path: bool = False,
747 represents_outer_join: bool = False,
748 ):
749 insp = cast(
750 "_InternalEntityType[_O]", inspection.inspect(mapped_class_or_ac)
751 )
752 mapper = insp.mapper
754 nest_adapters = False
756 if alias is None:
757 if insp_is_aliased_class(insp):
758 if insp._is_with_polymorphic:
759 # polymorphic arguments are only passed along with an
760 # explicit selectable, by with_polymorphic() and
761 # AliasedInsp._merge_with()
762 assert (
763 not with_polymorphic_mappers
764 and with_polymorphic_discriminator is None
765 and not represents_outer_join
766 )
768 # aliased() of a with_polymorphic() carries along its
769 # polymorphic configuration, #13584
770 with_polymorphic_mappers = insp.with_polymorphic_mappers
771 with_polymorphic_discriminator = insp.polymorphic_on
772 represents_outer_join = insp.represents_outer_join
774 if not _is_alias_of_selectable(
775 insp.selectable, mapper._with_polymorphic_selectable
776 ):
777 # aliased() of an aliased() that refers to a subquery,
778 # CTE, with_polymorphic() or other selectable; alias
779 # that selectable rather than the mapped table,
780 # #13583, #13584
781 alias = insp.selectable._anonymous_fromclause(
782 name=name, flat=flat
783 )
784 adapt_on_names = adapt_on_names or insp._adapt_on_names
785 elif insp.selectable._is_subquery:
786 alias = insp.selectable.alias()
788 if alias is None:
789 alias = (
790 mapper._with_polymorphic_selectable._anonymous_fromclause(
791 name=name,
792 flat=flat,
793 )
794 )
795 elif insp.is_aliased_class:
796 nest_adapters = True
798 assert alias is not None
799 self._aliased_insp = AliasedInsp(
800 self,
801 insp,
802 alias,
803 name,
804 (
805 with_polymorphic_mappers
806 if with_polymorphic_mappers
807 else mapper.with_polymorphic_mappers
808 ),
809 (
810 with_polymorphic_discriminator
811 if with_polymorphic_discriminator is not None
812 else mapper.polymorphic_on
813 ),
814 base_alias,
815 use_mapper_path,
816 adapt_on_names,
817 represents_outer_join,
818 nest_adapters,
819 )
821 self.__name__ = f"aliased({mapper.class_.__name__})"
823 @classmethod
824 def _reconstitute_from_aliased_insp(
825 cls, aliased_insp: AliasedInsp[_O]
826 ) -> AliasedClass[_O]:
827 obj = cls.__new__(cls)
828 obj.__name__ = f"aliased({aliased_insp.mapper.class_.__name__})"
829 obj._aliased_insp = aliased_insp
831 if aliased_insp._is_with_polymorphic:
832 for sub_aliased_insp in aliased_insp._with_polymorphic_entities:
833 if sub_aliased_insp is not aliased_insp:
834 ent = AliasedClass._reconstitute_from_aliased_insp(
835 sub_aliased_insp
836 )
837 setattr(obj, sub_aliased_insp.class_.__name__, ent)
839 return obj
841 def __getattr__(self, key: str) -> Any:
842 try:
843 _aliased_insp = self.__dict__["_aliased_insp"]
844 except KeyError:
845 raise AttributeError()
846 else:
847 target = _aliased_insp._target
848 # maintain all getattr mechanics
849 attr = getattr(target, key)
851 # attribute is a method, that will be invoked against a
852 # "self"; so just return a new method with the same function and
853 # new self
854 if hasattr(attr, "__call__") and hasattr(attr, "__self__"):
855 return types.MethodType(attr.__func__, self)
857 # attribute is a descriptor, that will be invoked against a
858 # "self"; so invoke the descriptor against this self
859 if hasattr(attr, "__get__"):
860 attr = attr.__get__(None, self)
862 # attributes within the QueryableAttribute system will want this
863 # to be invoked so the object can be adapted
864 if hasattr(attr, "adapt_to_entity"):
865 attr = attr.adapt_to_entity(_aliased_insp)
866 setattr(self, key, attr)
868 return attr
870 def _get_from_serialized(
871 self, key: str, mapped_class: _O, aliased_insp: AliasedInsp[_O]
872 ) -> Any:
873 # this method is only used in terms of the
874 # sqlalchemy.ext.serializer extension
875 attr = getattr(mapped_class, key)
876 if hasattr(attr, "__call__") and hasattr(attr, "__self__"):
877 return types.MethodType(attr.__func__, self)
879 # attribute is a descriptor, that will be invoked against a
880 # "self"; so invoke the descriptor against this self
881 if hasattr(attr, "__get__"):
882 attr = attr.__get__(None, self)
884 # attributes within the QueryableAttribute system will want this
885 # to be invoked so the object can be adapted
886 if hasattr(attr, "adapt_to_entity"):
887 aliased_insp._weak_entity = weakref.ref(self)
888 attr = attr.adapt_to_entity(aliased_insp)
889 setattr(self, key, attr)
891 return attr
893 def __repr__(self) -> str:
894 return "<AliasedClass at 0x%x; %s>" % (
895 id(self),
896 self._aliased_insp._target.__name__,
897 )
899 def __str__(self) -> str:
900 return str(self._aliased_insp)
903@inspection._self_inspects
904class AliasedInsp(
905 ORMEntityColumnsClauseRole[_O],
906 ORMFromClauseRole,
907 HasCacheKey,
908 InspectionAttr,
909 MemoizedSlots,
910 inspection.Inspectable["AliasedInsp[_O]"],
911 Generic[_O],
912):
913 """Provide an inspection interface for an
914 :class:`.AliasedClass` object.
916 The :class:`.AliasedInsp` object is returned
917 given an :class:`.AliasedClass` using the
918 :func:`_sa.inspect` function::
920 from sqlalchemy import inspect
921 from sqlalchemy.orm import aliased
923 my_alias = aliased(MyMappedClass)
924 insp = inspect(my_alias)
926 Attributes on :class:`.AliasedInsp`
927 include:
929 * ``entity`` - the :class:`.AliasedClass` represented.
930 * ``mapper`` - the :class:`_orm.Mapper` mapping the underlying class.
931 * ``selectable`` - the :class:`_expression.Alias`
932 construct which ultimately
933 represents an aliased :class:`_schema.Table` or
934 :class:`_expression.Select`
935 construct.
936 * ``name`` - the name of the alias. Also is used as the attribute
937 name when returned in a result tuple from :class:`_query.Query`.
938 * ``with_polymorphic_mappers`` - collection of :class:`_orm.Mapper`
939 objects
940 indicating all those mappers expressed in the select construct
941 for the :class:`.AliasedClass`.
942 * ``polymorphic_on`` - an alternate column or SQL expression which
943 will be used as the "discriminator" for a polymorphic load.
945 .. seealso::
947 :ref:`inspection_toplevel`
949 """
951 __slots__ = (
952 "__weakref__",
953 "_weak_entity",
954 "mapper",
955 "selectable",
956 "name",
957 "_adapt_on_names",
958 "with_polymorphic_mappers",
959 "polymorphic_on",
960 "_use_mapper_path",
961 "_base_alias",
962 "represents_outer_join",
963 "persist_selectable",
964 "local_table",
965 "_is_with_polymorphic",
966 "_with_polymorphic_entities",
967 "_adapter",
968 "_target",
969 "__clause_element__",
970 "_memoized_values",
971 "_all_column_expressions",
972 "_nest_adapters",
973 )
975 _cache_key_traversal = [
976 ("name", visitors.ExtendedInternalTraversal.dp_string),
977 ("_adapt_on_names", visitors.ExtendedInternalTraversal.dp_boolean),
978 ("_use_mapper_path", visitors.ExtendedInternalTraversal.dp_boolean),
979 ("_target", visitors.ExtendedInternalTraversal.dp_inspectable),
980 ("selectable", visitors.ExtendedInternalTraversal.dp_clauseelement),
981 (
982 "with_polymorphic_mappers",
983 visitors.InternalTraversal.dp_has_cache_key_list,
984 ),
985 ("polymorphic_on", visitors.InternalTraversal.dp_clauseelement),
986 ]
988 mapper: Mapper[_O]
989 selectable: FromClause
990 _adapter: ORMAdapter
991 with_polymorphic_mappers: Sequence[Mapper[Any]]
992 _with_polymorphic_entities: Sequence[AliasedInsp[Any]]
993 _adapt_on_names: bool
995 _weak_entity: weakref.ref[AliasedClass[_O]]
996 """the AliasedClass that refers to this AliasedInsp"""
998 _target: Union[Type[_O], AliasedClass[_O]]
999 """the thing referenced by the AliasedClass/AliasedInsp.
1001 In the vast majority of cases, this is the mapped class. However
1002 it may also be another AliasedClass (alias of alias).
1004 """
1006 def __init__(
1007 self,
1008 entity: AliasedClass[_O],
1009 inspected: _InternalEntityType[_O],
1010 selectable: FromClause,
1011 name: Optional[str],
1012 with_polymorphic_mappers: Optional[Sequence[Mapper[Any]]],
1013 polymorphic_on: Optional[ColumnElement[Any]],
1014 _base_alias: Optional[AliasedInsp[Any]],
1015 _use_mapper_path: bool,
1016 adapt_on_names: bool,
1017 represents_outer_join: bool,
1018 nest_adapters: bool,
1019 ):
1020 mapped_class_or_ac = inspected.entity
1021 mapper = inspected.mapper
1023 self._weak_entity = weakref.ref(entity)
1024 self.mapper = mapper
1025 self.selectable = self.persist_selectable = self.local_table = (
1026 selectable
1027 )
1028 self.name = name
1029 self.polymorphic_on = polymorphic_on
1030 self._base_alias = weakref.ref(_base_alias or self)
1031 self._use_mapper_path = _use_mapper_path
1032 self.represents_outer_join = represents_outer_join
1033 self._nest_adapters = nest_adapters
1035 if with_polymorphic_mappers:
1036 self._is_with_polymorphic = True
1037 self.with_polymorphic_mappers = with_polymorphic_mappers
1038 self._with_polymorphic_entities = []
1039 for poly in self.with_polymorphic_mappers:
1040 if poly is not mapper:
1041 ent = AliasedClass(
1042 poly.class_,
1043 selectable,
1044 base_alias=self,
1045 adapt_on_names=adapt_on_names,
1046 use_mapper_path=_use_mapper_path,
1047 )
1049 setattr(self.entity, poly.class_.__name__, ent)
1050 self._with_polymorphic_entities.append(ent._aliased_insp)
1052 else:
1053 self._is_with_polymorphic = False
1054 self.with_polymorphic_mappers = [mapper]
1056 self._adapter = ORMAdapter(
1057 _TraceAdaptRole.ALIASED_INSP,
1058 mapper,
1059 selectable=selectable,
1060 equivalents=mapper._equivalent_columns,
1061 adapt_on_names=adapt_on_names,
1062 anonymize_labels=True,
1063 # make sure the adapter doesn't try to grab other tables that
1064 # are not even the thing we are mapping, such as embedded
1065 # selectables in subqueries or CTEs. See issue #6060
1066 adapt_from_selectables={
1067 m.selectable
1068 for m in self.with_polymorphic_mappers
1069 if not adapt_on_names
1070 },
1071 limit_on_entity=False,
1072 )
1074 if nest_adapters:
1075 # supports "aliased class of aliased class" use case
1076 assert isinstance(inspected, AliasedInsp)
1077 self._adapter = inspected._adapter.wrap(self._adapter)
1079 self._adapt_on_names = adapt_on_names
1080 self._target = mapped_class_or_ac
1082 @property
1083 def _post_inspect(self): # type: ignore[override]
1084 self.mapper._check_configure()
1086 @classmethod
1087 def _alias_factory(
1088 cls,
1089 element: Union[_EntityType[_O], FromClause],
1090 alias: Optional[FromClause] = None,
1091 name: Optional[str] = None,
1092 flat: bool = False,
1093 adapt_on_names: bool = False,
1094 ) -> Union[AliasedClass[_O], FromClause]:
1095 if isinstance(element, GenerativeSelect):
1096 return coercions.expect(roles.FromClauseRole, element, flat=flat)
1097 elif isinstance(element, FromClause):
1098 if adapt_on_names:
1099 raise sa_exc.ArgumentError(
1100 "adapt_on_names only applies to ORM elements"
1101 )
1102 if name:
1103 return element.alias(name=name, flat=flat)
1104 else:
1105 # see selectable.py->Alias._factory() for similar
1106 # mypy issue. Cannot get the overload to see this
1107 # in mypy (works fine in pyright)
1108 return coercions.expect( # type: ignore[no-any-return]
1109 roles.AnonymizedFromClauseRole, element, flat=flat
1110 )
1111 else:
1112 return AliasedClass(
1113 element,
1114 alias=alias,
1115 flat=flat,
1116 name=name,
1117 adapt_on_names=adapt_on_names,
1118 )
1120 @classmethod
1121 def _with_polymorphic_factory(
1122 cls,
1123 base: Union[Type[_O], Mapper[_O]],
1124 classes: Union[Literal["*"], Iterable[_EntityType[Any]]],
1125 selectable: Union[Literal[False, None], FromClause] = False,
1126 flat: bool = False,
1127 polymorphic_on: Optional[ColumnElement[Any]] = None,
1128 aliased: bool = False,
1129 innerjoin: bool = False,
1130 adapt_on_names: bool = False,
1131 name: Optional[str] = None,
1132 _use_mapper_path: bool = False,
1133 ) -> AliasedClass[_O]:
1134 primary_mapper = _class_to_mapper(base)
1136 if selectable not in (None, False) and flat:
1137 raise sa_exc.ArgumentError(
1138 "the 'flat' and 'selectable' arguments cannot be passed "
1139 "simultaneously to with_polymorphic()"
1140 )
1142 mappers, selectable = primary_mapper._with_polymorphic_args(
1143 classes, selectable, innerjoin=innerjoin
1144 )
1145 if aliased or flat:
1146 assert selectable is not None
1147 selectable = selectable._anonymous_fromclause(flat=flat)
1149 return AliasedClass(
1150 base,
1151 selectable,
1152 name=name,
1153 with_polymorphic_mappers=mappers,
1154 adapt_on_names=adapt_on_names,
1155 with_polymorphic_discriminator=polymorphic_on,
1156 use_mapper_path=_use_mapper_path,
1157 represents_outer_join=not innerjoin,
1158 )
1160 @property
1161 def entity(self) -> AliasedClass[_O]:
1162 # to eliminate reference cycles, the AliasedClass is held weakly.
1163 # this produces some situations where the AliasedClass gets lost,
1164 # particularly when one is created internally and only the AliasedInsp
1165 # is passed around.
1166 # to work around this case, we just generate a new one when we need
1167 # it, as it is a simple class with very little initial state on it.
1168 ent = self._weak_entity()
1169 if ent is None:
1170 ent = AliasedClass._reconstitute_from_aliased_insp(self)
1171 self._weak_entity = weakref.ref(ent)
1172 return ent
1174 is_aliased_class = True
1175 "always returns True"
1177 def _memoized_method___clause_element__(self) -> FromClause:
1178 return self.selectable._annotate(
1179 {
1180 "parentmapper": self.mapper,
1181 "parententity": self,
1182 "entity_namespace": self,
1183 }
1184 )._set_propagate_attrs(
1185 {"compile_state_plugin": "orm", "plugin_subject": self}
1186 )
1188 @property
1189 def entity_namespace(self) -> AliasedClass[_O]:
1190 return self.entity
1192 @property
1193 def class_(self) -> Type[_O]:
1194 """Return the mapped class ultimately represented by this
1195 :class:`.AliasedInsp`."""
1196 return self.mapper.class_
1198 @property
1199 def _path_registry(self) -> _AbstractEntityRegistry:
1200 if self._use_mapper_path:
1201 return self.mapper._path_registry
1202 else:
1203 return PathRegistry.per_mapper(self)
1205 def __getstate__(self) -> Dict[str, Any]:
1206 return {
1207 "entity": self.entity,
1208 "mapper": self.mapper,
1209 "alias": self.selectable,
1210 "name": self.name,
1211 "adapt_on_names": self._adapt_on_names,
1212 "with_polymorphic_mappers": self.with_polymorphic_mappers,
1213 "with_polymorphic_discriminator": self.polymorphic_on,
1214 "base_alias": self._base_alias(),
1215 "use_mapper_path": self._use_mapper_path,
1216 "represents_outer_join": self.represents_outer_join,
1217 "nest_adapters": self._nest_adapters,
1218 }
1220 def __setstate__(self, state: Dict[str, Any]) -> None:
1221 self.__init__( # type: ignore[misc]
1222 state["entity"],
1223 state["mapper"],
1224 state["alias"],
1225 state["name"],
1226 state["with_polymorphic_mappers"],
1227 state["with_polymorphic_discriminator"],
1228 state["base_alias"],
1229 state["use_mapper_path"],
1230 state["adapt_on_names"],
1231 state["represents_outer_join"],
1232 state["nest_adapters"],
1233 )
1235 def _merge_with(self, other: AliasedInsp[_O]) -> AliasedInsp[_O]:
1236 # assert self._is_with_polymorphic
1237 # assert other._is_with_polymorphic
1239 primary_mapper = other.mapper
1241 assert self.mapper is primary_mapper
1243 our_classes = util.to_set(
1244 mp.class_ for mp in self.with_polymorphic_mappers
1245 )
1246 new_classes = {mp.class_ for mp in other.with_polymorphic_mappers}
1247 if our_classes == new_classes:
1248 return other
1249 else:
1250 classes = our_classes.union(new_classes)
1252 mappers, selectable = primary_mapper._with_polymorphic_args(
1253 classes, None, innerjoin=not other.represents_outer_join
1254 )
1255 selectable = selectable._anonymous_fromclause(flat=True)
1256 return AliasedClass(
1257 primary_mapper,
1258 selectable,
1259 with_polymorphic_mappers=mappers,
1260 with_polymorphic_discriminator=other.polymorphic_on,
1261 use_mapper_path=other._use_mapper_path,
1262 represents_outer_join=other.represents_outer_join,
1263 )._aliased_insp
1265 def _adapt_element(
1266 self, expr: _ORMCOLEXPR, key: Optional[str] = None
1267 ) -> _ORMCOLEXPR:
1268 assert isinstance(expr, ColumnElement)
1269 d: Dict[str, Any] = {
1270 "parententity": self,
1271 "parentmapper": self.mapper,
1272 }
1273 if key:
1274 d["proxy_key"] = key
1276 # userspace adapt of an attribute from AliasedClass; validate that
1277 # it actually was present
1278 adapted = self._adapter.adapt_check_present(expr)
1279 if adapted is None:
1280 adapted = expr
1281 if self._adapter.adapt_on_names:
1282 util.warn_limited(
1283 "Did not locate an expression in selectable for "
1284 "attribute %r; ensure name is correct in expression",
1285 (key,),
1286 )
1287 else:
1288 util.warn_limited(
1289 "Did not locate an expression in selectable for "
1290 "attribute %r; to match by name, use the "
1291 "adapt_on_names parameter",
1292 (key,),
1293 )
1295 return adapted._annotate(d)._set_propagate_attrs(
1296 {"compile_state_plugin": "orm", "plugin_subject": self}
1297 )
1299 if TYPE_CHECKING:
1300 # establish compatibility with the _ORMAdapterProto protocol,
1301 # which in turn is compatible with _CoreAdapterProto.
1303 def _orm_adapt_element(
1304 self,
1305 obj: _CE,
1306 key: Optional[str] = None,
1307 ) -> _CE: ...
1309 else:
1310 _orm_adapt_element = _adapt_element
1312 def _entity_for_mapper(self, mapper):
1313 self_poly = self.with_polymorphic_mappers
1314 if mapper in self_poly:
1315 if mapper is self.mapper:
1316 return self
1317 else:
1318 return getattr(
1319 self.entity, mapper.class_.__name__
1320 )._aliased_insp
1321 elif mapper.isa(self.mapper):
1322 return self
1323 else:
1324 assert False, "mapper %s doesn't correspond to %s" % (mapper, self)
1326 def _memoized_attr__get_clause(self):
1327 onclause, replacemap = self.mapper._get_clause
1328 return (
1329 self._adapter.traverse(onclause),
1330 {
1331 self._adapter.traverse(col): param
1332 for col, param in replacemap.items()
1333 },
1334 )
1336 def _memoized_attr__memoized_values(self):
1337 return {}
1339 def _memoized_attr__all_column_expressions(self):
1340 if self._is_with_polymorphic:
1341 cols_plus_keys = self.mapper._columns_plus_keys(
1342 [ent.mapper for ent in self._with_polymorphic_entities]
1343 )
1344 else:
1345 cols_plus_keys = self.mapper._columns_plus_keys()
1347 cols_plus_keys = [
1348 (key, self._adapt_element(col)) for key, col in cols_plus_keys
1349 ]
1351 return WriteableColumnCollection(cols_plus_keys)
1353 def _memo(self, key, callable_, *args, **kw):
1354 if key in self._memoized_values:
1355 return self._memoized_values[key]
1356 else:
1357 self._memoized_values[key] = value = callable_(*args, **kw)
1358 return value
1360 def __repr__(self):
1361 if self.with_polymorphic_mappers:
1362 with_poly = "(%s)" % ", ".join(
1363 mp.class_.__name__ for mp in self.with_polymorphic_mappers
1364 )
1365 else:
1366 with_poly = ""
1367 return "<AliasedInsp at 0x%x; %s%s>" % (
1368 id(self),
1369 self.class_.__name__,
1370 with_poly,
1371 )
1373 def __str__(self):
1374 return self.path_string()
1376 def path_string(self) -> str:
1377 """Return a user-facing name for this :class:`.AliasedInsp`,
1378 for use in a :class:`_orm.PathRegistry` string representation.
1380 """
1381 if self._is_with_polymorphic:
1382 return "with_polymorphic(%s, [%s])" % (
1383 self._target.__name__,
1384 ", ".join(
1385 mp.class_.__name__
1386 for mp in self.with_polymorphic_mappers
1387 if mp is not self.mapper
1388 ),
1389 )
1390 else:
1391 return "aliased(%s)" % (self._target.__name__,)
1394class _WrapUserEntity:
1395 """A wrapper used within the loader_criteria lambda caller so that
1396 we can bypass declared_attr descriptors on unmapped mixins, which
1397 normally emit a warning for such use.
1399 might also be useful for other per-lambda instrumentations should
1400 the need arise.
1402 """
1404 __slots__ = ("subject",)
1406 def __init__(self, subject):
1407 self.subject = subject
1409 @util.preload_module("sqlalchemy.orm.decl_api")
1410 def __getattribute__(self, name):
1411 decl_api = util.preloaded.orm.decl_api
1413 subject = object.__getattribute__(self, "subject")
1414 if name in subject.__dict__ and isinstance(
1415 subject.__dict__[name], decl_api.declared_attr
1416 ):
1417 return subject.__dict__[name].fget(subject)
1418 else:
1419 return getattr(subject, name)
1422class LoaderCriteriaOption(CriteriaOption):
1423 """Add additional WHERE criteria to the load for all occurrences of
1424 a particular entity.
1426 :class:`_orm.LoaderCriteriaOption` is invoked using the
1427 :func:`_orm.with_loader_criteria` function; see that function for
1428 details.
1430 .. versionadded:: 1.4
1432 """
1434 __slots__ = (
1435 "root_entity",
1436 "entity",
1437 "deferred_where_criteria",
1438 "where_criteria",
1439 "_where_crit_orig",
1440 "include_aliases",
1441 "propagate_to_loaders",
1442 )
1444 _traverse_internals = [
1445 ("root_entity", visitors.ExtendedInternalTraversal.dp_plain_obj),
1446 ("entity", visitors.ExtendedInternalTraversal.dp_has_cache_key),
1447 ("where_criteria", visitors.InternalTraversal.dp_clauseelement),
1448 ("include_aliases", visitors.InternalTraversal.dp_boolean),
1449 ("propagate_to_loaders", visitors.InternalTraversal.dp_boolean),
1450 ]
1452 root_entity: Optional[Type[Any]]
1453 entity: Optional[_InternalEntityType[Any]]
1454 where_criteria: Union[ColumnElement[bool], lambdas.DeferredLambdaElement]
1455 deferred_where_criteria: bool
1456 include_aliases: bool
1457 propagate_to_loaders: bool
1459 _where_crit_orig: Any
1461 def __init__(
1462 self,
1463 entity_or_base: _EntityType[Any],
1464 where_criteria: Union[
1465 _ColumnExpressionArgument[bool],
1466 Callable[[Any], _ColumnExpressionArgument[bool]],
1467 ],
1468 loader_only: bool = False,
1469 include_aliases: bool = False,
1470 propagate_to_loaders: bool = True,
1471 track_closure_variables: bool = True,
1472 ):
1473 entity = cast(
1474 "_InternalEntityType[Any]",
1475 inspection.inspect(entity_or_base, False),
1476 )
1477 if entity is None:
1478 self.root_entity = cast("Type[Any]", entity_or_base)
1479 self.entity = None
1480 else:
1481 self.root_entity = None
1482 self.entity = entity
1484 self._where_crit_orig = where_criteria
1485 if callable(where_criteria):
1486 if self.root_entity is not None:
1487 wrap_entity = self.root_entity
1488 else:
1489 assert entity is not None
1490 wrap_entity = entity.entity
1492 self.deferred_where_criteria = True
1493 self.where_criteria = lambdas.DeferredLambdaElement(
1494 where_criteria,
1495 roles.WhereHavingRole,
1496 lambda_args=(_WrapUserEntity(wrap_entity),),
1497 opts=lambdas.LambdaOptions(
1498 track_closure_variables=track_closure_variables
1499 ),
1500 )
1501 else:
1502 self.deferred_where_criteria = False
1503 self.where_criteria = coercions.expect(
1504 roles.WhereHavingRole, where_criteria
1505 )
1507 self.include_aliases = include_aliases
1508 self.propagate_to_loaders = propagate_to_loaders
1510 @classmethod
1511 def _unreduce(
1512 cls, entity, where_criteria, include_aliases, propagate_to_loaders
1513 ):
1514 return LoaderCriteriaOption(
1515 entity,
1516 where_criteria,
1517 include_aliases=include_aliases,
1518 propagate_to_loaders=propagate_to_loaders,
1519 )
1521 def __reduce__(self):
1522 return (
1523 LoaderCriteriaOption._unreduce,
1524 (
1525 self.entity.class_ if self.entity else self.root_entity,
1526 self._where_crit_orig,
1527 self.include_aliases,
1528 self.propagate_to_loaders,
1529 ),
1530 )
1532 def _all_mappers(self) -> Iterator[Mapper[Any]]:
1533 if self.entity:
1534 yield from self.entity.mapper.self_and_descendants
1535 else:
1536 assert self.root_entity
1537 stack = list(self.root_entity.__subclasses__())
1538 while stack:
1539 subclass = stack.pop(0)
1540 ent = cast(
1541 "_InternalEntityType[Any]",
1542 inspection.inspect(subclass, raiseerr=False),
1543 )
1544 if ent:
1545 yield from ent.mapper.self_and_descendants
1546 else:
1547 stack.extend(subclass.__subclasses__())
1549 def _should_include(self, compile_state: _ORMCompileState) -> bool:
1550 if (
1551 compile_state.select_statement._annotations.get(
1552 "for_loader_criteria", None
1553 )
1554 is self
1555 ):
1556 return False
1557 return True
1559 def _resolve_where_criteria(
1560 self, ext_info: _InternalEntityType[Any]
1561 ) -> ColumnElement[bool]:
1562 if self.deferred_where_criteria:
1563 crit = cast(
1564 "ColumnElement[bool]",
1565 self.where_criteria._resolve_with_args(ext_info.entity),
1566 )
1567 else:
1568 crit = self.where_criteria # type: ignore[assignment]
1569 assert isinstance(crit, ColumnElement)
1570 return sql_util._deep_annotate(
1571 crit,
1572 {"for_loader_criteria": self},
1573 detect_subquery_cols=True,
1574 ind_cols_on_fromclause=True,
1575 )
1577 def process_compile_state_replaced_entities(
1578 self,
1579 compile_state: _ORMCompileState,
1580 mapper_entities: Iterable[_MapperEntity],
1581 ) -> None:
1582 self.process_compile_state(compile_state)
1584 def process_compile_state(self, compile_state: _ORMCompileState) -> None:
1585 """Apply a modification to a given :class:`.CompileState`."""
1587 # if options to limit the criteria to immediate query only,
1588 # use compile_state.attributes instead
1590 self.get_global_criteria(compile_state.global_attributes)
1592 def get_global_criteria(self, attributes: Dict[Any, Any]) -> None:
1593 for mp in self._all_mappers():
1594 load_criteria = attributes.setdefault(
1595 ("additional_entity_criteria", mp), []
1596 )
1598 load_criteria.append(self)
1601inspection._inspects(AliasedClass)(lambda target: target._aliased_insp)
1604@inspection._inspects(type)
1605def _inspect_mc(
1606 class_: Type[_O],
1607) -> Optional[Mapper[_O]]:
1608 try:
1609 class_manager = opt_manager_of_class(class_)
1610 if class_manager is None or not class_manager.is_mapped:
1611 return None
1612 mapper = class_manager.mapper
1613 except orm_exc.NO_STATE:
1614 return None
1615 else:
1616 return mapper
1619GenericAlias = type(List[Any])
1622@inspection._inspects(GenericAlias)
1623def _inspect_generic_alias(
1624 class_: Type[_O],
1625) -> Optional[Mapper[_O]]:
1626 origin = cast("Type[_O]", get_origin(class_))
1627 return _inspect_mc(origin)
1630@inspection._self_inspects
1631class Bundle(
1632 ORMColumnsClauseRole[_T],
1633 SupportsCloneAnnotations,
1634 MemoizedHasCacheKey,
1635 inspection.Inspectable["Bundle[_T]"],
1636 InspectionAttr,
1637):
1638 """A grouping of SQL expressions that are returned by a :class:`.Query`
1639 under one namespace.
1641 The :class:`.Bundle` essentially allows nesting of the tuple-based
1642 results returned by a column-oriented :class:`_query.Query` object.
1643 It also
1644 is extensible via simple subclassing, where the primary capability
1645 to override is that of how the set of expressions should be returned,
1646 allowing post-processing as well as custom return types, without
1647 involving ORM identity-mapped classes.
1649 .. seealso::
1651 :ref:`bundles`
1653 :class:`.DictBundle`
1655 """
1657 single_entity = False
1658 """If True, queries for a single Bundle will be returned as a single
1659 entity, rather than an element within a keyed tuple."""
1661 is_clause_element = False
1663 is_mapper = False
1665 is_aliased_class = False
1667 is_bundle = True
1669 _propagate_attrs: _PropagateAttrsType = util.immutabledict()
1671 proxy_set = util.EMPTY_SET
1673 exprs: List[_ColumnsClauseElement]
1675 def __init__(
1676 self, name: str, *exprs: _ColumnExpressionArgument[Any], **kw: Any
1677 ) -> None:
1678 r"""Construct a new :class:`.Bundle`.
1680 e.g.::
1682 bn = Bundle("mybundle", MyClass.x, MyClass.y)
1684 for row in session.query(bn).filter(bn.c.x == 5).filter(bn.c.y == 4):
1685 print(row.mybundle.x, row.mybundle.y)
1687 :param name: name of the bundle.
1688 :param \*exprs: columns or SQL expressions comprising the bundle.
1689 :param single_entity=False: if True, rows for this :class:`.Bundle`
1690 can be returned as a "single entity" outside of any enclosing tuple
1691 in the same manner as a mapped entity.
1693 """ # noqa: E501
1694 self.name = self._label = name
1695 coerced_exprs = [
1696 coercions.expect(
1697 roles.ColumnsClauseRole, expr, apply_propagate_attrs=self
1698 )
1699 for expr in exprs
1700 ]
1701 self.exprs = coerced_exprs
1703 self.c = self.columns = WriteableColumnCollection(
1704 (getattr(col, "key", col._label), col)
1705 for col in [e._annotations.get("bundle", e) for e in coerced_exprs]
1706 ).as_readonly()
1707 self.single_entity = kw.pop("single_entity", self.single_entity)
1709 def _gen_cache_key(
1710 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
1711 ) -> Tuple[Any, ...]:
1712 return (self.__class__, self.name, self.single_entity) + tuple(
1713 [expr._gen_cache_key(anon_map, bindparams) for expr in self.exprs]
1714 )
1716 @property
1717 def mapper(self) -> Optional[Mapper[Any]]:
1718 mp: Optional[Mapper[Any]] = self.exprs[0]._annotations.get(
1719 "parentmapper", None
1720 )
1721 return mp
1723 @property
1724 def entity(self) -> Optional[_InternalEntityType[Any]]:
1725 ie: Optional[_InternalEntityType[Any]] = self.exprs[
1726 0
1727 ]._annotations.get("parententity", None)
1728 return ie
1730 @property
1731 def entity_namespace(
1732 self,
1733 ) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]:
1734 return self.c
1736 columns: ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]
1738 """A namespace of SQL expressions referred to by this :class:`.Bundle`.
1740 e.g.::
1742 bn = Bundle("mybundle", MyClass.x, MyClass.y)
1744 q = sess.query(bn).filter(bn.c.x == 5)
1746 Nesting of bundles is also supported::
1748 b1 = Bundle(
1749 "b1",
1750 Bundle("b2", MyClass.a, MyClass.b),
1751 Bundle("b3", MyClass.x, MyClass.y),
1752 )
1754 q = sess.query(b1).filter(b1.c.b2.c.a == 5).filter(b1.c.b3.c.y == 9)
1756 .. seealso::
1758 :attr:`.Bundle.c`
1760 """ # noqa: E501
1762 c: ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]
1763 """An alias for :attr:`.Bundle.columns`."""
1765 def _clone(self, **kw):
1766 cloned = self.__class__.__new__(self.__class__)
1767 cloned.__dict__.update(self.__dict__)
1768 return cloned
1770 def __clause_element__(self):
1771 # ensure existing entity_namespace remains
1772 annotations = {"bundle": self, "entity_namespace": self}
1773 annotations.update(self._annotations)
1775 plugin_subject = self.exprs[0]._propagate_attrs.get(
1776 "plugin_subject", self.entity
1777 )
1778 return (
1779 expression.ClauseList(
1780 _literal_as_text_role=roles.ColumnsClauseRole,
1781 group=False,
1782 *[e._annotations.get("bundle", e) for e in self.exprs],
1783 )
1784 ._annotate(annotations)
1785 ._set_propagate_attrs(
1786 # the Bundle *must* use the orm plugin no matter what. the
1787 # subject can be None but it's much better if it's not.
1788 {
1789 "compile_state_plugin": "orm",
1790 "plugin_subject": plugin_subject,
1791 }
1792 )
1793 )
1795 @property
1796 def clauses(self):
1797 return self.__clause_element__().clauses
1799 def label(self, name):
1800 """Provide a copy of this :class:`.Bundle` passing a new label."""
1802 cloned = self._clone()
1803 cloned.name = name
1804 return cloned
1806 def create_row_processor(
1807 self,
1808 query: Select[Unpack[TupleAny]],
1809 procs: Sequence[Callable[[Row[Unpack[TupleAny]]], Any]],
1810 labels: Sequence[str],
1811 ) -> Callable[[Row[Unpack[TupleAny]]], Any]:
1812 """Produce the "row processing" function for this :class:`.Bundle`.
1814 May be overridden by subclasses to provide custom behaviors when
1815 results are fetched. The method is passed the statement object and a
1816 set of "row processor" functions at query execution time; these
1817 processor functions when given a result row will return the individual
1818 attribute value, which can then be adapted into any kind of return data
1819 structure.
1821 The example below illustrates replacing the usual :class:`.Row`
1822 return structure with a straight Python dictionary::
1824 from sqlalchemy.orm import Bundle
1827 class DictBundle(Bundle):
1828 def create_row_processor(self, query, procs, labels):
1829 "Override create_row_processor to return values as dictionaries"
1831 def proc(row):
1832 return dict(zip(labels, (proc(row) for proc in procs)))
1834 return proc
1836 A result from the above :class:`_orm.Bundle` will return dictionary
1837 values::
1839 bn = DictBundle("mybundle", MyClass.data1, MyClass.data2)
1840 for row in session.execute(select(bn)).where(bn.c.data1 == "d1"):
1841 print(row.mybundle["data1"], row.mybundle["data2"])
1843 The above example is available natively using :class:`.DictBundle`
1845 .. seealso::
1847 :class:`.DictBundle`
1849 """ # noqa: E501
1850 keyed_tuple = result_tuple(labels, [() for l in labels])
1852 def proc(row: Row[Unpack[TupleAny]]) -> Any:
1853 return keyed_tuple([proc(row) for proc in procs])
1855 return proc
1858class DictBundle(Bundle[_T]):
1859 """Like :class:`.Bundle` but returns ``dict`` instances instead of
1860 named tuple like objects::
1862 bn = DictBundle("mybundle", MyClass.data1, MyClass.data2)
1863 for row in session.execute(select(bn)).where(bn.c.data1 == "d1"):
1864 print(row.mybundle["data1"], row.mybundle["data2"])
1866 Differently from :class:`.Bundle`, multiple columns with the same name are
1867 not supported.
1869 .. versionadded:: 2.1
1871 .. seealso::
1873 :ref:`bundles`
1875 :class:`.Bundle`
1876 """
1878 def __init__(
1879 self, name: str, *exprs: _ColumnExpressionArgument[Any], **kw: Any
1880 ) -> None:
1881 super().__init__(name, *exprs, **kw)
1882 if len(set(self.c.keys())) != len(self.c):
1883 raise sa_exc.ArgumentError(
1884 "DictBundle does not support duplicate column names"
1885 )
1887 def create_row_processor(
1888 self,
1889 query: Select[Unpack[TupleAny]],
1890 procs: Sequence[Callable[[Row[Unpack[TupleAny]]], Any]],
1891 labels: Sequence[str],
1892 ) -> Callable[[Row[Unpack[TupleAny]]], dict[str, Any]]:
1893 def proc(row: Row[Unpack[TupleAny]]) -> dict[str, Any]:
1894 return dict(zip(labels, (proc(row) for proc in procs)))
1896 return proc
1899def _orm_full_deannotate(element: _SA) -> _SA:
1900 return sql_util._deep_deannotate(element)
1903class _ORMJoin(expression.Join):
1904 """Extend Join to support ORM constructs as input."""
1906 __visit_name__ = expression.Join.__visit_name__
1908 inherit_cache = True
1910 def __init__(
1911 self,
1912 left: _FromClauseArgument,
1913 right: _FromClauseArgument,
1914 onclause: Optional[_OnClauseArgument] = None,
1915 isouter: bool = False,
1916 full: bool = False,
1917 _left_memo: Optional[Any] = None,
1918 _right_memo: Optional[Any] = None,
1919 _extra_criteria: Tuple[ColumnElement[bool], ...] = (),
1920 ):
1921 left_info = cast(
1922 "Union[FromClause, _InternalEntityType[Any]]",
1923 inspection.inspect(left),
1924 )
1926 right_info = cast(
1927 "Union[FromClause, _InternalEntityType[Any]]",
1928 inspection.inspect(right),
1929 )
1930 adapt_to = right_info.selectable
1932 # used by joined eager loader
1933 self._left_memo = _left_memo
1934 self._right_memo = _right_memo
1936 if isinstance(onclause, attributes.QueryableAttribute):
1937 if TYPE_CHECKING:
1938 assert isinstance(
1939 onclause.comparator, RelationshipProperty.Comparator
1940 )
1941 on_selectable = onclause.comparator._source_selectable()
1942 prop = onclause.property
1943 _extra_criteria += onclause._extra_criteria
1944 elif isinstance(onclause, MapperProperty):
1945 # used internally by joined eager loader...possibly not ideal
1946 prop = onclause
1947 on_selectable = prop.parent.selectable
1948 else:
1949 prop = None
1950 on_selectable = None
1952 left_selectable = left_info.selectable
1953 if prop:
1954 adapt_from: Optional[FromClause]
1955 if sql_util.clause_is_present(on_selectable, left_selectable):
1956 adapt_from = on_selectable
1957 else:
1958 assert isinstance(left_selectable, FromClause)
1959 adapt_from = left_selectable
1961 (
1962 pj,
1963 sj,
1964 source,
1965 dest,
1966 secondary,
1967 target_adapter,
1968 ) = prop._create_joins(
1969 source_selectable=adapt_from,
1970 dest_selectable=adapt_to,
1971 source_polymorphic=True,
1972 of_type_entity=right_info,
1973 alias_secondary=True,
1974 extra_criteria=_extra_criteria,
1975 )
1977 if sj is not None:
1978 if isouter:
1979 # note this is an inner join from secondary->right
1980 right = sql.join(secondary, right, sj)
1981 onclause = pj
1982 else:
1983 left = sql.join(left, secondary, pj, isouter)
1984 onclause = sj
1985 else:
1986 onclause = pj
1988 self._target_adapter = target_adapter
1990 # we don't use the normal coercions logic for _ORMJoin
1991 # (probably should), so do some gymnastics to get the entity.
1992 # logic here is for #8721, which was a major bug in 1.4
1993 # for almost two years, not reported/fixed until 1.4.43 (!)
1994 if is_selectable(left_info):
1995 parententity = left_selectable._annotations.get(
1996 "parententity", None
1997 )
1998 elif insp_is_mapper(left_info) or insp_is_aliased_class(left_info):
1999 parententity = left_info
2000 else:
2001 parententity = None
2003 if parententity is not None:
2004 self._annotations = self._annotations.union(
2005 {"parententity": parententity}
2006 )
2008 augment_onclause = bool(_extra_criteria) and not prop
2009 expression.Join.__init__(self, left, right, onclause, isouter, full)
2011 assert self.onclause is not None
2013 if augment_onclause:
2014 self.onclause &= sql.and_(*_extra_criteria)
2016 if (
2017 not prop
2018 and getattr(right_info, "mapper", None)
2019 and right_info.mapper.single # type: ignore[union-attr]
2020 ):
2021 right_info = cast("_InternalEntityType[Any]", right_info)
2022 # if single inheritance target and we are using a manual
2023 # or implicit ON clause, augment it the same way we'd augment the
2024 # WHERE.
2025 single_crit = right_info.mapper._single_table_criterion
2026 if single_crit is not None:
2027 if insp_is_aliased_class(right_info):
2028 single_crit = right_info._adapter.traverse(single_crit)
2029 self.onclause = self.onclause & single_crit
2031 def _splice_into_center(self, other):
2032 """Splice a join into the center.
2034 Given join(a, b) and join(b, c), return join(a, b).join(c)
2036 """
2037 leftmost = other
2038 while isinstance(leftmost, sql.Join):
2039 leftmost = leftmost.left
2041 assert self.right is leftmost
2043 left = _ORMJoin(
2044 self.left,
2045 other.left,
2046 self.onclause,
2047 isouter=self.isouter,
2048 _left_memo=self._left_memo,
2049 _right_memo=other._left_memo._path_registry,
2050 )
2052 return _ORMJoin(
2053 left,
2054 other.right,
2055 other.onclause,
2056 isouter=other.isouter,
2057 _right_memo=other._right_memo,
2058 )
2060 def join(
2061 self,
2062 right: _FromClauseArgument,
2063 onclause: Optional[_OnClauseArgument] = None,
2064 isouter: bool = False,
2065 full: bool = False,
2066 ) -> _ORMJoin:
2067 return _ORMJoin(self, right, onclause, full=full, isouter=isouter)
2069 def outerjoin(
2070 self,
2071 right: _FromClauseArgument,
2072 onclause: Optional[_OnClauseArgument] = None,
2073 full: bool = False,
2074 ) -> _ORMJoin:
2075 return _ORMJoin(self, right, onclause, isouter=True, full=full)
2078def with_parent(
2079 instance: object,
2080 prop: attributes.QueryableAttribute[Any],
2081 from_entity: Optional[_EntityType[Any]] = None,
2082) -> ColumnElement[bool]:
2083 """Create filtering criterion that relates this query's primary entity
2084 to the given related instance, using established
2085 :func:`_orm.relationship()`
2086 configuration.
2088 E.g.::
2090 stmt = select(Address).where(with_parent(some_user, User.addresses))
2092 The SQL rendered is the same as that rendered when a lazy loader
2093 would fire off from the given parent on that attribute, meaning
2094 that the appropriate state is taken from the parent object in
2095 Python without the need to render joins to the parent table
2096 in the rendered statement.
2098 The given property may also make use of :meth:`_orm.PropComparator.of_type`
2099 to indicate the left side of the criteria::
2102 a1 = aliased(Address)
2103 a2 = aliased(Address)
2104 stmt = select(a1, a2).where(with_parent(u1, User.addresses.of_type(a2)))
2106 The above use is equivalent to using the
2107 :func:`_orm.with_parent.from_entity` argument::
2109 a1 = aliased(Address)
2110 a2 = aliased(Address)
2111 stmt = select(a1, a2).where(
2112 with_parent(u1, User.addresses, from_entity=a2)
2113 )
2115 :param instance:
2116 An instance which has some :func:`_orm.relationship`.
2118 :param property:
2119 Class-bound attribute, which indicates
2120 what relationship from the instance should be used to reconcile the
2121 parent/child relationship.
2123 :param from_entity:
2124 Entity in which to consider as the left side. This defaults to the
2125 "zero" entity of the :class:`_query.Query` itself.
2127 """ # noqa: E501
2128 prop_t: RelationshipProperty[Any]
2130 if isinstance(prop, str):
2131 raise sa_exc.ArgumentError(
2132 "with_parent() accepts class-bound mapped attributes, not strings"
2133 )
2134 elif isinstance(prop, attributes.QueryableAttribute):
2135 if prop._of_type:
2136 from_entity = prop._of_type
2137 mapper_property = prop.property
2138 if mapper_property is None or not prop_is_relationship(
2139 mapper_property
2140 ):
2141 raise sa_exc.ArgumentError(
2142 f"Expected relationship property for with_parent(), "
2143 f"got {mapper_property}"
2144 )
2145 prop_t = mapper_property
2146 else:
2147 prop_t = prop
2149 return prop_t._with_parent(instance, from_entity=from_entity)
2152def has_identity(object_: object) -> bool:
2153 """Return True if the given object has a database
2154 identity.
2156 This typically corresponds to the object being
2157 in either the persistent or detached state.
2159 .. seealso::
2161 :func:`.was_deleted`
2163 """
2164 state = attributes.instance_state(object_)
2165 return state.has_identity
2168def was_deleted(object_: object) -> bool:
2169 """Return True if the given object was deleted
2170 within a session flush.
2172 This is regardless of whether or not the object is
2173 persistent or detached.
2175 .. seealso::
2177 :attr:`.InstanceState.was_deleted`
2179 """
2181 state = attributes.instance_state(object_)
2182 return state.was_deleted
2185def _entity_corresponds_to(
2186 given: _InternalEntityType[Any], entity: _InternalEntityType[Any]
2187) -> bool:
2188 """determine if 'given' corresponds to 'entity', in terms
2189 of an entity passed to Query that would match the same entity
2190 being referred to elsewhere in the query.
2192 """
2193 if insp_is_aliased_class(entity):
2194 if insp_is_aliased_class(given):
2195 if entity._base_alias() is given._base_alias():
2196 return True
2197 return False
2198 elif insp_is_aliased_class(given):
2199 if given._use_mapper_path:
2200 return entity in given.with_polymorphic_mappers
2201 else:
2202 return entity is given
2204 assert insp_is_mapper(given)
2205 return entity.common_parent(given)
2208def _entity_corresponds_to_use_path_impl(
2209 given: _InternalEntityType[Any], entity: _InternalEntityType[Any]
2210) -> bool:
2211 """determine if 'given' corresponds to 'entity', in terms
2212 of a path of loader options where a mapped attribute is taken to
2213 be a member of a parent entity.
2215 e.g.::
2217 someoption(A).someoption(A.b) # -> fn(A, A) -> True
2218 someoption(A).someoption(C.d) # -> fn(A, C) -> False
2220 a1 = aliased(A)
2221 someoption(a1).someoption(A.b) # -> fn(a1, A) -> False
2222 someoption(a1).someoption(a1.b) # -> fn(a1, a1) -> True
2224 wp = with_polymorphic(A, [A1, A2])
2225 someoption(wp).someoption(A1.foo) # -> fn(wp, A1) -> False
2226 someoption(wp).someoption(wp.A1.foo) # -> fn(wp, wp.A1) -> True
2228 """
2229 if insp_is_aliased_class(given):
2230 return (
2231 insp_is_aliased_class(entity)
2232 and not entity._use_mapper_path
2233 and (given is entity or entity in given._with_polymorphic_entities)
2234 )
2235 elif not insp_is_aliased_class(entity):
2236 return given.isa(entity.mapper)
2237 else:
2238 return (
2239 entity._use_mapper_path
2240 and given in entity.with_polymorphic_mappers
2241 )
2244def _entity_isa(given: _InternalEntityType[Any], mapper: Mapper[Any]) -> bool:
2245 """determine if 'given' "is a" mapper, in terms of the given
2246 would load rows of type 'mapper'.
2248 """
2249 if given.is_aliased_class:
2250 return mapper in given.with_polymorphic_mappers or given.mapper.isa(
2251 mapper
2252 )
2253 elif given.with_polymorphic_mappers:
2254 return mapper in given.with_polymorphic_mappers or given.isa(mapper)
2255 else:
2256 return given.isa(mapper)
2259def _getitem(iterable_query: Query[Any], item: Any) -> Any:
2260 """calculate __getitem__ in terms of an iterable query object
2261 that also has a slice() method.
2263 """
2265 def _no_negative_indexes():
2266 raise IndexError(
2267 "negative indexes are not accepted by SQL "
2268 "index / slice operators"
2269 )
2271 if isinstance(item, slice):
2272 start, stop, step = util.decode_slice(item)
2274 if (
2275 isinstance(stop, int)
2276 and isinstance(start, int)
2277 and stop - start <= 0
2278 ):
2279 return []
2281 elif (isinstance(start, int) and start < 0) or (
2282 isinstance(stop, int) and stop < 0
2283 ):
2284 _no_negative_indexes()
2286 res = iterable_query.slice(start, stop)
2287 if step is not None:
2288 return list(res)[None : None : item.step]
2289 else:
2290 return list(res)
2291 else:
2292 if item == -1:
2293 _no_negative_indexes()
2294 else:
2295 return list(iterable_query[item : item + 1])[0]
2298def _is_mapped_annotation(
2299 raw_annotation: _AnnotationScanType,
2300 cls: Type[Any],
2301 originating_cls: Type[Any],
2302) -> bool:
2303 try:
2304 annotated = de_stringify_annotation(
2305 cls, raw_annotation, originating_cls.__module__
2306 )
2307 except NameError:
2308 # in most cases, at least within our own tests, we can raise
2309 # here, which is more accurate as it prevents us from returning
2310 # false negatives. However, in the real world, try to avoid getting
2311 # involved with end-user annotations that have nothing to do with us.
2312 # see issue #8888 where we bypass using this function in the case
2313 # that we want to detect an unresolvable Mapped[] type.
2314 return False
2315 else:
2316 return is_origin_of_cls(annotated, _MappedAnnotationBase)
2319class _CleanupError(Exception):
2320 pass
2323def _cleanup_mapped_str_annotation(
2324 annotation: str, originating_module: str
2325) -> str:
2326 # fix up an annotation that comes in as the form:
2327 # 'Mapped[List[Address]]' so that it instead looks like:
2328 # 'Mapped[List["Address"]]' , which will allow us to get
2329 # "Address" as a string
2331 # additionally, resolve symbols for these names since this is where
2332 # we'd have to do it
2334 inner: Optional[Match[str]]
2336 mm = re.match(r"^([^ \|]+?)\[(.+)\]$", annotation)
2338 if not mm:
2339 return annotation
2341 # ticket #8759. Resolve the Mapped name to a real symbol.
2342 # originally this just checked the name.
2343 try:
2344 obj = eval_name_only(mm.group(1), originating_module)
2345 except NameError as ne:
2346 raise _CleanupError(
2347 f'For annotation "{annotation}", could not resolve '
2348 f'container type "{mm.group(1)}". '
2349 "Please ensure this type is imported at the module level "
2350 "outside of TYPE_CHECKING blocks"
2351 ) from ne
2353 if obj is typing.ClassVar:
2354 real_symbol = "ClassVar"
2355 else:
2356 try:
2357 if issubclass(obj, _MappedAnnotationBase):
2358 real_symbol = obj.__name__
2359 else:
2360 return annotation
2361 except TypeError:
2362 # avoid isinstance(obj, type) check, just catch TypeError
2363 return annotation
2365 # note: if one of the codepaths above didn't define real_symbol and
2366 # then didn't return, real_symbol raises UnboundLocalError
2367 # which is actually a NameError, and the calling routines don't
2368 # notice this since they are catching NameError anyway. Just in case
2369 # this is being modified in the future, something to be aware of.
2371 stack = []
2372 inner = mm
2373 while True:
2374 stack.append(real_symbol if mm is inner else inner.group(1))
2375 g2 = inner.group(2)
2376 inner = re.match(r"^([^ \|]+?)\[(.+)\]$", g2)
2377 if inner is None:
2378 stack.append(g2)
2379 break
2381 # stacks we want to rewrite, that is, quote the last entry which
2382 # we think is a relationship class name:
2383 #
2384 # ['Mapped', 'List', 'Address']
2385 # ['Mapped', 'A']
2386 #
2387 # stacks we dont want to rewrite, which are generally MappedColumn
2388 # use cases:
2389 #
2390 # ['Mapped', "'Optional[Dict[str, str]]'"]
2391 # ['Mapped', 'dict[str, str] | None']
2393 if (
2394 # avoid already quoted symbols such as
2395 # ['Mapped', "'Optional[Dict[str, str]]'"]
2396 not re.match(r"""^["'].*["']$""", stack[-1])
2397 # avoid further generics like Dict[] such as
2398 # ['Mapped', 'dict[str, str] | None'],
2399 # ['Mapped', 'list[int] | list[str]'],
2400 # ['Mapped', 'Union[list[int], list[str]]'],
2401 and not re.search(r"[\[\]]", stack[-1])
2402 ):
2403 stripchars = "\"' "
2404 stack[-1] = ", ".join(
2405 f'"{elem.strip(stripchars)}"' for elem in stack[-1].split(",")
2406 )
2408 annotation = "[".join(stack) + ("]" * (len(stack) - 1))
2410 return annotation
2413def _unresolved_annotation_is_mapped(
2414 raw_annotation: _AnnotationScanType,
2415) -> bool:
2416 """given an un-resolvable annotation, guess if it's ``Mapped[]``.
2418 Used only to decide whether a ``NameError`` raised while de-stringifying
2419 is worth reporting as a mapping error. The annotation may be a plain
2420 string (``__future__`` annotations / explicitly quoted), a ``ForwardRef``
2421 (:pep:`649` deferred annotations on Python 3.14 and above), or an
2422 already-resolved object.
2424 """
2425 if isinstance(raw_annotation, str):
2426 return "Mapped[" in raw_annotation
2427 elif isinstance(raw_annotation, typing.ForwardRef):
2428 return "Mapped[" in raw_annotation.__forward_arg__
2429 else:
2430 origin = typing.get_origin(raw_annotation)
2431 return isinstance(origin, type) and issubclass(
2432 origin, _MappedAnnotationBase
2433 )
2436def _extract_mapped_subtype(
2437 raw_annotation: Optional[_AnnotationScanType],
2438 cls: type,
2439 originating_module: str,
2440 key: str,
2441 attr_cls: Type[Any],
2442 required: bool,
2443 is_dataclass_field: bool,
2444 expect_mapped: bool = True,
2445 raiseerr: bool = True,
2446) -> Optional[Tuple[Union[_AnnotationScanType, str], Optional[type]]]:
2447 """given an annotation, figure out if it's ``Mapped[something]`` and if
2448 so, return the ``something`` part.
2450 Includes error raise scenarios and other options.
2452 """
2454 if raw_annotation is None:
2455 if required:
2456 raise orm_exc.MappedAnnotationError(
2457 f"Python typing annotation is required for attribute "
2458 f'"{cls.__name__}.{key}" when primary argument(s) for '
2459 f'"{attr_cls.__name__}" construct are None or not present'
2460 )
2461 return None
2463 try:
2464 # destringify the "outside" of the annotation. note we are not
2465 # adding include_generic so it will *not* dig into generic contents,
2466 # which will remain as ForwardRef or plain str under future annotations
2467 # mode. The full destringify happens later when mapped_column goes
2468 # to do a full lookup in the registry type_annotations_map.
2469 annotated = de_stringify_annotation(
2470 cls,
2471 raw_annotation,
2472 originating_module,
2473 str_cleanup_fn=_cleanup_mapped_str_annotation,
2474 )
2475 except _CleanupError as ce:
2476 raise orm_exc.MappedAnnotationError(
2477 f"Could not interpret annotation {raw_annotation}. "
2478 "Check that it uses names that are correctly imported at the "
2479 "module level. See chained stack trace for more hints."
2480 ) from ce
2481 except NameError as ne:
2482 if raiseerr and _unresolved_annotation_is_mapped(raw_annotation):
2483 raise orm_exc.MappedAnnotationError(
2484 f"Could not interpret annotation {raw_annotation}. "
2485 "Check that it uses names that are correctly imported at the "
2486 "module level. See chained stack trace for more hints."
2487 ) from ne
2489 annotated = raw_annotation # type: ignore[assignment]
2491 if is_dataclass_field:
2492 return annotated, None
2493 else:
2494 if not hasattr(annotated, "__origin__") or not is_origin_of_cls(
2495 annotated, _MappedAnnotationBase
2496 ):
2497 if expect_mapped:
2498 if not raiseerr:
2499 return None
2501 origin = getattr(annotated, "__origin__", None)
2502 if origin is typing.ClassVar:
2503 return None
2505 # check for other kind of ORM descriptor like AssociationProxy,
2506 # don't raise for that (issue #9957)
2507 elif isinstance(origin, type) and issubclass(
2508 origin, ORMDescriptor
2509 ):
2510 return None
2512 raise orm_exc.MappedAnnotationError(
2513 f'Type annotation for "{cls.__name__}.{key}" '
2514 "can't be correctly interpreted for "
2515 "Annotated Declarative Table form. ORM annotations "
2516 "should normally make use of the ``Mapped[]`` generic "
2517 "type, or other ORM-compatible generic type, as a "
2518 "container for the actual type, which indicates the "
2519 "intent that the attribute is mapped. "
2520 "Class variables that are not intended to be mapped "
2521 "by the ORM should use ClassVar[]. "
2522 "To allow Annotated Declarative to disregard legacy "
2523 "annotations which don't use Mapped[] to pass, set "
2524 '"__allow_unmapped__ = True" on the class or a '
2525 "superclass this class.",
2526 code="zlpr",
2527 )
2529 else:
2530 return annotated, None
2532 generic_annotated = cast(GenericProtocol[Any], annotated)
2533 if len(generic_annotated.__args__) != 1:
2534 raise orm_exc.MappedAnnotationError(
2535 "Expected sub-type for Mapped[] annotation"
2536 )
2538 return (
2539 # fix dict/list/set args to be ForwardRef, see #11814
2540 fixup_container_fwd_refs(generic_annotated.__args__[0]),
2541 generic_annotated.__origin__,
2542 )
2545def _mapper_property_as_plain_name(prop: Type[Any]) -> str:
2546 if hasattr(prop, "_mapper_property_name"):
2547 name = prop._mapper_property_name()
2548 else:
2549 name = None
2550 return util.clsname_as_plain_name(prop, name)