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 Unmapped
65from .base import WriteOnlyMapped
66from .interfaces import CriteriaOption
67from .interfaces import MapperProperty as MapperProperty
68from .interfaces import ORMColumnsClauseRole
69from .interfaces import ORMEntityColumnsClauseRole
70from .interfaces import ORMFromClauseRole
71from .path_registry import PathRegistry as PathRegistry
72from .. import event
73from .. import exc as sa_exc
74from .. import inspection
75from .. import sql
76from .. import util
77from ..engine.result import result_tuple
78from ..sql import coercions
79from ..sql import expression
80from ..sql import lambdas
81from ..sql import roles
82from ..sql import util as sql_util
83from ..sql import visitors
84from ..sql._typing import is_selectable
85from ..sql.annotation import SupportsCloneAnnotations
86from ..sql.base import WriteableColumnCollection
87from ..sql.cache_key import HasCacheKey
88from ..sql.cache_key import MemoizedHasCacheKey
89from ..sql.elements import ColumnElement
90from ..sql.elements import KeyedColumnElement
91from ..sql.schema import MetaData
92from ..sql.selectable import FromClause
93from ..sql.selectable import GenerativeSelect
94from ..util.langhelpers import MemoizedSlots
95from ..util.typing import de_stringify_annotation as _de_stringify_annotation
96from ..util.typing import eval_name_only as _eval_name_only
97from ..util.typing import fixup_container_fwd_refs
98from ..util.typing import GenericProtocol
99from ..util.typing import is_origin_of_cls
100from ..util.typing import TupleAny
101from ..util.typing import Unpack
103if typing.TYPE_CHECKING:
104 from ._typing import _EntityType
105 from ._typing import _IdentityKeyType
106 from ._typing import _InternalEntityType
107 from ._typing import _ORMCOLEXPR
108 from .context import _MapperEntity
109 from .context import _ORMCompileState
110 from .context import _ORMSelectCompileState
111 from .decl_api import RegistryType
112 from .mapper import Mapper
113 from .path_registry import _AbstractEntityRegistry
114 from .query import Query
115 from .relationships import RelationshipProperty
116 from ..engine import Row
117 from ..engine import RowMapping
118 from ..sql._typing import _CE
119 from ..sql._typing import _ColumnExpressionArgument
120 from ..sql._typing import _EquivalentColumnMap
121 from ..sql._typing import _FromClauseArgument
122 from ..sql._typing import _OnClauseArgument
123 from ..sql._typing import _PropagateAttrsType
124 from ..sql.annotation import _SA
125 from ..sql.base import ReadOnlyColumnCollection
126 from ..sql.elements import BindParameter
127 from ..sql.selectable import _ColumnsClauseElement
128 from ..sql.selectable import Select
129 from ..sql.selectable import Selectable
130 from ..sql.visitors import anon_map
131 from ..util.typing import _AnnotationScanType
132 from ..util.typing import _MatchedOnType
134_T = TypeVar("_T", bound=Any)
136all_cascades = frozenset(
137 (
138 "delete",
139 "delete-orphan",
140 "all",
141 "merge",
142 "expunge",
143 "save-update",
144 "refresh-expire",
145 "none",
146 )
147)
149_de_stringify_partial = functools.partial(
150 functools.partial,
151 locals_=util.immutabledict(
152 {
153 "Mapped": Mapped,
154 "WriteOnlyMapped": WriteOnlyMapped,
155 "DynamicMapped": DynamicMapped,
156 "Unmapped": Unmapped,
157 }
158 ),
159)
161# partial is practically useless as we have to write out the whole
162# function and maintain the signature anyway
165class _DeStringifyAnnotation(Protocol):
166 def __call__(
167 self,
168 cls: Type[Any],
169 annotation: _AnnotationScanType,
170 originating_module: str,
171 *,
172 str_cleanup_fn: Optional[Callable[[str, str], str]] = None,
173 include_generic: bool = False,
174 ) -> _MatchedOnType: ...
177de_stringify_annotation = cast(
178 _DeStringifyAnnotation, _de_stringify_partial(_de_stringify_annotation)
179)
182class _EvalNameOnly(Protocol):
183 def __call__(self, name: str, module_name: str) -> Any: ...
186eval_name_only = cast(_EvalNameOnly, _de_stringify_partial(_eval_name_only))
189class CascadeOptions(FrozenSet[str]):
190 """Keeps track of the options sent to
191 :paramref:`.relationship.cascade`"""
193 _add_w_all_cascades = all_cascades.difference(
194 ["all", "none", "delete-orphan"]
195 )
196 _allowed_cascades = all_cascades
198 _viewonly_cascades = ["expunge", "all", "none", "refresh-expire", "merge"]
200 __slots__ = (
201 "save_update",
202 "delete",
203 "refresh_expire",
204 "merge",
205 "expunge",
206 "delete_orphan",
207 )
209 save_update: bool
210 delete: bool
211 refresh_expire: bool
212 merge: bool
213 expunge: bool
214 delete_orphan: bool
216 def __new__(
217 cls, value_list: Optional[Union[Iterable[str], str]]
218 ) -> CascadeOptions:
219 if isinstance(value_list, str) or value_list is None:
220 return cls.from_string(value_list) # type: ignore[no-any-return]
221 values = set(value_list)
222 if values.difference(cls._allowed_cascades):
223 raise sa_exc.ArgumentError(
224 "Invalid cascade option(s): %s"
225 % ", ".join(
226 [
227 repr(x)
228 for x in sorted(
229 values.difference(cls._allowed_cascades)
230 )
231 ]
232 )
233 )
235 if "all" in values:
236 values.update(cls._add_w_all_cascades)
237 if "none" in values:
238 values.clear()
239 values.discard("all")
241 self = super().__new__(cls, values)
242 self.save_update = "save-update" in values
243 self.delete = "delete" in values
244 self.refresh_expire = "refresh-expire" in values
245 self.merge = "merge" in values
246 self.expunge = "expunge" in values
247 self.delete_orphan = "delete-orphan" in values
249 if self.delete_orphan and not self.delete:
250 util.warn("The 'delete-orphan' cascade option requires 'delete'.")
251 return self
253 def __repr__(self):
254 return "CascadeOptions(%r)" % (",".join([x for x in sorted(self)]))
256 @classmethod
257 def from_string(cls, arg):
258 values = [c for c in re.split(r"\s*,\s*", arg or "") if c]
259 return cls(values)
262def _metadata_for_cls(cls: Type[Any], registry: RegistryType) -> MetaData:
263 meta = getattr(cls, "metadata", None)
264 if meta is not None and isinstance(meta, MetaData):
265 return meta
266 return registry.metadata
269def _validator_events(desc, key, validator, include_removes, include_backrefs):
270 """Runs a validation method on an attribute value to be set or
271 appended.
272 """
274 if not include_backrefs:
276 def detect_is_backref(state, initiator):
277 impl = state.manager[key].impl
278 return initiator.impl is not impl
280 if include_removes:
282 def append(state, value, initiator):
283 if initiator.op is not attributes.OP_BULK_REPLACE and (
284 include_backrefs or not detect_is_backref(state, initiator)
285 ):
286 return validator(state.obj(), key, value, False)
287 else:
288 return value
290 def bulk_set(state, values, initiator):
291 if include_backrefs or not detect_is_backref(state, initiator):
292 obj = state.obj()
293 values[:] = [
294 validator(obj, key, value, False) for value in values
295 ]
297 def set_(state, value, oldvalue, initiator):
298 if include_backrefs or not detect_is_backref(state, initiator):
299 return validator(state.obj(), key, value, False)
300 else:
301 return value
303 def remove(state, value, initiator):
304 if include_backrefs or not detect_is_backref(state, initiator):
305 validator(state.obj(), key, value, True)
307 else:
309 def append(state, value, initiator):
310 if initiator.op is not attributes.OP_BULK_REPLACE and (
311 include_backrefs or not detect_is_backref(state, initiator)
312 ):
313 return validator(state.obj(), key, value)
314 else:
315 return value
317 def bulk_set(state, values, initiator):
318 if include_backrefs or not detect_is_backref(state, initiator):
319 obj = state.obj()
320 values[:] = [validator(obj, key, value) for value in values]
322 def set_(state, value, oldvalue, initiator):
323 if include_backrefs or not detect_is_backref(state, initiator):
324 return validator(state.obj(), key, value)
325 else:
326 return value
328 event.listen(desc, "append", append, raw=True, retval=True)
329 event.listen(desc, "bulk_replace", bulk_set, raw=True)
330 event.listen(desc, "set", set_, raw=True, retval=True)
331 if include_removes:
332 event.listen(desc, "remove", remove, raw=True, retval=True)
335def polymorphic_union(
336 table_map, typecolname, aliasname="p_union", cast_nulls=True
337):
338 """Create a ``UNION`` statement used by a polymorphic mapper.
340 See :ref:`concrete_inheritance` for an example of how
341 this is used.
343 :param table_map: mapping of polymorphic identities to
344 :class:`_schema.Table` objects.
345 :param typecolname: string name of a "discriminator" column, which will be
346 derived from the query, producing the polymorphic identity for
347 each row. If ``None``, no polymorphic discriminator is generated.
348 :param aliasname: name of the :func:`~sqlalchemy.sql.expression.alias()`
349 construct generated.
350 :param cast_nulls: if True, non-existent columns, which are represented
351 as labeled NULLs, will be passed into CAST. This is a legacy behavior
352 that is problematic on some backends such as Oracle - in which case it
353 can be set to False.
355 """
357 colnames: util.OrderedSet[str] = util.OrderedSet()
358 colnamemaps = {}
359 types = {}
360 for key in table_map:
361 table = table_map[key]
363 table = coercions.expect(roles.FromClauseRole, table)
364 table_map[key] = table
366 m = {}
367 for c in table.c:
368 if c.key == typecolname:
369 raise sa_exc.InvalidRequestError(
370 "Polymorphic union can't use '%s' as the discriminator "
371 "column due to mapped column %r; please apply the "
372 "'typecolname' "
373 "argument; this is available on "
374 "ConcreteBase as '_concrete_discriminator_name'"
375 % (typecolname, c)
376 )
377 colnames.add(c.key)
378 m[c.key] = c
379 types[c.key] = c.type
380 colnamemaps[table] = m
382 def col(name, table):
383 try:
384 return colnamemaps[table][name]
385 except KeyError:
386 if cast_nulls:
387 return sql.cast(sql.null(), types[name]).label(name)
388 else:
389 return sql.type_coerce(sql.null(), types[name]).label(name)
391 result = []
392 for type_, table in table_map.items():
393 if typecolname is not None:
394 result.append(
395 sql.select(
396 *(
397 [col(name, table) for name in colnames]
398 + [
399 sql.literal_column(
400 sql_util._quote_ddl_expr(type_)
401 ).label(typecolname)
402 ]
403 )
404 ).select_from(table)
405 )
406 else:
407 result.append(
408 sql.select(
409 *[col(name, table) for name in colnames]
410 ).select_from(table)
411 )
412 return sql.union_all(*result).alias(aliasname)
415def identity_key(
416 class_: Optional[Type[_T]] = None,
417 ident: Union[Any, Tuple[Any, ...]] = None,
418 *,
419 instance: Optional[_T] = None,
420 row: Optional[Union[Row[Unpack[TupleAny]], RowMapping]] = None,
421 identity_token: Optional[Any] = None,
422) -> _IdentityKeyType[_T]:
423 r"""Generate "identity key" tuples, as are used as keys in the
424 :attr:`.Session.identity_map` dictionary.
426 This function has several call styles:
428 * ``identity_key(class, ident, identity_token=token)``
430 This form receives a mapped class and a primary key scalar or
431 tuple as an argument.
433 E.g.::
435 >>> identity_key(MyClass, (1, 2))
436 (<class '__main__.MyClass'>, (1, 2), None)
438 :param class: mapped class (must be a positional argument)
439 :param ident: primary key, may be a scalar or tuple argument.
440 :param identity_token: optional identity token
442 * ``identity_key(instance=instance)``
444 This form will produce the identity key for a given instance. The
445 instance need not be persistent, only that its primary key attributes
446 are populated (else the key will contain ``None`` for those missing
447 values).
449 E.g.::
451 >>> instance = MyClass(1, 2)
452 >>> identity_key(instance=instance)
453 (<class '__main__.MyClass'>, (1, 2), None)
455 In this form, the given instance is ultimately run though
456 :meth:`_orm.Mapper.identity_key_from_instance`, which will have the
457 effect of performing a database check for the corresponding row
458 if the object is expired.
460 :param instance: object instance (must be given as a keyword arg)
462 * ``identity_key(class, row=row, identity_token=token)``
464 This form is similar to the class/tuple form, except is passed a
465 database result row as a :class:`.Row` or :class:`.RowMapping` object.
467 E.g.::
469 >>> row = engine.execute(text("select * from table where a=1 and b=2")).first()
470 >>> identity_key(MyClass, row=row)
471 (<class '__main__.MyClass'>, (1, 2), None)
473 :param class: mapped class (must be a positional argument)
474 :param row: :class:`.Row` row returned by a :class:`_engine.CursorResult`
475 (must be given as a keyword arg)
476 :param identity_token: optional identity token
478 """ # noqa: E501
479 if class_ is not None:
480 mapper = class_mapper(class_)
481 if row is None:
482 if ident is None:
483 raise sa_exc.ArgumentError("ident or row is required")
484 return mapper.identity_key_from_primary_key(
485 tuple(util.to_list(ident)), identity_token=identity_token
486 )
487 else:
488 return mapper.identity_key_from_row(
489 row, identity_token=identity_token
490 )
491 elif instance is not None:
492 mapper = object_mapper(instance)
493 return mapper.identity_key_from_instance(instance)
494 else:
495 raise sa_exc.ArgumentError("class or instance is required")
498class _TraceAdaptRole(enum.Enum):
499 """Enumeration of all the use cases for ORMAdapter.
501 ORMAdapter remains one of the most complicated aspects of the ORM, as it is
502 used for in-place adaption of column expressions to be applied to a SELECT,
503 replacing :class:`.Table` and other objects that are mapped to classes with
504 aliases of those tables in the case of joined eager loading, or in the case
505 of polymorphic loading as used with concrete mappings or other custom "with
506 polymorphic" parameters, with whole user-defined subqueries. The
507 enumerations provide an overview of all the use cases used by ORMAdapter, a
508 layer of formality as to the introduction of new ORMAdapter use cases (of
509 which none are anticipated), as well as a means to trace the origins of a
510 particular ORMAdapter within runtime debugging.
512 SQLAlchemy 2.0 has greatly scaled back ORM features which relied heavily on
513 open-ended statement adaption, including the ``Query.with_polymorphic()``
514 method and the ``Query.select_from_entity()`` methods, favoring
515 user-explicit aliasing schemes using the ``aliased()`` and
516 ``with_polymorphic()`` standalone constructs; these still use adaption,
517 however the adaption is applied in a narrower scope.
519 """
521 # aliased() use that is used to adapt individual attributes at query
522 # construction time
523 ALIASED_INSP = enum.auto()
525 # joinedload cases; typically adapt an ON clause of a relationship
526 # join
527 JOINEDLOAD_USER_DEFINED_ALIAS = enum.auto()
528 JOINEDLOAD_PATH_WITH_POLYMORPHIC = enum.auto()
529 JOINEDLOAD_MEMOIZED_ADAPTER = enum.auto()
531 # polymorphic cases - these are complex ones that replace FROM
532 # clauses, replacing tables with subqueries
533 MAPPER_POLYMORPHIC_ADAPTER = enum.auto()
534 WITH_POLYMORPHIC_ADAPTER = enum.auto()
535 WITH_POLYMORPHIC_ADAPTER_RIGHT_JOIN = enum.auto()
536 DEPRECATED_JOIN_ADAPT_RIGHT_SIDE = enum.auto()
538 # the from_statement() case, used only to adapt individual attributes
539 # from a given statement to local ORM attributes at result fetching
540 # time. assigned to ORMCompileState._from_obj_alias
541 ADAPT_FROM_STATEMENT = enum.auto()
543 # the joinedload for queries that have LIMIT/OFFSET/DISTINCT case;
544 # the query is placed inside of a subquery with the LIMIT/OFFSET/etc.,
545 # joinedloads are then placed on the outside.
546 # assigned to ORMCompileState.compound_eager_adapter
547 COMPOUND_EAGER_STATEMENT = enum.auto()
549 # the legacy Query._set_select_from() case.
550 # this is needed for Query's set operations (i.e. UNION, etc. )
551 # as well as "legacy from_self()", which while removed from 2.0 as
552 # public API, is used for the Query.count() method. this one
553 # still does full statement traversal
554 # assigned to ORMCompileState._from_obj_alias
555 LEGACY_SELECT_FROM_ALIAS = enum.auto()
558class ORMStatementAdapter(sql_util.ColumnAdapter):
559 """ColumnAdapter which includes a role attribute."""
561 __slots__ = ("role",)
563 def __init__(
564 self,
565 role: _TraceAdaptRole,
566 selectable: Selectable,
567 *,
568 equivalents: Optional[_EquivalentColumnMap] = None,
569 adapt_required: bool = False,
570 allow_label_resolve: bool = True,
571 anonymize_labels: bool = False,
572 adapt_on_names: bool = False,
573 adapt_from_selectables: Optional[AbstractSet[FromClause]] = None,
574 ):
575 self.role = role
576 super().__init__(
577 selectable,
578 equivalents=equivalents,
579 adapt_required=adapt_required,
580 allow_label_resolve=allow_label_resolve,
581 anonymize_labels=anonymize_labels,
582 adapt_on_names=adapt_on_names,
583 adapt_from_selectables=adapt_from_selectables,
584 )
587class ORMAdapter(sql_util.ColumnAdapter):
588 """ColumnAdapter subclass which excludes adaptation of entities from
589 non-matching mappers.
591 """
593 __slots__ = ("role", "mapper", "is_aliased_class", "aliased_insp")
595 is_aliased_class: bool
596 aliased_insp: Optional[AliasedInsp[Any]]
598 def __init__(
599 self,
600 role: _TraceAdaptRole,
601 entity: _InternalEntityType[Any],
602 *,
603 equivalents: Optional[_EquivalentColumnMap] = None,
604 adapt_required: bool = False,
605 allow_label_resolve: bool = True,
606 anonymize_labels: bool = False,
607 selectable: Optional[Selectable] = None,
608 limit_on_entity: bool = True,
609 adapt_on_names: bool = False,
610 adapt_from_selectables: Optional[AbstractSet[FromClause]] = None,
611 ):
612 self.role = role
613 self.mapper = entity.mapper
614 if selectable is None:
615 selectable = entity.selectable
616 if insp_is_aliased_class(entity):
617 self.is_aliased_class = True
618 self.aliased_insp = entity
619 else:
620 self.is_aliased_class = False
621 self.aliased_insp = None
623 super().__init__(
624 selectable,
625 equivalents,
626 adapt_required=adapt_required,
627 allow_label_resolve=allow_label_resolve,
628 anonymize_labels=anonymize_labels,
629 include_fn=self._include_fn if limit_on_entity else None,
630 adapt_on_names=adapt_on_names,
631 adapt_from_selectables=adapt_from_selectables,
632 )
634 def _include_fn(self, elem):
635 entity = elem._annotations.get("parentmapper", None)
637 return not entity or entity.isa(self.mapper) or self.mapper.isa(entity)
640def _is_alias_of_selectable(
641 selectable: FromClause, target: FromClause
642) -> bool:
643 """Return True if ``selectable`` is ``target``, or a plain alias of it,
644 including a flat alias of a join of the same structure and join types."""
646 if selectable is target or (
647 isinstance(selectable, expression.Alias)
648 and selectable.element is target
649 ):
650 return True
652 return all(
653 elem is tgt
654 or (isinstance(elem, expression.Alias) and elem.element is tgt)
655 or (
656 # zip_longest() pads the shorter of the two with None
657 elem is not None
658 and tgt is not None
659 and elem._is_join
660 and tgt._is_join
661 and elem.isouter == tgt.isouter
662 and elem.full == tgt.full
663 )
664 or (
665 isinstance(elem, expression.FromGrouping)
666 and isinstance(tgt, expression.FromGrouping)
667 )
668 for elem, tgt in itertools.zip_longest(
669 sql_util.surface_selectables(selectable),
670 sql_util.surface_selectables(target),
671 )
672 )
675class AliasedClass(
676 inspection.Inspectable["AliasedInsp[_O]"], ORMColumnsClauseRole[_O]
677):
678 r"""Represents an "aliased" form of a mapped class for usage with Query.
680 The ORM equivalent of a :func:`~sqlalchemy.sql.expression.alias`
681 construct, this object mimics the mapped class using a
682 ``__getattr__`` scheme and maintains a reference to a
683 real :class:`~sqlalchemy.sql.expression.Alias` object.
685 A primary purpose of :class:`.AliasedClass` is to serve as an alternate
686 within a SQL statement generated by the ORM, such that an existing
687 mapped entity can be used in multiple contexts. A simple example::
689 # find all pairs of users with the same name
690 user_alias = aliased(User)
691 session.query(User, user_alias).join(
692 (user_alias, User.id > user_alias.id)
693 ).filter(User.name == user_alias.name)
695 :class:`.AliasedClass` is also capable of mapping an existing mapped
696 class to an entirely new selectable, provided this selectable is column-
697 compatible with the existing mapped selectable, and it can also be
698 configured in a mapping as the target of a :func:`_orm.relationship`.
699 See the links below for examples.
701 The :class:`.AliasedClass` object is constructed typically using the
702 :func:`_orm.aliased` function. It also is produced with additional
703 configuration when using the :func:`_orm.with_polymorphic` function.
705 The resulting object is an instance of :class:`.AliasedClass`.
706 This object implements an attribute scheme which produces the
707 same attribute and method interface as the original mapped
708 class, allowing :class:`.AliasedClass` to be compatible
709 with any attribute technique which works on the original class,
710 including hybrid attributes (see :ref:`hybrids_toplevel`).
712 The :class:`.AliasedClass` can be inspected for its underlying
713 :class:`_orm.Mapper`, aliased selectable, and other information
714 using :func:`_sa.inspect`::
716 from sqlalchemy import inspect
718 my_alias = aliased(MyClass)
719 insp = inspect(my_alias)
721 The resulting inspection object is an instance of :class:`.AliasedInsp`.
724 .. seealso::
726 :func:`.aliased`
728 :func:`.with_polymorphic`
730 :ref:`relationship_aliased_class`
732 :ref:`relationship_to_window_function`
735 """
737 __name__: str
739 def __init__(
740 self,
741 mapped_class_or_ac: _EntityType[_O],
742 alias: Optional[FromClause] = None,
743 name: Optional[str] = None,
744 flat: bool = False,
745 adapt_on_names: bool = False,
746 with_polymorphic_mappers: Optional[Sequence[Mapper[Any]]] = None,
747 with_polymorphic_discriminator: Optional[ColumnElement[Any]] = None,
748 base_alias: Optional[AliasedInsp[Any]] = None,
749 use_mapper_path: bool = False,
750 represents_outer_join: bool = False,
751 ):
752 insp = cast(
753 "_InternalEntityType[_O]", inspection.inspect(mapped_class_or_ac)
754 )
755 mapper = insp.mapper
757 nest_adapters = False
759 if alias is None:
760 if insp_is_aliased_class(insp):
761 if insp._is_with_polymorphic:
762 # polymorphic arguments are only passed along with an
763 # explicit selectable, by with_polymorphic() and
764 # AliasedInsp._merge_with()
765 assert (
766 not with_polymorphic_mappers
767 and with_polymorphic_discriminator is None
768 and not represents_outer_join
769 )
771 # aliased() of a with_polymorphic() carries along its
772 # polymorphic configuration, #13584
773 with_polymorphic_mappers = insp.with_polymorphic_mappers
774 with_polymorphic_discriminator = insp.polymorphic_on
775 represents_outer_join = insp.represents_outer_join
777 if not _is_alias_of_selectable(
778 insp.selectable, mapper._with_polymorphic_selectable
779 ):
780 # aliased() of an aliased() that refers to a subquery,
781 # CTE, with_polymorphic() or other selectable; alias
782 # that selectable rather than the mapped table,
783 # #13583, #13584
784 alias = insp.selectable._anonymous_fromclause(
785 name=name, flat=flat
786 )
787 adapt_on_names = adapt_on_names or insp._adapt_on_names
788 elif insp.selectable._is_subquery:
789 alias = insp.selectable.alias()
791 if alias is None:
792 alias = (
793 mapper._with_polymorphic_selectable._anonymous_fromclause(
794 name=name,
795 flat=flat,
796 )
797 )
798 elif insp.is_aliased_class:
799 nest_adapters = True
801 assert alias is not None
802 self._aliased_insp = AliasedInsp(
803 self,
804 insp,
805 alias,
806 name,
807 (
808 with_polymorphic_mappers
809 if with_polymorphic_mappers
810 else mapper.with_polymorphic_mappers
811 ),
812 (
813 with_polymorphic_discriminator
814 if with_polymorphic_discriminator is not None
815 else mapper.polymorphic_on
816 ),
817 base_alias,
818 use_mapper_path,
819 adapt_on_names,
820 represents_outer_join,
821 nest_adapters,
822 )
824 self.__name__ = f"aliased({mapper.class_.__name__})"
826 @classmethod
827 def _reconstitute_from_aliased_insp(
828 cls, aliased_insp: AliasedInsp[_O]
829 ) -> AliasedClass[_O]:
830 obj = cls.__new__(cls)
831 obj.__name__ = f"aliased({aliased_insp.mapper.class_.__name__})"
832 obj._aliased_insp = aliased_insp
834 if aliased_insp._is_with_polymorphic:
835 for sub_aliased_insp in aliased_insp._with_polymorphic_entities:
836 if sub_aliased_insp is not aliased_insp:
837 ent = AliasedClass._reconstitute_from_aliased_insp(
838 sub_aliased_insp
839 )
840 setattr(obj, sub_aliased_insp.class_.__name__, ent)
842 return obj
844 def __getattr__(self, key: str) -> Any:
845 try:
846 _aliased_insp = self.__dict__["_aliased_insp"]
847 except KeyError:
848 raise AttributeError()
849 else:
850 target = _aliased_insp._target
851 # maintain all getattr mechanics
852 attr = getattr(target, key)
854 # attribute is a method, that will be invoked against a
855 # "self"; so just return a new method with the same function and
856 # new self
857 if hasattr(attr, "__call__") and hasattr(attr, "__self__"):
858 return types.MethodType(attr.__func__, self)
860 # attribute is a descriptor, that will be invoked against a
861 # "self"; so invoke the descriptor against this self
862 if hasattr(attr, "__get__"):
863 attr = attr.__get__(None, self)
865 # attributes within the QueryableAttribute system will want this
866 # to be invoked so the object can be adapted
867 if hasattr(attr, "adapt_to_entity"):
868 attr = attr.adapt_to_entity(_aliased_insp)
869 setattr(self, key, attr)
871 return attr
873 def _get_from_serialized(
874 self, key: str, mapped_class: _O, aliased_insp: AliasedInsp[_O]
875 ) -> Any:
876 # this method is only used in terms of the
877 # sqlalchemy.ext.serializer extension
878 attr = getattr(mapped_class, key)
879 if hasattr(attr, "__call__") and hasattr(attr, "__self__"):
880 return types.MethodType(attr.__func__, self)
882 # attribute is a descriptor, that will be invoked against a
883 # "self"; so invoke the descriptor against this self
884 if hasattr(attr, "__get__"):
885 attr = attr.__get__(None, self)
887 # attributes within the QueryableAttribute system will want this
888 # to be invoked so the object can be adapted
889 if hasattr(attr, "adapt_to_entity"):
890 aliased_insp._weak_entity = weakref.ref(self)
891 attr = attr.adapt_to_entity(aliased_insp)
892 setattr(self, key, attr)
894 return attr
896 def __repr__(self) -> str:
897 return "<AliasedClass at 0x%x; %s>" % (
898 id(self),
899 self._aliased_insp._target.__name__,
900 )
902 def __str__(self) -> str:
903 return str(self._aliased_insp)
906@inspection._self_inspects
907class AliasedInsp(
908 ORMEntityColumnsClauseRole[_O],
909 ORMFromClauseRole,
910 HasCacheKey,
911 InspectionAttr,
912 MemoizedSlots,
913 inspection.Inspectable["AliasedInsp[_O]"],
914 Generic[_O],
915):
916 """Provide an inspection interface for an
917 :class:`.AliasedClass` object.
919 The :class:`.AliasedInsp` object is returned
920 given an :class:`.AliasedClass` using the
921 :func:`_sa.inspect` function::
923 from sqlalchemy import inspect
924 from sqlalchemy.orm import aliased
926 my_alias = aliased(MyMappedClass)
927 insp = inspect(my_alias)
929 Attributes on :class:`.AliasedInsp`
930 include:
932 * ``entity`` - the :class:`.AliasedClass` represented.
933 * ``mapper`` - the :class:`_orm.Mapper` mapping the underlying class.
934 * ``selectable`` - the :class:`_expression.Alias`
935 construct which ultimately
936 represents an aliased :class:`_schema.Table` or
937 :class:`_expression.Select`
938 construct.
939 * ``name`` - the name of the alias. Also is used as the attribute
940 name when returned in a result tuple from :class:`_query.Query`.
941 * ``with_polymorphic_mappers`` - collection of :class:`_orm.Mapper`
942 objects
943 indicating all those mappers expressed in the select construct
944 for the :class:`.AliasedClass`.
945 * ``polymorphic_on`` - an alternate column or SQL expression which
946 will be used as the "discriminator" for a polymorphic load.
948 .. seealso::
950 :ref:`inspection_toplevel`
952 """
954 __slots__ = (
955 "__weakref__",
956 "_weak_entity",
957 "mapper",
958 "selectable",
959 "name",
960 "_adapt_on_names",
961 "with_polymorphic_mappers",
962 "polymorphic_on",
963 "_use_mapper_path",
964 "_base_alias",
965 "represents_outer_join",
966 "persist_selectable",
967 "local_table",
968 "_is_with_polymorphic",
969 "_with_polymorphic_entities",
970 "_adapter",
971 "_target",
972 "__clause_element__",
973 "_memoized_values",
974 "_all_column_expressions",
975 "_nest_adapters",
976 )
978 _cache_key_traversal = [
979 ("name", visitors.ExtendedInternalTraversal.dp_string),
980 ("_adapt_on_names", visitors.ExtendedInternalTraversal.dp_boolean),
981 ("_use_mapper_path", visitors.ExtendedInternalTraversal.dp_boolean),
982 ("_target", visitors.ExtendedInternalTraversal.dp_inspectable),
983 ("selectable", visitors.ExtendedInternalTraversal.dp_clauseelement),
984 (
985 "with_polymorphic_mappers",
986 visitors.InternalTraversal.dp_has_cache_key_list,
987 ),
988 ("polymorphic_on", visitors.InternalTraversal.dp_clauseelement),
989 ]
991 mapper: Mapper[_O]
992 selectable: FromClause
993 _adapter: ORMAdapter
994 with_polymorphic_mappers: Sequence[Mapper[Any]]
995 _with_polymorphic_entities: Sequence[AliasedInsp[Any]]
996 _adapt_on_names: bool
998 _weak_entity: weakref.ref[AliasedClass[_O]]
999 """the AliasedClass that refers to this AliasedInsp"""
1001 _target: Union[Type[_O], AliasedClass[_O]]
1002 """the thing referenced by the AliasedClass/AliasedInsp.
1004 In the vast majority of cases, this is the mapped class. However
1005 it may also be another AliasedClass (alias of alias).
1007 """
1009 def __init__(
1010 self,
1011 entity: AliasedClass[_O],
1012 inspected: _InternalEntityType[_O],
1013 selectable: FromClause,
1014 name: Optional[str],
1015 with_polymorphic_mappers: Optional[Sequence[Mapper[Any]]],
1016 polymorphic_on: Optional[ColumnElement[Any]],
1017 _base_alias: Optional[AliasedInsp[Any]],
1018 _use_mapper_path: bool,
1019 adapt_on_names: bool,
1020 represents_outer_join: bool,
1021 nest_adapters: bool,
1022 ):
1023 mapped_class_or_ac = inspected.entity
1024 mapper = inspected.mapper
1026 self._weak_entity = weakref.ref(entity)
1027 self.mapper = mapper
1028 self.selectable = self.persist_selectable = self.local_table = (
1029 selectable
1030 )
1031 self.name = name
1032 self.polymorphic_on = polymorphic_on
1033 self._base_alias = weakref.ref(_base_alias or self)
1034 self._use_mapper_path = _use_mapper_path
1035 self.represents_outer_join = represents_outer_join
1036 self._nest_adapters = nest_adapters
1038 if with_polymorphic_mappers:
1039 self._is_with_polymorphic = True
1040 self.with_polymorphic_mappers = with_polymorphic_mappers
1041 self._with_polymorphic_entities = []
1042 for poly in self.with_polymorphic_mappers:
1043 if poly is not mapper:
1044 ent = AliasedClass(
1045 poly.class_,
1046 selectable,
1047 base_alias=self,
1048 adapt_on_names=adapt_on_names,
1049 use_mapper_path=_use_mapper_path,
1050 )
1052 setattr(self.entity, poly.class_.__name__, ent)
1053 self._with_polymorphic_entities.append(ent._aliased_insp)
1055 else:
1056 self._is_with_polymorphic = False
1057 self.with_polymorphic_mappers = [mapper]
1059 self._adapter = ORMAdapter(
1060 _TraceAdaptRole.ALIASED_INSP,
1061 mapper,
1062 selectable=selectable,
1063 equivalents=mapper._equivalent_columns,
1064 adapt_on_names=adapt_on_names,
1065 anonymize_labels=True,
1066 # make sure the adapter doesn't try to grab other tables that
1067 # are not even the thing we are mapping, such as embedded
1068 # selectables in subqueries or CTEs. See issue #6060
1069 adapt_from_selectables={
1070 m.selectable
1071 for m in self.with_polymorphic_mappers
1072 if not adapt_on_names
1073 },
1074 limit_on_entity=False,
1075 )
1077 if nest_adapters:
1078 # supports "aliased class of aliased class" use case
1079 assert isinstance(inspected, AliasedInsp)
1080 self._adapter = inspected._adapter.wrap(self._adapter)
1082 self._adapt_on_names = adapt_on_names
1083 self._target = mapped_class_or_ac
1085 @property
1086 def _post_inspect(self): # type: ignore[override]
1087 self.mapper._check_configure()
1089 @classmethod
1090 def _alias_factory(
1091 cls,
1092 element: Union[_EntityType[_O], FromClause],
1093 alias: Optional[FromClause] = None,
1094 name: Optional[str] = None,
1095 flat: bool = False,
1096 adapt_on_names: bool = False,
1097 ) -> Union[AliasedClass[_O], FromClause]:
1098 if isinstance(element, GenerativeSelect):
1099 return coercions.expect(roles.FromClauseRole, element, flat=flat)
1100 elif isinstance(element, FromClause):
1101 if adapt_on_names:
1102 raise sa_exc.ArgumentError(
1103 "adapt_on_names only applies to ORM elements"
1104 )
1105 if name:
1106 return element.alias(name=name, flat=flat)
1107 else:
1108 # see selectable.py->Alias._factory() for similar
1109 # mypy issue. Cannot get the overload to see this
1110 # in mypy (works fine in pyright)
1111 return coercions.expect( # type: ignore[no-any-return]
1112 roles.AnonymizedFromClauseRole, element, flat=flat
1113 )
1114 else:
1115 return AliasedClass(
1116 element,
1117 alias=alias,
1118 flat=flat,
1119 name=name,
1120 adapt_on_names=adapt_on_names,
1121 )
1123 @classmethod
1124 def _with_polymorphic_factory(
1125 cls,
1126 base: Union[Type[_O], Mapper[_O]],
1127 classes: Union[Literal["*"], Iterable[_EntityType[Any]]],
1128 selectable: Union[Literal[False, None], FromClause] = False,
1129 flat: bool = False,
1130 polymorphic_on: Optional[ColumnElement[Any]] = None,
1131 aliased: bool = False,
1132 innerjoin: bool = False,
1133 adapt_on_names: bool = False,
1134 name: Optional[str] = None,
1135 _use_mapper_path: bool = False,
1136 ) -> AliasedClass[_O]:
1137 primary_mapper = _class_to_mapper(base)
1139 if selectable not in (None, False) and flat:
1140 raise sa_exc.ArgumentError(
1141 "the 'flat' and 'selectable' arguments cannot be passed "
1142 "simultaneously to with_polymorphic()"
1143 )
1145 mappers, selectable = primary_mapper._with_polymorphic_args(
1146 classes, selectable, innerjoin=innerjoin
1147 )
1148 if aliased or flat:
1149 assert selectable is not None
1150 selectable = selectable._anonymous_fromclause(flat=flat)
1152 return AliasedClass(
1153 base,
1154 selectable,
1155 name=name,
1156 with_polymorphic_mappers=mappers,
1157 adapt_on_names=adapt_on_names,
1158 with_polymorphic_discriminator=polymorphic_on,
1159 use_mapper_path=_use_mapper_path,
1160 represents_outer_join=not innerjoin,
1161 )
1163 @property
1164 def entity(self) -> AliasedClass[_O]:
1165 # to eliminate reference cycles, the AliasedClass is held weakly.
1166 # this produces some situations where the AliasedClass gets lost,
1167 # particularly when one is created internally and only the AliasedInsp
1168 # is passed around.
1169 # to work around this case, we just generate a new one when we need
1170 # it, as it is a simple class with very little initial state on it.
1171 ent = self._weak_entity()
1172 if ent is None:
1173 ent = AliasedClass._reconstitute_from_aliased_insp(self)
1174 self._weak_entity = weakref.ref(ent)
1175 return ent
1177 is_aliased_class = True
1178 "always returns True"
1180 def _memoized_method___clause_element__(self) -> FromClause:
1181 return self.selectable._annotate(
1182 {
1183 "parentmapper": self.mapper,
1184 "parententity": self,
1185 "entity_namespace": self,
1186 }
1187 )._set_propagate_attrs(
1188 {"compile_state_plugin": "orm", "plugin_subject": self}
1189 )
1191 @property
1192 def entity_namespace(self) -> AliasedClass[_O]:
1193 return self.entity
1195 @property
1196 def class_(self) -> Type[_O]:
1197 """Return the mapped class ultimately represented by this
1198 :class:`.AliasedInsp`."""
1199 return self.mapper.class_
1201 @property
1202 def _path_registry(self) -> _AbstractEntityRegistry:
1203 if self._use_mapper_path:
1204 return self.mapper._path_registry
1205 else:
1206 return PathRegistry.per_mapper(self)
1208 def __getstate__(self) -> Dict[str, Any]:
1209 return {
1210 "entity": self.entity,
1211 "mapper": self.mapper,
1212 "alias": self.selectable,
1213 "name": self.name,
1214 "adapt_on_names": self._adapt_on_names,
1215 "with_polymorphic_mappers": self.with_polymorphic_mappers,
1216 "with_polymorphic_discriminator": self.polymorphic_on,
1217 "base_alias": self._base_alias(),
1218 "use_mapper_path": self._use_mapper_path,
1219 "represents_outer_join": self.represents_outer_join,
1220 "nest_adapters": self._nest_adapters,
1221 }
1223 def __setstate__(self, state: Dict[str, Any]) -> None:
1224 self.__init__( # type: ignore[misc]
1225 state["entity"],
1226 state["mapper"],
1227 state["alias"],
1228 state["name"],
1229 state["with_polymorphic_mappers"],
1230 state["with_polymorphic_discriminator"],
1231 state["base_alias"],
1232 state["use_mapper_path"],
1233 state["adapt_on_names"],
1234 state["represents_outer_join"],
1235 state["nest_adapters"],
1236 )
1238 def _merge_with(self, other: AliasedInsp[_O]) -> AliasedInsp[_O]:
1239 # assert self._is_with_polymorphic
1240 # assert other._is_with_polymorphic
1242 primary_mapper = other.mapper
1244 assert self.mapper is primary_mapper
1246 our_classes = util.to_set(
1247 mp.class_ for mp in self.with_polymorphic_mappers
1248 )
1249 new_classes = {mp.class_ for mp in other.with_polymorphic_mappers}
1250 if our_classes == new_classes:
1251 return other
1252 else:
1253 classes = our_classes.union(new_classes)
1255 mappers, selectable = primary_mapper._with_polymorphic_args(
1256 classes, None, innerjoin=not other.represents_outer_join
1257 )
1258 selectable = selectable._anonymous_fromclause(flat=True)
1259 return AliasedClass(
1260 primary_mapper,
1261 selectable,
1262 with_polymorphic_mappers=mappers,
1263 with_polymorphic_discriminator=other.polymorphic_on,
1264 use_mapper_path=other._use_mapper_path,
1265 represents_outer_join=other.represents_outer_join,
1266 )._aliased_insp
1268 def _adapt_element(
1269 self, expr: _ORMCOLEXPR, key: Optional[str] = None
1270 ) -> _ORMCOLEXPR:
1271 assert isinstance(expr, ColumnElement)
1272 d: Dict[str, Any] = {
1273 "parententity": self,
1274 "parentmapper": self.mapper,
1275 }
1276 if key:
1277 d["proxy_key"] = key
1279 # userspace adapt of an attribute from AliasedClass; validate that
1280 # it actually was present
1281 adapted = self._adapter.adapt_check_present(expr)
1282 if adapted is None:
1283 adapted = expr
1284 if self._adapter.adapt_on_names:
1285 util.warn_limited(
1286 "Did not locate an expression in selectable for "
1287 "attribute %r; ensure name is correct in expression",
1288 (key,),
1289 )
1290 else:
1291 util.warn_limited(
1292 "Did not locate an expression in selectable for "
1293 "attribute %r; to match by name, use the "
1294 "adapt_on_names parameter",
1295 (key,),
1296 )
1298 return adapted._annotate(d)._set_propagate_attrs(
1299 {"compile_state_plugin": "orm", "plugin_subject": self}
1300 )
1302 if TYPE_CHECKING:
1303 # establish compatibility with the _ORMAdapterProto protocol,
1304 # which in turn is compatible with _CoreAdapterProto.
1306 def _orm_adapt_element(
1307 self,
1308 obj: _CE,
1309 key: Optional[str] = None,
1310 ) -> _CE: ...
1312 else:
1313 _orm_adapt_element = _adapt_element
1315 def _entity_for_mapper(self, mapper):
1316 self_poly = self.with_polymorphic_mappers
1317 if mapper in self_poly:
1318 if mapper is self.mapper:
1319 return self
1320 else:
1321 return getattr(
1322 self.entity, mapper.class_.__name__
1323 )._aliased_insp
1324 elif mapper.isa(self.mapper):
1325 return self
1326 else:
1327 assert False, "mapper %s doesn't correspond to %s" % (mapper, self)
1329 def _memoized_attr__get_clause(self):
1330 onclause, replacemap = self.mapper._get_clause
1331 return (
1332 self._adapter.traverse(onclause),
1333 {
1334 self._adapter.traverse(col): param
1335 for col, param in replacemap.items()
1336 },
1337 )
1339 def _memoized_attr__memoized_values(self):
1340 return {}
1342 def _memoized_attr__all_column_expressions(self):
1343 if self._is_with_polymorphic:
1344 cols_plus_keys = self.mapper._columns_plus_keys(
1345 [ent.mapper for ent in self._with_polymorphic_entities]
1346 )
1347 else:
1348 cols_plus_keys = self.mapper._columns_plus_keys()
1350 cols_plus_keys = [
1351 (key, self._adapt_element(col)) for key, col in cols_plus_keys
1352 ]
1354 return WriteableColumnCollection(cols_plus_keys)
1356 def _memo(self, key, callable_, *args, **kw):
1357 if key in self._memoized_values:
1358 return self._memoized_values[key]
1359 else:
1360 self._memoized_values[key] = value = callable_(*args, **kw)
1361 return value
1363 def __repr__(self):
1364 if self.with_polymorphic_mappers:
1365 with_poly = "(%s)" % ", ".join(
1366 mp.class_.__name__ for mp in self.with_polymorphic_mappers
1367 )
1368 else:
1369 with_poly = ""
1370 return "<AliasedInsp at 0x%x; %s%s>" % (
1371 id(self),
1372 self.class_.__name__,
1373 with_poly,
1374 )
1376 def __str__(self):
1377 return self.path_string()
1379 def path_string(self) -> str:
1380 """Return a user-facing name for this :class:`.AliasedInsp`,
1381 for use in a :class:`_orm.PathRegistry` string representation.
1383 """
1384 if self._is_with_polymorphic:
1385 return "with_polymorphic(%s, [%s])" % (
1386 self._target.__name__,
1387 ", ".join(
1388 mp.class_.__name__
1389 for mp in self.with_polymorphic_mappers
1390 if mp is not self.mapper
1391 ),
1392 )
1393 else:
1394 return "aliased(%s)" % (self._target.__name__,)
1397class _WrapUserEntity:
1398 """A wrapper used within the loader_criteria lambda caller so that
1399 we can bypass declared_attr descriptors on unmapped mixins, which
1400 normally emit a warning for such use.
1402 might also be useful for other per-lambda instrumentations should
1403 the need arise.
1405 """
1407 __slots__ = ("subject",)
1409 def __init__(self, subject):
1410 self.subject = subject
1412 @util.preload_module("sqlalchemy.orm.decl_api")
1413 def __getattribute__(self, name):
1414 decl_api = util.preloaded.orm.decl_api
1416 subject = object.__getattribute__(self, "subject")
1417 if name in subject.__dict__ and isinstance(
1418 subject.__dict__[name], decl_api.declared_attr
1419 ):
1420 return subject.__dict__[name].fget(subject)
1421 else:
1422 return getattr(subject, name)
1425class LoaderCriteriaOption(CriteriaOption):
1426 """Add additional WHERE criteria to the load for all occurrences of
1427 a particular entity.
1429 :class:`_orm.LoaderCriteriaOption` is invoked using the
1430 :func:`_orm.with_loader_criteria` function; see that function for
1431 details.
1433 .. versionadded:: 1.4
1435 """
1437 __slots__ = (
1438 "root_entity",
1439 "entity",
1440 "deferred_where_criteria",
1441 "where_criteria",
1442 "_where_crit_orig",
1443 "include_aliases",
1444 "propagate_to_loaders",
1445 )
1447 _traverse_internals = [
1448 ("root_entity", visitors.ExtendedInternalTraversal.dp_plain_obj),
1449 ("entity", visitors.ExtendedInternalTraversal.dp_has_cache_key),
1450 ("where_criteria", visitors.InternalTraversal.dp_clauseelement),
1451 ("include_aliases", visitors.InternalTraversal.dp_boolean),
1452 ("propagate_to_loaders", visitors.InternalTraversal.dp_boolean),
1453 ]
1455 root_entity: Optional[Type[Any]]
1456 entity: Optional[_InternalEntityType[Any]]
1457 where_criteria: Union[ColumnElement[bool], lambdas.DeferredLambdaElement]
1458 deferred_where_criteria: bool
1459 include_aliases: bool
1460 propagate_to_loaders: bool
1462 _where_crit_orig: Any
1464 def __init__(
1465 self,
1466 entity_or_base: _EntityType[Any],
1467 where_criteria: Union[
1468 _ColumnExpressionArgument[bool],
1469 Callable[[Any], _ColumnExpressionArgument[bool]],
1470 ],
1471 loader_only: bool = False,
1472 include_aliases: bool = False,
1473 propagate_to_loaders: bool = True,
1474 track_closure_variables: bool = True,
1475 ):
1476 entity = cast(
1477 "_InternalEntityType[Any]",
1478 inspection.inspect(entity_or_base, False),
1479 )
1480 if entity is None:
1481 self.root_entity = cast("Type[Any]", entity_or_base)
1482 self.entity = None
1483 else:
1484 self.root_entity = None
1485 self.entity = entity
1487 self._where_crit_orig = where_criteria
1488 if callable(where_criteria):
1489 if self.root_entity is not None:
1490 wrap_entity = self.root_entity
1491 else:
1492 assert entity is not None
1493 wrap_entity = entity.entity
1495 self.deferred_where_criteria = True
1496 self.where_criteria = lambdas.DeferredLambdaElement(
1497 where_criteria,
1498 roles.WhereHavingRole,
1499 lambda_args=(_WrapUserEntity(wrap_entity),),
1500 opts=lambdas.LambdaOptions(
1501 track_closure_variables=track_closure_variables
1502 ),
1503 )
1504 else:
1505 self.deferred_where_criteria = False
1506 self.where_criteria = coercions.expect(
1507 roles.WhereHavingRole, where_criteria
1508 )
1510 self.include_aliases = include_aliases
1511 self.propagate_to_loaders = propagate_to_loaders
1513 @classmethod
1514 def _unreduce(
1515 cls, entity, where_criteria, include_aliases, propagate_to_loaders
1516 ):
1517 return LoaderCriteriaOption(
1518 entity,
1519 where_criteria,
1520 include_aliases=include_aliases,
1521 propagate_to_loaders=propagate_to_loaders,
1522 )
1524 def __reduce__(self):
1525 return (
1526 LoaderCriteriaOption._unreduce,
1527 (
1528 self.entity.class_ if self.entity else self.root_entity,
1529 self._where_crit_orig,
1530 self.include_aliases,
1531 self.propagate_to_loaders,
1532 ),
1533 )
1535 def _all_mappers(self) -> Iterator[Mapper[Any]]:
1536 if self.entity:
1537 yield from self.entity.mapper.self_and_descendants
1538 else:
1539 assert self.root_entity
1540 stack = list(self.root_entity.__subclasses__())
1541 while stack:
1542 subclass = stack.pop(0)
1543 ent = cast(
1544 "_InternalEntityType[Any]",
1545 inspection.inspect(subclass, raiseerr=False),
1546 )
1547 if ent:
1548 yield from ent.mapper.self_and_descendants
1549 else:
1550 stack.extend(subclass.__subclasses__())
1552 def _should_include(self, compile_state: _ORMSelectCompileState) -> bool:
1553 # don't apply the criteria to a SELECT that's part of the criteria
1554 # itself, either directly or nested at any level within other
1555 # SELECTs, such as another option's criteria that in turn refers
1556 # back to our entity; this would otherwise recurse endlessly (#13714)
1557 return self not in compile_state._containing_loader_criteria
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
2413_mapped_str_annotation = re.compile(r"(?:Mapped|Unmapped)\[")
2416def _unresolved_annotation_is_mapped(
2417 raw_annotation: _AnnotationScanType,
2418) -> bool:
2419 """given an un-resolvable annotation, guess if it's ``Mapped[]``.
2421 Used only to decide whether a ``NameError`` raised while de-stringifying
2422 is worth reporting as a mapping error. The annotation may be a plain
2423 string (``__future__`` annotations / explicitly quoted), a ``ForwardRef``
2424 (:pep:`649` deferred annotations on Python 3.14 and above), or an
2425 already-resolved object.
2427 """
2428 if isinstance(raw_annotation, str):
2429 return bool(_mapped_str_annotation.search(raw_annotation))
2430 elif isinstance(raw_annotation, typing.ForwardRef):
2431 return bool(
2432 _mapped_str_annotation.search(raw_annotation.__forward_arg__)
2433 )
2434 else:
2435 origin = typing.get_origin(raw_annotation)
2436 return isinstance(origin, type) and issubclass(
2437 origin, _MappedAnnotationBase
2438 )
2441def _extract_mapped_subtype(
2442 raw_annotation: Optional[_AnnotationScanType],
2443 cls: type,
2444 originating_module: str,
2445 key: str,
2446 attr_cls: Type[Any],
2447 required: bool,
2448 is_dataclass_field: bool,
2449 expect_mapped: bool = True,
2450 raiseerr: bool = True,
2451) -> Optional[Tuple[Union[_AnnotationScanType, str], Optional[type]]]:
2452 """given an annotation, figure out if it's ``Mapped[something]`` and if
2453 so, return the ``something`` part.
2455 Includes error raise scenarios and other options.
2457 """
2459 if raw_annotation is None:
2460 if required:
2461 raise orm_exc.MappedAnnotationError(
2462 f"Python typing annotation is required for attribute "
2463 f'"{cls.__name__}.{key}" when primary argument(s) for '
2464 f'"{attr_cls.__name__}" construct are None or not present'
2465 )
2466 return None
2468 try:
2469 # destringify the "outside" of the annotation. note we are not
2470 # adding include_generic so it will *not* dig into generic contents,
2471 # which will remain as ForwardRef or plain str under future annotations
2472 # mode. The full destringify happens later when mapped_column goes
2473 # to do a full lookup in the registry type_annotations_map.
2474 annotated = de_stringify_annotation(
2475 cls,
2476 raw_annotation,
2477 originating_module,
2478 str_cleanup_fn=_cleanup_mapped_str_annotation,
2479 )
2480 except _CleanupError as ce:
2481 raise orm_exc.MappedAnnotationError(
2482 f"Could not interpret annotation {raw_annotation}. "
2483 "Check that it uses names that are correctly imported at the "
2484 "module level. See chained stack trace for more hints."
2485 ) from ce
2486 except NameError as ne:
2487 if raiseerr and _unresolved_annotation_is_mapped(raw_annotation):
2488 raise orm_exc.MappedAnnotationError(
2489 f"Could not interpret annotation {raw_annotation}. "
2490 "Check that it uses names that are correctly imported at the "
2491 "module level. See chained stack trace for more hints."
2492 ) from ne
2494 annotated = raw_annotation # type: ignore[assignment]
2496 if is_dataclass_field:
2497 return annotated, None
2498 else:
2499 if not hasattr(annotated, "__origin__") or not is_origin_of_cls(
2500 annotated, _MappedAnnotationBase
2501 ):
2502 if expect_mapped:
2503 if not raiseerr:
2504 return None
2506 origin = getattr(annotated, "__origin__", None)
2507 if origin is typing.ClassVar:
2508 return None
2510 # check for other kind of ORM descriptor like AssociationProxy,
2511 # don't raise for that (issue #9957)
2512 elif isinstance(origin, type) and issubclass(
2513 origin, ORMDescriptor
2514 ):
2515 return None
2517 raise orm_exc.MappedAnnotationError(
2518 f'Type annotation for "{cls.__name__}.{key}" '
2519 "can't be correctly interpreted for "
2520 "Annotated Declarative Table form. ORM annotations "
2521 "should normally make use of the ``Mapped[]`` generic "
2522 "type, or other ORM-compatible generic type, as a "
2523 "container for the actual type, which indicates the "
2524 "intent that the attribute is mapped. "
2525 "Class variables that are not intended to be mapped "
2526 "by the ORM should use ClassVar[]. "
2527 "To allow Annotated Declarative to disregard legacy "
2528 "annotations which don't use Mapped[] to pass, set "
2529 '"__allow_unmapped__ = True" on the class or a '
2530 "superclass this class.",
2531 code="zlpr",
2532 )
2534 else:
2535 return annotated, None
2537 generic_annotated = cast(GenericProtocol[Any], annotated)
2538 if len(generic_annotated.__args__) != 1:
2539 raise orm_exc.MappedAnnotationError(
2540 "Expected sub-type for Mapped[] annotation"
2541 )
2543 return (
2544 # fix dict/list/set args to be ForwardRef, see #11814
2545 fixup_container_fwd_refs(generic_annotated.__args__[0]),
2546 generic_annotated.__origin__,
2547 )
2550def _mapper_property_as_plain_name(prop: Type[Any]) -> str:
2551 if hasattr(prop, "_mapper_property_name"):
2552 name = prop._mapper_property_name()
2553 else:
2554 name = None
2555 return util.clsname_as_plain_name(prop, name)