Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/util.py: 34%
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 re
14import types
15import typing
16from typing import AbstractSet
17from typing import Any
18from typing import Callable
19from typing import cast
20from typing import Dict
21from typing import FrozenSet
22from typing import Generic
23from typing import Iterable
24from typing import Iterator
25from typing import List
26from typing import Match
27from typing import Optional
28from typing import Sequence
29from typing import Tuple
30from typing import Type
31from typing import TYPE_CHECKING
32from typing import TypeVar
33from typing import Union
34import weakref
36from . import attributes # noqa
37from . import exc
38from . import exc as orm_exc
39from ._typing import _O
40from ._typing import insp_is_aliased_class
41from ._typing import insp_is_mapper
42from ._typing import prop_is_relationship
43from .base import _class_to_mapper as _class_to_mapper
44from .base import _MappedAnnotationBase
45from .base import _never_set as _never_set # noqa: F401
46from .base import _none_only_set as _none_only_set # noqa: F401
47from .base import _none_set as _none_set # noqa: F401
48from .base import attribute_str as attribute_str # noqa: F401
49from .base import class_mapper as class_mapper
50from .base import DynamicMapped
51from .base import InspectionAttr as InspectionAttr
52from .base import instance_str as instance_str # noqa: F401
53from .base import Mapped
54from .base import object_mapper as object_mapper
55from .base import object_state as object_state # noqa: F401
56from .base import opt_manager_of_class
57from .base import ORMDescriptor
58from .base import state_attribute_str as state_attribute_str # noqa: F401
59from .base import state_class_str as state_class_str # noqa: F401
60from .base import state_str as state_str # noqa: F401
61from .base import WriteOnlyMapped
62from .interfaces import CriteriaOption
63from .interfaces import MapperProperty as MapperProperty
64from .interfaces import ORMColumnsClauseRole
65from .interfaces import ORMEntityColumnsClauseRole
66from .interfaces import ORMFromClauseRole
67from .path_registry import PathRegistry as PathRegistry
68from .. import event
69from .. import exc as sa_exc
70from .. import inspection
71from .. import sql
72from .. import util
73from ..engine.result import result_tuple
74from ..sql import coercions
75from ..sql import expression
76from ..sql import lambdas
77from ..sql import roles
78from ..sql import util as sql_util
79from ..sql import visitors
80from ..sql._typing import is_selectable
81from ..sql.annotation import SupportsCloneAnnotations
82from ..sql.base import ColumnCollection
83from ..sql.cache_key import HasCacheKey
84from ..sql.cache_key import MemoizedHasCacheKey
85from ..sql.elements import ColumnElement
86from ..sql.elements import KeyedColumnElement
87from ..sql.selectable import FromClause
88from ..sql.selectable import GenerativeSelect
89from ..util.langhelpers import MemoizedSlots
90from ..util.typing import de_stringify_annotation as _de_stringify_annotation
91from ..util.typing import eval_name_only as _eval_name_only
92from ..util.typing import fixup_container_fwd_refs
93from ..util.typing import get_origin
94from ..util.typing import is_origin_of_cls
95from ..util.typing import Literal
96from ..util.typing import Protocol
98if typing.TYPE_CHECKING:
99 from ._typing import _EntityType
100 from ._typing import _IdentityKeyType
101 from ._typing import _InternalEntityType
102 from ._typing import _ORMCOLEXPR
103 from .context import _MapperEntity
104 from .context import ORMCompileState
105 from .mapper import Mapper
106 from .path_registry import AbstractEntityRegistry
107 from .query import Query
108 from .relationships import RelationshipProperty
109 from ..engine import Row
110 from ..engine import RowMapping
111 from ..sql._typing import _CE
112 from ..sql._typing import _ColumnExpressionArgument
113 from ..sql._typing import _EquivalentColumnMap
114 from ..sql._typing import _FromClauseArgument
115 from ..sql._typing import _OnClauseArgument
116 from ..sql._typing import _PropagateAttrsType
117 from ..sql.annotation import _SA
118 from ..sql.base import ReadOnlyColumnCollection
119 from ..sql.elements import BindParameter
120 from ..sql.selectable import _ColumnsClauseElement
121 from ..sql.selectable import Select
122 from ..sql.selectable import Selectable
123 from ..sql.visitors import anon_map
124 from ..util.typing import _AnnotationScanType
126_T = TypeVar("_T", bound=Any)
128all_cascades = frozenset(
129 (
130 "delete",
131 "delete-orphan",
132 "all",
133 "merge",
134 "expunge",
135 "save-update",
136 "refresh-expire",
137 "none",
138 )
139)
141_de_stringify_partial = functools.partial(
142 functools.partial,
143 locals_=util.immutabledict(
144 {
145 "Mapped": Mapped,
146 "WriteOnlyMapped": WriteOnlyMapped,
147 "DynamicMapped": DynamicMapped,
148 }
149 ),
150)
152# partial is practically useless as we have to write out the whole
153# function and maintain the signature anyway
156class _DeStringifyAnnotation(Protocol):
157 def __call__(
158 self,
159 cls: Type[Any],
160 annotation: _AnnotationScanType,
161 originating_module: str,
162 *,
163 str_cleanup_fn: Optional[Callable[[str, str], str]] = None,
164 include_generic: bool = False,
165 ) -> Type[Any]: ...
168de_stringify_annotation = cast(
169 _DeStringifyAnnotation, _de_stringify_partial(_de_stringify_annotation)
170)
173class _EvalNameOnly(Protocol):
174 def __call__(self, name: str, module_name: str) -> Any: ...
177eval_name_only = cast(_EvalNameOnly, _de_stringify_partial(_eval_name_only))
180class CascadeOptions(FrozenSet[str]):
181 """Keeps track of the options sent to
182 :paramref:`.relationship.cascade`"""
184 _add_w_all_cascades = all_cascades.difference(
185 ["all", "none", "delete-orphan"]
186 )
187 _allowed_cascades = all_cascades
189 _viewonly_cascades = ["expunge", "all", "none", "refresh-expire", "merge"]
191 __slots__ = (
192 "save_update",
193 "delete",
194 "refresh_expire",
195 "merge",
196 "expunge",
197 "delete_orphan",
198 )
200 save_update: bool
201 delete: bool
202 refresh_expire: bool
203 merge: bool
204 expunge: bool
205 delete_orphan: bool
207 def __new__(
208 cls, value_list: Optional[Union[Iterable[str], str]]
209 ) -> CascadeOptions:
210 if isinstance(value_list, str) or value_list is None:
211 return cls.from_string(value_list) # type: ignore
212 values = set(value_list)
213 if values.difference(cls._allowed_cascades):
214 raise sa_exc.ArgumentError(
215 "Invalid cascade option(s): %s"
216 % ", ".join(
217 [
218 repr(x)
219 for x in sorted(
220 values.difference(cls._allowed_cascades)
221 )
222 ]
223 )
224 )
226 if "all" in values:
227 values.update(cls._add_w_all_cascades)
228 if "none" in values:
229 values.clear()
230 values.discard("all")
232 self = super().__new__(cls, values)
233 self.save_update = "save-update" in values
234 self.delete = "delete" in values
235 self.refresh_expire = "refresh-expire" in values
236 self.merge = "merge" in values
237 self.expunge = "expunge" in values
238 self.delete_orphan = "delete-orphan" in values
240 if self.delete_orphan and not self.delete:
241 util.warn("The 'delete-orphan' cascade option requires 'delete'.")
242 return self
244 def __repr__(self):
245 return "CascadeOptions(%r)" % (",".join([x for x in sorted(self)]))
247 @classmethod
248 def from_string(cls, arg):
249 values = [c for c in re.split(r"\s*,\s*", arg or "") if c]
250 return cls(values)
253def _validator_events(desc, key, validator, include_removes, include_backrefs):
254 """Runs a validation method on an attribute value to be set or
255 appended.
256 """
258 if not include_backrefs:
260 def detect_is_backref(state, initiator):
261 impl = state.manager[key].impl
262 return initiator.impl is not impl
264 if include_removes:
266 def append(state, value, initiator):
267 if initiator.op is not attributes.OP_BULK_REPLACE and (
268 include_backrefs or not detect_is_backref(state, initiator)
269 ):
270 return validator(state.obj(), key, value, False)
271 else:
272 return value
274 def bulk_set(state, values, initiator):
275 if include_backrefs or not detect_is_backref(state, initiator):
276 obj = state.obj()
277 values[:] = [
278 validator(obj, key, value, False) for value in values
279 ]
281 def set_(state, value, oldvalue, initiator):
282 if include_backrefs or not detect_is_backref(state, initiator):
283 return validator(state.obj(), key, value, False)
284 else:
285 return value
287 def remove(state, value, initiator):
288 if include_backrefs or not detect_is_backref(state, initiator):
289 validator(state.obj(), key, value, True)
291 else:
293 def append(state, value, initiator):
294 if initiator.op is not attributes.OP_BULK_REPLACE and (
295 include_backrefs or not detect_is_backref(state, initiator)
296 ):
297 return validator(state.obj(), key, value)
298 else:
299 return value
301 def bulk_set(state, values, initiator):
302 if include_backrefs or not detect_is_backref(state, initiator):
303 obj = state.obj()
304 values[:] = [validator(obj, key, value) for value in values]
306 def set_(state, value, oldvalue, initiator):
307 if include_backrefs or not detect_is_backref(state, initiator):
308 return validator(state.obj(), key, value)
309 else:
310 return value
312 event.listen(desc, "append", append, raw=True, retval=True)
313 event.listen(desc, "bulk_replace", bulk_set, raw=True)
314 event.listen(desc, "set", set_, raw=True, retval=True)
315 if include_removes:
316 event.listen(desc, "remove", remove, raw=True, retval=True)
319def polymorphic_union(
320 table_map, typecolname, aliasname="p_union", cast_nulls=True
321):
322 """Create a ``UNION`` statement used by a polymorphic mapper.
324 See :ref:`concrete_inheritance` for an example of how
325 this is used.
327 :param table_map: mapping of polymorphic identities to
328 :class:`_schema.Table` objects.
329 :param typecolname: string name of a "discriminator" column, which will be
330 derived from the query, producing the polymorphic identity for
331 each row. If ``None``, no polymorphic discriminator is generated.
332 :param aliasname: name of the :func:`~sqlalchemy.sql.expression.alias()`
333 construct generated.
334 :param cast_nulls: if True, non-existent columns, which are represented
335 as labeled NULLs, will be passed into CAST. This is a legacy behavior
336 that is problematic on some backends such as Oracle - in which case it
337 can be set to False.
339 """
341 colnames: util.OrderedSet[str] = util.OrderedSet()
342 colnamemaps = {}
343 types = {}
344 for key in table_map:
345 table = table_map[key]
347 table = coercions.expect(
348 roles.StrictFromClauseRole, table, allow_select=True
349 )
350 table_map[key] = table
352 m = {}
353 for c in table.c:
354 if c.key == typecolname:
355 raise sa_exc.InvalidRequestError(
356 "Polymorphic union can't use '%s' as the discriminator "
357 "column due to mapped column %r; please apply the "
358 "'typecolname' "
359 "argument; this is available on "
360 "ConcreteBase as '_concrete_discriminator_name'"
361 % (typecolname, c)
362 )
363 colnames.add(c.key)
364 m[c.key] = c
365 types[c.key] = c.type
366 colnamemaps[table] = m
368 def col(name, table):
369 try:
370 return colnamemaps[table][name]
371 except KeyError:
372 if cast_nulls:
373 return sql.cast(sql.null(), types[name]).label(name)
374 else:
375 return sql.type_coerce(sql.null(), types[name]).label(name)
377 result = []
378 for type_, table in table_map.items():
379 if typecolname is not None:
380 result.append(
381 sql.select(
382 *(
383 [col(name, table) for name in colnames]
384 + [
385 sql.literal_column(
386 sql_util._quote_ddl_expr(type_)
387 ).label(typecolname)
388 ]
389 )
390 ).select_from(table)
391 )
392 else:
393 result.append(
394 sql.select(
395 *[col(name, table) for name in colnames]
396 ).select_from(table)
397 )
398 return sql.union_all(*result).alias(aliasname)
401def identity_key(
402 class_: Optional[Type[_T]] = None,
403 ident: Union[Any, Tuple[Any, ...]] = None,
404 *,
405 instance: Optional[_T] = None,
406 row: Optional[Union[Row[Any], RowMapping]] = None,
407 identity_token: Optional[Any] = None,
408) -> _IdentityKeyType[_T]:
409 r"""Generate "identity key" tuples, as are used as keys in the
410 :attr:`.Session.identity_map` dictionary.
412 This function has several call styles:
414 * ``identity_key(class, ident, identity_token=token)``
416 This form receives a mapped class and a primary key scalar or
417 tuple as an argument.
419 E.g.::
421 >>> identity_key(MyClass, (1, 2))
422 (<class '__main__.MyClass'>, (1, 2), None)
424 :param class: mapped class (must be a positional argument)
425 :param ident: primary key, may be a scalar or tuple argument.
426 :param identity_token: optional identity token
428 .. versionadded:: 1.2 added identity_token
431 * ``identity_key(instance=instance)``
433 This form will produce the identity key for a given instance. The
434 instance need not be persistent, only that its primary key attributes
435 are populated (else the key will contain ``None`` for those missing
436 values).
438 E.g.::
440 >>> instance = MyClass(1, 2)
441 >>> identity_key(instance=instance)
442 (<class '__main__.MyClass'>, (1, 2), None)
444 In this form, the given instance is ultimately run though
445 :meth:`_orm.Mapper.identity_key_from_instance`, which will have the
446 effect of performing a database check for the corresponding row
447 if the object is expired.
449 :param instance: object instance (must be given as a keyword arg)
451 * ``identity_key(class, row=row, identity_token=token)``
453 This form is similar to the class/tuple form, except is passed a
454 database result row as a :class:`.Row` or :class:`.RowMapping` object.
456 E.g.::
458 >>> row = engine.execute(text("select * from table where a=1 and b=2")).first()
459 >>> identity_key(MyClass, row=row)
460 (<class '__main__.MyClass'>, (1, 2), None)
462 :param class: mapped class (must be a positional argument)
463 :param row: :class:`.Row` row returned by a :class:`_engine.CursorResult`
464 (must be given as a keyword arg)
465 :param identity_token: optional identity token
467 .. versionadded:: 1.2 added identity_token
469 """ # noqa: E501
470 if class_ is not None:
471 mapper = class_mapper(class_)
472 if row is None:
473 if ident is None:
474 raise sa_exc.ArgumentError("ident or row is required")
475 return mapper.identity_key_from_primary_key(
476 tuple(util.to_list(ident)), identity_token=identity_token
477 )
478 else:
479 return mapper.identity_key_from_row(
480 row, identity_token=identity_token
481 )
482 elif instance is not None:
483 mapper = object_mapper(instance)
484 return mapper.identity_key_from_instance(instance)
485 else:
486 raise sa_exc.ArgumentError("class or instance is required")
489class _TraceAdaptRole(enum.Enum):
490 """Enumeration of all the use cases for ORMAdapter.
492 ORMAdapter remains one of the most complicated aspects of the ORM, as it is
493 used for in-place adaption of column expressions to be applied to a SELECT,
494 replacing :class:`.Table` and other objects that are mapped to classes with
495 aliases of those tables in the case of joined eager loading, or in the case
496 of polymorphic loading as used with concrete mappings or other custom "with
497 polymorphic" parameters, with whole user-defined subqueries. The
498 enumerations provide an overview of all the use cases used by ORMAdapter, a
499 layer of formality as to the introduction of new ORMAdapter use cases (of
500 which none are anticipated), as well as a means to trace the origins of a
501 particular ORMAdapter within runtime debugging.
503 SQLAlchemy 2.0 has greatly scaled back ORM features which relied heavily on
504 open-ended statement adaption, including the ``Query.with_polymorphic()``
505 method and the ``Query.select_from_entity()`` methods, favoring
506 user-explicit aliasing schemes using the ``aliased()`` and
507 ``with_polymorphic()`` standalone constructs; these still use adaption,
508 however the adaption is applied in a narrower scope.
510 """
512 # aliased() use that is used to adapt individual attributes at query
513 # construction time
514 ALIASED_INSP = enum.auto()
516 # joinedload cases; typically adapt an ON clause of a relationship
517 # join
518 JOINEDLOAD_USER_DEFINED_ALIAS = enum.auto()
519 JOINEDLOAD_PATH_WITH_POLYMORPHIC = enum.auto()
520 JOINEDLOAD_MEMOIZED_ADAPTER = enum.auto()
522 # polymorphic cases - these are complex ones that replace FROM
523 # clauses, replacing tables with subqueries
524 MAPPER_POLYMORPHIC_ADAPTER = enum.auto()
525 WITH_POLYMORPHIC_ADAPTER = enum.auto()
526 WITH_POLYMORPHIC_ADAPTER_RIGHT_JOIN = enum.auto()
527 DEPRECATED_JOIN_ADAPT_RIGHT_SIDE = enum.auto()
529 # the from_statement() case, used only to adapt individual attributes
530 # from a given statement to local ORM attributes at result fetching
531 # time. assigned to ORMCompileState._from_obj_alias
532 ADAPT_FROM_STATEMENT = enum.auto()
534 # the joinedload for queries that have LIMIT/OFFSET/DISTINCT case;
535 # the query is placed inside of a subquery with the LIMIT/OFFSET/etc.,
536 # joinedloads are then placed on the outside.
537 # assigned to ORMCompileState.compound_eager_adapter
538 COMPOUND_EAGER_STATEMENT = enum.auto()
540 # the legacy Query._set_select_from() case.
541 # this is needed for Query's set operations (i.e. UNION, etc. )
542 # as well as "legacy from_self()", which while removed from 2.0 as
543 # public API, is used for the Query.count() method. this one
544 # still does full statement traversal
545 # assigned to ORMCompileState._from_obj_alias
546 LEGACY_SELECT_FROM_ALIAS = enum.auto()
549class ORMStatementAdapter(sql_util.ColumnAdapter):
550 """ColumnAdapter which includes a role attribute."""
552 __slots__ = ("role",)
554 def __init__(
555 self,
556 role: _TraceAdaptRole,
557 selectable: Selectable,
558 *,
559 equivalents: Optional[_EquivalentColumnMap] = None,
560 adapt_required: bool = False,
561 allow_label_resolve: bool = True,
562 anonymize_labels: bool = False,
563 adapt_on_names: bool = False,
564 adapt_from_selectables: Optional[AbstractSet[FromClause]] = None,
565 ):
566 self.role = role
567 super().__init__(
568 selectable,
569 equivalents=equivalents,
570 adapt_required=adapt_required,
571 allow_label_resolve=allow_label_resolve,
572 anonymize_labels=anonymize_labels,
573 adapt_on_names=adapt_on_names,
574 adapt_from_selectables=adapt_from_selectables,
575 )
578class ORMAdapter(sql_util.ColumnAdapter):
579 """ColumnAdapter subclass which excludes adaptation of entities from
580 non-matching mappers.
582 """
584 __slots__ = ("role", "mapper", "is_aliased_class", "aliased_insp")
586 is_aliased_class: bool
587 aliased_insp: Optional[AliasedInsp[Any]]
589 def __init__(
590 self,
591 role: _TraceAdaptRole,
592 entity: _InternalEntityType[Any],
593 *,
594 equivalents: Optional[_EquivalentColumnMap] = None,
595 adapt_required: bool = False,
596 allow_label_resolve: bool = True,
597 anonymize_labels: bool = False,
598 selectable: Optional[Selectable] = None,
599 limit_on_entity: bool = True,
600 adapt_on_names: bool = False,
601 adapt_from_selectables: Optional[AbstractSet[FromClause]] = None,
602 ):
603 self.role = role
604 self.mapper = entity.mapper
605 if selectable is None:
606 selectable = entity.selectable
607 if insp_is_aliased_class(entity):
608 self.is_aliased_class = True
609 self.aliased_insp = entity
610 else:
611 self.is_aliased_class = False
612 self.aliased_insp = None
614 super().__init__(
615 selectable,
616 equivalents,
617 adapt_required=adapt_required,
618 allow_label_resolve=allow_label_resolve,
619 anonymize_labels=anonymize_labels,
620 include_fn=self._include_fn if limit_on_entity else None,
621 adapt_on_names=adapt_on_names,
622 adapt_from_selectables=adapt_from_selectables,
623 )
625 def _include_fn(self, elem):
626 entity = elem._annotations.get("parentmapper", None)
628 return not entity or entity.isa(self.mapper) or self.mapper.isa(entity)
631class AliasedClass(
632 inspection.Inspectable["AliasedInsp[_O]"], ORMColumnsClauseRole[_O]
633):
634 r"""Represents an "aliased" form of a mapped class for usage with Query.
636 The ORM equivalent of a :func:`~sqlalchemy.sql.expression.alias`
637 construct, this object mimics the mapped class using a
638 ``__getattr__`` scheme and maintains a reference to a
639 real :class:`~sqlalchemy.sql.expression.Alias` object.
641 A primary purpose of :class:`.AliasedClass` is to serve as an alternate
642 within a SQL statement generated by the ORM, such that an existing
643 mapped entity can be used in multiple contexts. A simple example::
645 # find all pairs of users with the same name
646 user_alias = aliased(User)
647 session.query(User, user_alias).join(
648 (user_alias, User.id > user_alias.id)
649 ).filter(User.name == user_alias.name)
651 :class:`.AliasedClass` is also capable of mapping an existing mapped
652 class to an entirely new selectable, provided this selectable is column-
653 compatible with the existing mapped selectable, and it can also be
654 configured in a mapping as the target of a :func:`_orm.relationship`.
655 See the links below for examples.
657 The :class:`.AliasedClass` object is constructed typically using the
658 :func:`_orm.aliased` function. It also is produced with additional
659 configuration when using the :func:`_orm.with_polymorphic` function.
661 The resulting object is an instance of :class:`.AliasedClass`.
662 This object implements an attribute scheme which produces the
663 same attribute and method interface as the original mapped
664 class, allowing :class:`.AliasedClass` to be compatible
665 with any attribute technique which works on the original class,
666 including hybrid attributes (see :ref:`hybrids_toplevel`).
668 The :class:`.AliasedClass` can be inspected for its underlying
669 :class:`_orm.Mapper`, aliased selectable, and other information
670 using :func:`_sa.inspect`::
672 from sqlalchemy import inspect
674 my_alias = aliased(MyClass)
675 insp = inspect(my_alias)
677 The resulting inspection object is an instance of :class:`.AliasedInsp`.
680 .. seealso::
682 :func:`.aliased`
684 :func:`.with_polymorphic`
686 :ref:`relationship_aliased_class`
688 :ref:`relationship_to_window_function`
691 """
693 __name__: str
695 def __init__(
696 self,
697 mapped_class_or_ac: _EntityType[_O],
698 alias: Optional[FromClause] = None,
699 name: Optional[str] = None,
700 flat: bool = False,
701 adapt_on_names: bool = False,
702 with_polymorphic_mappers: Optional[Sequence[Mapper[Any]]] = None,
703 with_polymorphic_discriminator: Optional[ColumnElement[Any]] = None,
704 base_alias: Optional[AliasedInsp[Any]] = None,
705 use_mapper_path: bool = False,
706 represents_outer_join: bool = False,
707 ):
708 insp = cast(
709 "_InternalEntityType[_O]", inspection.inspect(mapped_class_or_ac)
710 )
711 mapper = insp.mapper
713 nest_adapters = False
715 if alias is None:
716 if insp.is_aliased_class and insp.selectable._is_subquery:
717 alias = insp.selectable.alias()
718 else:
719 alias = (
720 mapper._with_polymorphic_selectable._anonymous_fromclause(
721 name=name,
722 flat=flat,
723 )
724 )
725 elif insp.is_aliased_class:
726 nest_adapters = True
728 assert alias is not None
729 self._aliased_insp = AliasedInsp(
730 self,
731 insp,
732 alias,
733 name,
734 (
735 with_polymorphic_mappers
736 if with_polymorphic_mappers
737 else mapper.with_polymorphic_mappers
738 ),
739 (
740 with_polymorphic_discriminator
741 if with_polymorphic_discriminator is not None
742 else mapper.polymorphic_on
743 ),
744 base_alias,
745 use_mapper_path,
746 adapt_on_names,
747 represents_outer_join,
748 nest_adapters,
749 )
751 self.__name__ = f"aliased({mapper.class_.__name__})"
753 @classmethod
754 def _reconstitute_from_aliased_insp(
755 cls, aliased_insp: AliasedInsp[_O]
756 ) -> AliasedClass[_O]:
757 obj = cls.__new__(cls)
758 obj.__name__ = f"aliased({aliased_insp.mapper.class_.__name__})"
759 obj._aliased_insp = aliased_insp
761 if aliased_insp._is_with_polymorphic:
762 for sub_aliased_insp in aliased_insp._with_polymorphic_entities:
763 if sub_aliased_insp is not aliased_insp:
764 ent = AliasedClass._reconstitute_from_aliased_insp(
765 sub_aliased_insp
766 )
767 setattr(obj, sub_aliased_insp.class_.__name__, ent)
769 return obj
771 def __getattr__(self, key: str) -> Any:
772 try:
773 _aliased_insp = self.__dict__["_aliased_insp"]
774 except KeyError:
775 raise AttributeError()
776 else:
777 target = _aliased_insp._target
778 # maintain all getattr mechanics
779 attr = getattr(target, key)
781 # attribute is a method, that will be invoked against a
782 # "self"; so just return a new method with the same function and
783 # new self
784 if hasattr(attr, "__call__") and hasattr(attr, "__self__"):
785 return types.MethodType(attr.__func__, self)
787 # attribute is a descriptor, that will be invoked against a
788 # "self"; so invoke the descriptor against this self
789 if hasattr(attr, "__get__"):
790 attr = attr.__get__(None, self)
792 # attributes within the QueryableAttribute system will want this
793 # to be invoked so the object can be adapted
794 if hasattr(attr, "adapt_to_entity"):
795 attr = attr.adapt_to_entity(_aliased_insp)
796 setattr(self, key, attr)
798 return attr
800 def _get_from_serialized(
801 self, key: str, mapped_class: _O, aliased_insp: AliasedInsp[_O]
802 ) -> Any:
803 # this method is only used in terms of the
804 # sqlalchemy.ext.serializer extension
805 attr = getattr(mapped_class, key)
806 if hasattr(attr, "__call__") and hasattr(attr, "__self__"):
807 return types.MethodType(attr.__func__, self)
809 # attribute is a descriptor, that will be invoked against a
810 # "self"; so invoke the descriptor against this self
811 if hasattr(attr, "__get__"):
812 attr = attr.__get__(None, self)
814 # attributes within the QueryableAttribute system will want this
815 # to be invoked so the object can be adapted
816 if hasattr(attr, "adapt_to_entity"):
817 aliased_insp._weak_entity = weakref.ref(self)
818 attr = attr.adapt_to_entity(aliased_insp)
819 setattr(self, key, attr)
821 return attr
823 def __repr__(self) -> str:
824 return "<AliasedClass at 0x%x; %s>" % (
825 id(self),
826 self._aliased_insp._target.__name__,
827 )
829 def __str__(self) -> str:
830 return str(self._aliased_insp)
833@inspection._self_inspects
834class AliasedInsp(
835 ORMEntityColumnsClauseRole[_O],
836 ORMFromClauseRole,
837 HasCacheKey,
838 InspectionAttr,
839 MemoizedSlots,
840 inspection.Inspectable["AliasedInsp[_O]"],
841 Generic[_O],
842):
843 """Provide an inspection interface for an
844 :class:`.AliasedClass` object.
846 The :class:`.AliasedInsp` object is returned
847 given an :class:`.AliasedClass` using the
848 :func:`_sa.inspect` function::
850 from sqlalchemy import inspect
851 from sqlalchemy.orm import aliased
853 my_alias = aliased(MyMappedClass)
854 insp = inspect(my_alias)
856 Attributes on :class:`.AliasedInsp`
857 include:
859 * ``entity`` - the :class:`.AliasedClass` represented.
860 * ``mapper`` - the :class:`_orm.Mapper` mapping the underlying class.
861 * ``selectable`` - the :class:`_expression.Alias`
862 construct which ultimately
863 represents an aliased :class:`_schema.Table` or
864 :class:`_expression.Select`
865 construct.
866 * ``name`` - the name of the alias. Also is used as the attribute
867 name when returned in a result tuple from :class:`_query.Query`.
868 * ``with_polymorphic_mappers`` - collection of :class:`_orm.Mapper`
869 objects
870 indicating all those mappers expressed in the select construct
871 for the :class:`.AliasedClass`.
872 * ``polymorphic_on`` - an alternate column or SQL expression which
873 will be used as the "discriminator" for a polymorphic load.
875 .. seealso::
877 :ref:`inspection_toplevel`
879 """
881 __slots__ = (
882 "__weakref__",
883 "_weak_entity",
884 "mapper",
885 "selectable",
886 "name",
887 "_adapt_on_names",
888 "with_polymorphic_mappers",
889 "polymorphic_on",
890 "_use_mapper_path",
891 "_base_alias",
892 "represents_outer_join",
893 "persist_selectable",
894 "local_table",
895 "_is_with_polymorphic",
896 "_with_polymorphic_entities",
897 "_adapter",
898 "_target",
899 "__clause_element__",
900 "_memoized_values",
901 "_all_column_expressions",
902 "_nest_adapters",
903 )
905 _cache_key_traversal = [
906 ("name", visitors.ExtendedInternalTraversal.dp_string),
907 ("_adapt_on_names", visitors.ExtendedInternalTraversal.dp_boolean),
908 ("_use_mapper_path", visitors.ExtendedInternalTraversal.dp_boolean),
909 ("_target", visitors.ExtendedInternalTraversal.dp_inspectable),
910 ("selectable", visitors.ExtendedInternalTraversal.dp_clauseelement),
911 (
912 "with_polymorphic_mappers",
913 visitors.InternalTraversal.dp_has_cache_key_list,
914 ),
915 ("polymorphic_on", visitors.InternalTraversal.dp_clauseelement),
916 ]
918 mapper: Mapper[_O]
919 selectable: FromClause
920 _adapter: ORMAdapter
921 with_polymorphic_mappers: Sequence[Mapper[Any]]
922 _with_polymorphic_entities: Sequence[AliasedInsp[Any]]
924 _weak_entity: weakref.ref[AliasedClass[_O]]
925 """the AliasedClass that refers to this AliasedInsp"""
927 _target: Union[Type[_O], AliasedClass[_O]]
928 """the thing referenced by the AliasedClass/AliasedInsp.
930 In the vast majority of cases, this is the mapped class. However
931 it may also be another AliasedClass (alias of alias).
933 """
935 def __init__(
936 self,
937 entity: AliasedClass[_O],
938 inspected: _InternalEntityType[_O],
939 selectable: FromClause,
940 name: Optional[str],
941 with_polymorphic_mappers: Optional[Sequence[Mapper[Any]]],
942 polymorphic_on: Optional[ColumnElement[Any]],
943 _base_alias: Optional[AliasedInsp[Any]],
944 _use_mapper_path: bool,
945 adapt_on_names: bool,
946 represents_outer_join: bool,
947 nest_adapters: bool,
948 ):
949 mapped_class_or_ac = inspected.entity
950 mapper = inspected.mapper
952 self._weak_entity = weakref.ref(entity)
953 self.mapper = mapper
954 self.selectable = self.persist_selectable = self.local_table = (
955 selectable
956 )
957 self.name = name
958 self.polymorphic_on = polymorphic_on
959 self._base_alias = weakref.ref(_base_alias or self)
960 self._use_mapper_path = _use_mapper_path
961 self.represents_outer_join = represents_outer_join
962 self._nest_adapters = nest_adapters
964 if with_polymorphic_mappers:
965 self._is_with_polymorphic = True
966 self.with_polymorphic_mappers = with_polymorphic_mappers
967 self._with_polymorphic_entities = []
968 for poly in self.with_polymorphic_mappers:
969 if poly is not mapper:
970 ent = AliasedClass(
971 poly.class_,
972 selectable,
973 base_alias=self,
974 adapt_on_names=adapt_on_names,
975 use_mapper_path=_use_mapper_path,
976 )
978 setattr(self.entity, poly.class_.__name__, ent)
979 self._with_polymorphic_entities.append(ent._aliased_insp)
981 else:
982 self._is_with_polymorphic = False
983 self.with_polymorphic_mappers = [mapper]
985 self._adapter = ORMAdapter(
986 _TraceAdaptRole.ALIASED_INSP,
987 mapper,
988 selectable=selectable,
989 equivalents=mapper._equivalent_columns,
990 adapt_on_names=adapt_on_names,
991 anonymize_labels=True,
992 # make sure the adapter doesn't try to grab other tables that
993 # are not even the thing we are mapping, such as embedded
994 # selectables in subqueries or CTEs. See issue #6060
995 adapt_from_selectables={
996 m.selectable
997 for m in self.with_polymorphic_mappers
998 if not adapt_on_names
999 },
1000 limit_on_entity=False,
1001 )
1003 if nest_adapters:
1004 # supports "aliased class of aliased class" use case
1005 assert isinstance(inspected, AliasedInsp)
1006 self._adapter = inspected._adapter.wrap(self._adapter)
1008 self._adapt_on_names = adapt_on_names
1009 self._target = mapped_class_or_ac
1011 @property
1012 def _post_inspect(self): # type: ignore[override]
1013 self.mapper._check_configure()
1015 @classmethod
1016 def _alias_factory(
1017 cls,
1018 element: Union[_EntityType[_O], FromClause],
1019 alias: Optional[FromClause] = None,
1020 name: Optional[str] = None,
1021 flat: bool = False,
1022 adapt_on_names: bool = False,
1023 ) -> Union[AliasedClass[_O], FromClause]:
1024 if isinstance(element, GenerativeSelect):
1025 return coercions.expect(roles.FromClauseRole, element, flat=flat)
1026 elif isinstance(element, FromClause):
1027 if adapt_on_names:
1028 raise sa_exc.ArgumentError(
1029 "adapt_on_names only applies to ORM elements"
1030 )
1031 if name:
1032 return element.alias(name=name, flat=flat)
1033 else:
1034 return coercions.expect(
1035 roles.AnonymizedFromClauseRole, element, flat=flat
1036 )
1037 else:
1038 return AliasedClass(
1039 element,
1040 alias=alias,
1041 flat=flat,
1042 name=name,
1043 adapt_on_names=adapt_on_names,
1044 )
1046 @classmethod
1047 def _with_polymorphic_factory(
1048 cls,
1049 base: Union[Type[_O], Mapper[_O]],
1050 classes: Union[Literal["*"], Iterable[_EntityType[Any]]],
1051 selectable: Union[Literal[False, None], FromClause] = False,
1052 flat: bool = False,
1053 polymorphic_on: Optional[ColumnElement[Any]] = None,
1054 aliased: bool = False,
1055 innerjoin: bool = False,
1056 adapt_on_names: bool = False,
1057 name: Optional[str] = None,
1058 _use_mapper_path: bool = False,
1059 ) -> AliasedClass[_O]:
1060 primary_mapper = _class_to_mapper(base)
1062 if selectable not in (None, False) and flat:
1063 raise sa_exc.ArgumentError(
1064 "the 'flat' and 'selectable' arguments cannot be passed "
1065 "simultaneously to with_polymorphic()"
1066 )
1068 mappers, selectable = primary_mapper._with_polymorphic_args(
1069 classes, selectable, innerjoin=innerjoin
1070 )
1071 if aliased or flat:
1072 assert selectable is not None
1073 selectable = selectable._anonymous_fromclause(flat=flat)
1075 return AliasedClass(
1076 base,
1077 selectable,
1078 name=name,
1079 with_polymorphic_mappers=mappers,
1080 adapt_on_names=adapt_on_names,
1081 with_polymorphic_discriminator=polymorphic_on,
1082 use_mapper_path=_use_mapper_path,
1083 represents_outer_join=not innerjoin,
1084 )
1086 @property
1087 def entity(self) -> AliasedClass[_O]:
1088 # to eliminate reference cycles, the AliasedClass is held weakly.
1089 # this produces some situations where the AliasedClass gets lost,
1090 # particularly when one is created internally and only the AliasedInsp
1091 # is passed around.
1092 # to work around this case, we just generate a new one when we need
1093 # it, as it is a simple class with very little initial state on it.
1094 ent = self._weak_entity()
1095 if ent is None:
1096 ent = AliasedClass._reconstitute_from_aliased_insp(self)
1097 self._weak_entity = weakref.ref(ent)
1098 return ent
1100 is_aliased_class = True
1101 "always returns True"
1103 def _memoized_method___clause_element__(self) -> FromClause:
1104 return self.selectable._annotate(
1105 {
1106 "parentmapper": self.mapper,
1107 "parententity": self,
1108 "entity_namespace": self,
1109 }
1110 )._set_propagate_attrs(
1111 {"compile_state_plugin": "orm", "plugin_subject": self}
1112 )
1114 @property
1115 def entity_namespace(self) -> AliasedClass[_O]:
1116 return self.entity
1118 @property
1119 def class_(self) -> Type[_O]:
1120 """Return the mapped class ultimately represented by this
1121 :class:`.AliasedInsp`."""
1122 return self.mapper.class_
1124 @property
1125 def _path_registry(self) -> AbstractEntityRegistry:
1126 if self._use_mapper_path:
1127 return self.mapper._path_registry
1128 else:
1129 return PathRegistry.per_mapper(self)
1131 def __getstate__(self) -> Dict[str, Any]:
1132 return {
1133 "entity": self.entity,
1134 "mapper": self.mapper,
1135 "alias": self.selectable,
1136 "name": self.name,
1137 "adapt_on_names": self._adapt_on_names,
1138 "with_polymorphic_mappers": self.with_polymorphic_mappers,
1139 "with_polymorphic_discriminator": self.polymorphic_on,
1140 "base_alias": self._base_alias(),
1141 "use_mapper_path": self._use_mapper_path,
1142 "represents_outer_join": self.represents_outer_join,
1143 "nest_adapters": self._nest_adapters,
1144 }
1146 def __setstate__(self, state: Dict[str, Any]) -> None:
1147 self.__init__( # type: ignore
1148 state["entity"],
1149 state["mapper"],
1150 state["alias"],
1151 state["name"],
1152 state["with_polymorphic_mappers"],
1153 state["with_polymorphic_discriminator"],
1154 state["base_alias"],
1155 state["use_mapper_path"],
1156 state["adapt_on_names"],
1157 state["represents_outer_join"],
1158 state["nest_adapters"],
1159 )
1161 def _merge_with(self, other: AliasedInsp[_O]) -> AliasedInsp[_O]:
1162 # assert self._is_with_polymorphic
1163 # assert other._is_with_polymorphic
1165 primary_mapper = other.mapper
1167 assert self.mapper is primary_mapper
1169 our_classes = util.to_set(
1170 mp.class_ for mp in self.with_polymorphic_mappers
1171 )
1172 new_classes = {mp.class_ for mp in other.with_polymorphic_mappers}
1173 if our_classes == new_classes:
1174 return other
1175 else:
1176 classes = our_classes.union(new_classes)
1178 mappers, selectable = primary_mapper._with_polymorphic_args(
1179 classes, None, innerjoin=not other.represents_outer_join
1180 )
1181 selectable = selectable._anonymous_fromclause(flat=True)
1182 return AliasedClass(
1183 primary_mapper,
1184 selectable,
1185 with_polymorphic_mappers=mappers,
1186 with_polymorphic_discriminator=other.polymorphic_on,
1187 use_mapper_path=other._use_mapper_path,
1188 represents_outer_join=other.represents_outer_join,
1189 )._aliased_insp
1191 def _adapt_element(
1192 self, expr: _ORMCOLEXPR, key: Optional[str] = None
1193 ) -> _ORMCOLEXPR:
1194 assert isinstance(expr, ColumnElement)
1195 d: Dict[str, Any] = {
1196 "parententity": self,
1197 "parentmapper": self.mapper,
1198 }
1199 if key:
1200 d["proxy_key"] = key
1202 # IMO mypy should see this one also as returning the same type
1203 # we put into it, but it's not
1204 return (
1205 self._adapter.traverse(expr)
1206 ._annotate(d)
1207 ._set_propagate_attrs(
1208 {"compile_state_plugin": "orm", "plugin_subject": self}
1209 )
1210 )
1212 if TYPE_CHECKING:
1213 # establish compatibility with the _ORMAdapterProto protocol,
1214 # which in turn is compatible with _CoreAdapterProto.
1216 def _orm_adapt_element(
1217 self,
1218 obj: _CE,
1219 key: Optional[str] = None,
1220 ) -> _CE: ...
1222 else:
1223 _orm_adapt_element = _adapt_element
1225 def _entity_for_mapper(self, mapper):
1226 self_poly = self.with_polymorphic_mappers
1227 if mapper in self_poly:
1228 if mapper is self.mapper:
1229 return self
1230 else:
1231 return getattr(
1232 self.entity, mapper.class_.__name__
1233 )._aliased_insp
1234 elif mapper.isa(self.mapper):
1235 return self
1236 else:
1237 assert False, "mapper %s doesn't correspond to %s" % (mapper, self)
1239 def _memoized_attr__get_clause(self):
1240 onclause, replacemap = self.mapper._get_clause
1241 return (
1242 self._adapter.traverse(onclause),
1243 {
1244 self._adapter.traverse(col): param
1245 for col, param in replacemap.items()
1246 },
1247 )
1249 def _memoized_attr__memoized_values(self):
1250 return {}
1252 def _memoized_attr__all_column_expressions(self):
1253 if self._is_with_polymorphic:
1254 cols_plus_keys = self.mapper._columns_plus_keys(
1255 [ent.mapper for ent in self._with_polymorphic_entities]
1256 )
1257 else:
1258 cols_plus_keys = self.mapper._columns_plus_keys()
1260 cols_plus_keys = [
1261 (key, self._adapt_element(col)) for key, col in cols_plus_keys
1262 ]
1264 return ColumnCollection(cols_plus_keys)
1266 def _memo(self, key, callable_, *args, **kw):
1267 if key in self._memoized_values:
1268 return self._memoized_values[key]
1269 else:
1270 self._memoized_values[key] = value = callable_(*args, **kw)
1271 return value
1273 def __repr__(self):
1274 if self.with_polymorphic_mappers:
1275 with_poly = "(%s)" % ", ".join(
1276 mp.class_.__name__ for mp in self.with_polymorphic_mappers
1277 )
1278 else:
1279 with_poly = ""
1280 return "<AliasedInsp at 0x%x; %s%s>" % (
1281 id(self),
1282 self.class_.__name__,
1283 with_poly,
1284 )
1286 def __str__(self):
1287 if self._is_with_polymorphic:
1288 return "with_polymorphic(%s, [%s])" % (
1289 self._target.__name__,
1290 ", ".join(
1291 mp.class_.__name__
1292 for mp in self.with_polymorphic_mappers
1293 if mp is not self.mapper
1294 ),
1295 )
1296 else:
1297 return "aliased(%s)" % (self._target.__name__,)
1300class _WrapUserEntity:
1301 """A wrapper used within the loader_criteria lambda caller so that
1302 we can bypass declared_attr descriptors on unmapped mixins, which
1303 normally emit a warning for such use.
1305 might also be useful for other per-lambda instrumentations should
1306 the need arise.
1308 """
1310 __slots__ = ("subject",)
1312 def __init__(self, subject):
1313 self.subject = subject
1315 @util.preload_module("sqlalchemy.orm.decl_api")
1316 def __getattribute__(self, name):
1317 decl_api = util.preloaded.orm.decl_api
1319 subject = object.__getattribute__(self, "subject")
1320 if name in subject.__dict__ and isinstance(
1321 subject.__dict__[name], decl_api.declared_attr
1322 ):
1323 return subject.__dict__[name].fget(subject)
1324 else:
1325 return getattr(subject, name)
1328class LoaderCriteriaOption(CriteriaOption):
1329 """Add additional WHERE criteria to the load for all occurrences of
1330 a particular entity.
1332 :class:`_orm.LoaderCriteriaOption` is invoked using the
1333 :func:`_orm.with_loader_criteria` function; see that function for
1334 details.
1336 .. versionadded:: 1.4
1338 """
1340 __slots__ = (
1341 "root_entity",
1342 "entity",
1343 "deferred_where_criteria",
1344 "where_criteria",
1345 "_where_crit_orig",
1346 "include_aliases",
1347 "propagate_to_loaders",
1348 )
1350 _traverse_internals = [
1351 ("root_entity", visitors.ExtendedInternalTraversal.dp_plain_obj),
1352 ("entity", visitors.ExtendedInternalTraversal.dp_has_cache_key),
1353 ("where_criteria", visitors.InternalTraversal.dp_clauseelement),
1354 ("include_aliases", visitors.InternalTraversal.dp_boolean),
1355 ("propagate_to_loaders", visitors.InternalTraversal.dp_boolean),
1356 ]
1358 root_entity: Optional[Type[Any]]
1359 entity: Optional[_InternalEntityType[Any]]
1360 where_criteria: Union[ColumnElement[bool], lambdas.DeferredLambdaElement]
1361 deferred_where_criteria: bool
1362 include_aliases: bool
1363 propagate_to_loaders: bool
1365 _where_crit_orig: Any
1367 def __init__(
1368 self,
1369 entity_or_base: _EntityType[Any],
1370 where_criteria: Union[
1371 _ColumnExpressionArgument[bool],
1372 Callable[[Any], _ColumnExpressionArgument[bool]],
1373 ],
1374 loader_only: bool = False,
1375 include_aliases: bool = False,
1376 propagate_to_loaders: bool = True,
1377 track_closure_variables: bool = True,
1378 ):
1379 entity = cast(
1380 "_InternalEntityType[Any]",
1381 inspection.inspect(entity_or_base, False),
1382 )
1383 if entity is None:
1384 self.root_entity = cast("Type[Any]", entity_or_base)
1385 self.entity = None
1386 else:
1387 self.root_entity = None
1388 self.entity = entity
1390 self._where_crit_orig = where_criteria
1391 if callable(where_criteria):
1392 if self.root_entity is not None:
1393 wrap_entity = self.root_entity
1394 else:
1395 assert entity is not None
1396 wrap_entity = entity.entity
1398 self.deferred_where_criteria = True
1399 self.where_criteria = lambdas.DeferredLambdaElement(
1400 where_criteria,
1401 roles.WhereHavingRole,
1402 lambda_args=(_WrapUserEntity(wrap_entity),),
1403 opts=lambdas.LambdaOptions(
1404 track_closure_variables=track_closure_variables
1405 ),
1406 )
1407 else:
1408 self.deferred_where_criteria = False
1409 self.where_criteria = coercions.expect(
1410 roles.WhereHavingRole, where_criteria
1411 )
1413 self.include_aliases = include_aliases
1414 self.propagate_to_loaders = propagate_to_loaders
1416 @classmethod
1417 def _unreduce(
1418 cls, entity, where_criteria, include_aliases, propagate_to_loaders
1419 ):
1420 return LoaderCriteriaOption(
1421 entity,
1422 where_criteria,
1423 include_aliases=include_aliases,
1424 propagate_to_loaders=propagate_to_loaders,
1425 )
1427 def __reduce__(self):
1428 return (
1429 LoaderCriteriaOption._unreduce,
1430 (
1431 self.entity.class_ if self.entity else self.root_entity,
1432 self._where_crit_orig,
1433 self.include_aliases,
1434 self.propagate_to_loaders,
1435 ),
1436 )
1438 def _all_mappers(self) -> Iterator[Mapper[Any]]:
1439 if self.entity:
1440 yield from self.entity.mapper.self_and_descendants
1441 else:
1442 assert self.root_entity
1443 stack = list(self.root_entity.__subclasses__())
1444 while stack:
1445 subclass = stack.pop(0)
1446 ent = cast(
1447 "_InternalEntityType[Any]",
1448 inspection.inspect(subclass, raiseerr=False),
1449 )
1450 if ent:
1451 yield from ent.mapper.self_and_descendants
1452 else:
1453 stack.extend(subclass.__subclasses__())
1455 def _should_include(self, compile_state: ORMCompileState) -> bool:
1456 if (
1457 compile_state.select_statement._annotations.get(
1458 "for_loader_criteria", None
1459 )
1460 is self
1461 ):
1462 return False
1463 return True
1465 def _resolve_where_criteria(
1466 self, ext_info: _InternalEntityType[Any]
1467 ) -> ColumnElement[bool]:
1468 if self.deferred_where_criteria:
1469 crit = cast(
1470 "ColumnElement[bool]",
1471 self.where_criteria._resolve_with_args(ext_info.entity),
1472 )
1473 else:
1474 crit = self.where_criteria # type: ignore
1475 assert isinstance(crit, ColumnElement)
1476 return sql_util._deep_annotate(
1477 crit,
1478 {"for_loader_criteria": self},
1479 detect_subquery_cols=True,
1480 ind_cols_on_fromclause=True,
1481 )
1483 def process_compile_state_replaced_entities(
1484 self,
1485 compile_state: ORMCompileState,
1486 mapper_entities: Iterable[_MapperEntity],
1487 ) -> None:
1488 self.process_compile_state(compile_state)
1490 def process_compile_state(self, compile_state: ORMCompileState) -> None:
1491 """Apply a modification to a given :class:`.CompileState`."""
1493 # if options to limit the criteria to immediate query only,
1494 # use compile_state.attributes instead
1496 self.get_global_criteria(compile_state.global_attributes)
1498 def get_global_criteria(self, attributes: Dict[Any, Any]) -> None:
1499 for mp in self._all_mappers():
1500 load_criteria = attributes.setdefault(
1501 ("additional_entity_criteria", mp), []
1502 )
1504 load_criteria.append(self)
1507inspection._inspects(AliasedClass)(lambda target: target._aliased_insp)
1510@inspection._inspects(type)
1511def _inspect_mc(
1512 class_: Type[_O],
1513) -> Optional[Mapper[_O]]:
1514 try:
1515 class_manager = opt_manager_of_class(class_)
1516 if class_manager is None or not class_manager.is_mapped:
1517 return None
1518 mapper = class_manager.mapper
1519 except exc.NO_STATE:
1520 return None
1521 else:
1522 return mapper
1525GenericAlias = type(List[Any])
1528@inspection._inspects(GenericAlias)
1529def _inspect_generic_alias(
1530 class_: Type[_O],
1531) -> Optional[Mapper[_O]]:
1532 origin = cast("Type[_O]", get_origin(class_))
1533 return _inspect_mc(origin)
1536@inspection._self_inspects
1537class Bundle(
1538 ORMColumnsClauseRole[_T],
1539 SupportsCloneAnnotations,
1540 MemoizedHasCacheKey,
1541 inspection.Inspectable["Bundle[_T]"],
1542 InspectionAttr,
1543):
1544 """A grouping of SQL expressions that are returned by a :class:`.Query`
1545 under one namespace.
1547 The :class:`.Bundle` essentially allows nesting of the tuple-based
1548 results returned by a column-oriented :class:`_query.Query` object.
1549 It also
1550 is extensible via simple subclassing, where the primary capability
1551 to override is that of how the set of expressions should be returned,
1552 allowing post-processing as well as custom return types, without
1553 involving ORM identity-mapped classes.
1555 .. seealso::
1557 :ref:`bundles`
1560 """
1562 single_entity = False
1563 """If True, queries for a single Bundle will be returned as a single
1564 entity, rather than an element within a keyed tuple."""
1566 is_clause_element = False
1568 is_mapper = False
1570 is_aliased_class = False
1572 is_bundle = True
1574 _propagate_attrs: _PropagateAttrsType = util.immutabledict()
1576 proxy_set = util.EMPTY_SET
1578 exprs: List[_ColumnsClauseElement]
1580 def __init__(
1581 self, name: str, *exprs: _ColumnExpressionArgument[Any], **kw: Any
1582 ):
1583 r"""Construct a new :class:`.Bundle`.
1585 e.g.::
1587 bn = Bundle("mybundle", MyClass.x, MyClass.y)
1589 for row in session.query(bn).filter(bn.c.x == 5).filter(bn.c.y == 4):
1590 print(row.mybundle.x, row.mybundle.y)
1592 :param name: name of the bundle.
1593 :param \*exprs: columns or SQL expressions comprising the bundle.
1594 :param single_entity=False: if True, rows for this :class:`.Bundle`
1595 can be returned as a "single entity" outside of any enclosing tuple
1596 in the same manner as a mapped entity.
1598 """ # noqa: E501
1599 self.name = self._label = name
1600 coerced_exprs = [
1601 coercions.expect(
1602 roles.ColumnsClauseRole, expr, apply_propagate_attrs=self
1603 )
1604 for expr in exprs
1605 ]
1606 self.exprs = coerced_exprs
1608 self.c = self.columns = ColumnCollection(
1609 (getattr(col, "key", col._label), col)
1610 for col in [e._annotations.get("bundle", e) for e in coerced_exprs]
1611 ).as_readonly()
1612 self.single_entity = kw.pop("single_entity", self.single_entity)
1614 def _gen_cache_key(
1615 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
1616 ) -> Tuple[Any, ...]:
1617 return (self.__class__, self.name, self.single_entity) + tuple(
1618 [expr._gen_cache_key(anon_map, bindparams) for expr in self.exprs]
1619 )
1621 @property
1622 def mapper(self) -> Optional[Mapper[Any]]:
1623 mp: Optional[Mapper[Any]] = self.exprs[0]._annotations.get(
1624 "parentmapper", None
1625 )
1626 return mp
1628 @property
1629 def entity(self) -> Optional[_InternalEntityType[Any]]:
1630 ie: Optional[_InternalEntityType[Any]] = self.exprs[
1631 0
1632 ]._annotations.get("parententity", None)
1633 return ie
1635 @property
1636 def entity_namespace(
1637 self,
1638 ) -> ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]:
1639 return self.c
1641 columns: ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]
1643 """A namespace of SQL expressions referred to by this :class:`.Bundle`.
1645 e.g.::
1647 bn = Bundle("mybundle", MyClass.x, MyClass.y)
1649 q = sess.query(bn).filter(bn.c.x == 5)
1651 Nesting of bundles is also supported::
1653 b1 = Bundle(
1654 "b1",
1655 Bundle("b2", MyClass.a, MyClass.b),
1656 Bundle("b3", MyClass.x, MyClass.y),
1657 )
1659 q = sess.query(b1).filter(b1.c.b2.c.a == 5).filter(b1.c.b3.c.y == 9)
1661 .. seealso::
1663 :attr:`.Bundle.c`
1665 """ # noqa: E501
1667 c: ReadOnlyColumnCollection[str, KeyedColumnElement[Any]]
1668 """An alias for :attr:`.Bundle.columns`."""
1670 def _clone(self, **kw):
1671 cloned = self.__class__.__new__(self.__class__)
1672 cloned.__dict__.update(self.__dict__)
1673 return cloned
1675 def __clause_element__(self):
1676 # ensure existing entity_namespace remains
1677 annotations = {"bundle": self, "entity_namespace": self}
1678 annotations.update(self._annotations)
1680 plugin_subject = self.exprs[0]._propagate_attrs.get(
1681 "plugin_subject", self.entity
1682 )
1683 return (
1684 expression.ClauseList(
1685 _literal_as_text_role=roles.ColumnsClauseRole,
1686 group=False,
1687 *[e._annotations.get("bundle", e) for e in self.exprs],
1688 )
1689 ._annotate(annotations)
1690 ._set_propagate_attrs(
1691 # the Bundle *must* use the orm plugin no matter what. the
1692 # subject can be None but it's much better if it's not.
1693 {
1694 "compile_state_plugin": "orm",
1695 "plugin_subject": plugin_subject,
1696 }
1697 )
1698 )
1700 @property
1701 def clauses(self):
1702 return self.__clause_element__().clauses
1704 def label(self, name):
1705 """Provide a copy of this :class:`.Bundle` passing a new label."""
1707 cloned = self._clone()
1708 cloned.name = name
1709 return cloned
1711 def create_row_processor(
1712 self,
1713 query: Select[Any],
1714 procs: Sequence[Callable[[Row[Any]], Any]],
1715 labels: Sequence[str],
1716 ) -> Callable[[Row[Any]], Any]:
1717 """Produce the "row processing" function for this :class:`.Bundle`.
1719 May be overridden by subclasses to provide custom behaviors when
1720 results are fetched. The method is passed the statement object and a
1721 set of "row processor" functions at query execution time; these
1722 processor functions when given a result row will return the individual
1723 attribute value, which can then be adapted into any kind of return data
1724 structure.
1726 The example below illustrates replacing the usual :class:`.Row`
1727 return structure with a straight Python dictionary::
1729 from sqlalchemy.orm import Bundle
1732 class DictBundle(Bundle):
1733 def create_row_processor(self, query, procs, labels):
1734 "Override create_row_processor to return values as dictionaries"
1736 def proc(row):
1737 return dict(zip(labels, (proc(row) for proc in procs)))
1739 return proc
1741 A result from the above :class:`_orm.Bundle` will return dictionary
1742 values::
1744 bn = DictBundle("mybundle", MyClass.data1, MyClass.data2)
1745 for row in session.execute(select(bn)).where(bn.c.data1 == "d1"):
1746 print(row.mybundle["data1"], row.mybundle["data2"])
1748 """ # noqa: E501
1749 keyed_tuple = result_tuple(labels, [() for l in labels])
1751 def proc(row: Row[Any]) -> Any:
1752 return keyed_tuple([proc(row) for proc in procs])
1754 return proc
1757def _orm_annotate(element: _SA, exclude: Optional[Any] = None) -> _SA:
1758 """Deep copy the given ClauseElement, annotating each element with the
1759 "_orm_adapt" flag.
1761 Elements within the exclude collection will be cloned but not annotated.
1763 """
1764 return sql_util._deep_annotate(element, {"_orm_adapt": True}, exclude)
1767def _orm_deannotate(element: _SA) -> _SA:
1768 """Remove annotations that link a column to a particular mapping.
1770 Note this doesn't affect "remote" and "foreign" annotations
1771 passed by the :func:`_orm.foreign` and :func:`_orm.remote`
1772 annotators.
1774 """
1776 return sql_util._deep_deannotate(
1777 element, values=("_orm_adapt", "parententity")
1778 )
1781def _orm_full_deannotate(element: _SA) -> _SA:
1782 return sql_util._deep_deannotate(element)
1785class _ORMJoin(expression.Join):
1786 """Extend Join to support ORM constructs as input."""
1788 __visit_name__ = expression.Join.__visit_name__
1790 inherit_cache = True
1792 def __init__(
1793 self,
1794 left: _FromClauseArgument,
1795 right: _FromClauseArgument,
1796 onclause: Optional[_OnClauseArgument] = None,
1797 isouter: bool = False,
1798 full: bool = False,
1799 _left_memo: Optional[Any] = None,
1800 _right_memo: Optional[Any] = None,
1801 _extra_criteria: Tuple[ColumnElement[bool], ...] = (),
1802 ):
1803 left_info = cast(
1804 "Union[FromClause, _InternalEntityType[Any]]",
1805 inspection.inspect(left),
1806 )
1808 right_info = cast(
1809 "Union[FromClause, _InternalEntityType[Any]]",
1810 inspection.inspect(right),
1811 )
1812 adapt_to = right_info.selectable
1814 # used by joined eager loader
1815 self._left_memo = _left_memo
1816 self._right_memo = _right_memo
1818 if isinstance(onclause, attributes.QueryableAttribute):
1819 if TYPE_CHECKING:
1820 assert isinstance(
1821 onclause.comparator, RelationshipProperty.Comparator
1822 )
1823 on_selectable = onclause.comparator._source_selectable()
1824 prop = onclause.property
1825 _extra_criteria += onclause._extra_criteria
1826 elif isinstance(onclause, MapperProperty):
1827 # used internally by joined eager loader...possibly not ideal
1828 prop = onclause
1829 on_selectable = prop.parent.selectable
1830 else:
1831 prop = None
1832 on_selectable = None
1834 left_selectable = left_info.selectable
1835 if prop:
1836 adapt_from: Optional[FromClause]
1837 if sql_util.clause_is_present(on_selectable, left_selectable):
1838 adapt_from = on_selectable
1839 else:
1840 assert isinstance(left_selectable, FromClause)
1841 adapt_from = left_selectable
1843 (
1844 pj,
1845 sj,
1846 source,
1847 dest,
1848 secondary,
1849 target_adapter,
1850 ) = prop._create_joins(
1851 source_selectable=adapt_from,
1852 dest_selectable=adapt_to,
1853 source_polymorphic=True,
1854 of_type_entity=right_info,
1855 alias_secondary=True,
1856 extra_criteria=_extra_criteria,
1857 )
1859 if sj is not None:
1860 if isouter:
1861 # note this is an inner join from secondary->right
1862 right = sql.join(secondary, right, sj)
1863 onclause = pj
1864 else:
1865 left = sql.join(left, secondary, pj, isouter)
1866 onclause = sj
1867 else:
1868 onclause = pj
1870 self._target_adapter = target_adapter
1872 # we don't use the normal coercions logic for _ORMJoin
1873 # (probably should), so do some gymnastics to get the entity.
1874 # logic here is for #8721, which was a major bug in 1.4
1875 # for almost two years, not reported/fixed until 1.4.43 (!)
1876 if is_selectable(left_info):
1877 parententity = left_selectable._annotations.get(
1878 "parententity", None
1879 )
1880 elif insp_is_mapper(left_info) or insp_is_aliased_class(left_info):
1881 parententity = left_info
1882 else:
1883 parententity = None
1885 if parententity is not None:
1886 self._annotations = self._annotations.union(
1887 {"parententity": parententity}
1888 )
1890 augment_onclause = bool(_extra_criteria) and not prop
1891 expression.Join.__init__(self, left, right, onclause, isouter, full)
1893 assert self.onclause is not None
1895 if augment_onclause:
1896 self.onclause &= sql.and_(*_extra_criteria)
1898 if (
1899 not prop
1900 and getattr(right_info, "mapper", None)
1901 and right_info.mapper.single # type: ignore
1902 ):
1903 right_info = cast("_InternalEntityType[Any]", right_info)
1904 # if single inheritance target and we are using a manual
1905 # or implicit ON clause, augment it the same way we'd augment the
1906 # WHERE.
1907 single_crit = right_info.mapper._single_table_criterion
1908 if single_crit is not None:
1909 if insp_is_aliased_class(right_info):
1910 single_crit = right_info._adapter.traverse(single_crit)
1911 self.onclause = self.onclause & single_crit
1913 def _splice_into_center(self, other):
1914 """Splice a join into the center.
1916 Given join(a, b) and join(b, c), return join(a, b).join(c)
1918 """
1919 leftmost = other
1920 while isinstance(leftmost, sql.Join):
1921 leftmost = leftmost.left
1923 assert self.right is leftmost
1925 left = _ORMJoin(
1926 self.left,
1927 other.left,
1928 self.onclause,
1929 isouter=self.isouter,
1930 _left_memo=self._left_memo,
1931 _right_memo=other._left_memo._path_registry,
1932 )
1934 return _ORMJoin(
1935 left,
1936 other.right,
1937 other.onclause,
1938 isouter=other.isouter,
1939 _right_memo=other._right_memo,
1940 )
1942 def join(
1943 self,
1944 right: _FromClauseArgument,
1945 onclause: Optional[_OnClauseArgument] = None,
1946 isouter: bool = False,
1947 full: bool = False,
1948 ) -> _ORMJoin:
1949 return _ORMJoin(self, right, onclause, full=full, isouter=isouter)
1951 def outerjoin(
1952 self,
1953 right: _FromClauseArgument,
1954 onclause: Optional[_OnClauseArgument] = None,
1955 full: bool = False,
1956 ) -> _ORMJoin:
1957 return _ORMJoin(self, right, onclause, isouter=True, full=full)
1960def with_parent(
1961 instance: object,
1962 prop: attributes.QueryableAttribute[Any],
1963 from_entity: Optional[_EntityType[Any]] = None,
1964) -> ColumnElement[bool]:
1965 """Create filtering criterion that relates this query's primary entity
1966 to the given related instance, using established
1967 :func:`_orm.relationship()`
1968 configuration.
1970 E.g.::
1972 stmt = select(Address).where(with_parent(some_user, User.addresses))
1974 The SQL rendered is the same as that rendered when a lazy loader
1975 would fire off from the given parent on that attribute, meaning
1976 that the appropriate state is taken from the parent object in
1977 Python without the need to render joins to the parent table
1978 in the rendered statement.
1980 The given property may also make use of :meth:`_orm.PropComparator.of_type`
1981 to indicate the left side of the criteria::
1984 a1 = aliased(Address)
1985 a2 = aliased(Address)
1986 stmt = select(a1, a2).where(with_parent(u1, User.addresses.of_type(a2)))
1988 The above use is equivalent to using the
1989 :func:`_orm.with_parent.from_entity` argument::
1991 a1 = aliased(Address)
1992 a2 = aliased(Address)
1993 stmt = select(a1, a2).where(
1994 with_parent(u1, User.addresses, from_entity=a2)
1995 )
1997 :param instance:
1998 An instance which has some :func:`_orm.relationship`.
2000 :param property:
2001 Class-bound attribute, which indicates
2002 what relationship from the instance should be used to reconcile the
2003 parent/child relationship.
2005 :param from_entity:
2006 Entity in which to consider as the left side. This defaults to the
2007 "zero" entity of the :class:`_query.Query` itself.
2009 .. versionadded:: 1.2
2011 """ # noqa: E501
2012 prop_t: RelationshipProperty[Any]
2014 if isinstance(prop, str):
2015 raise sa_exc.ArgumentError(
2016 "with_parent() accepts class-bound mapped attributes, not strings"
2017 )
2018 elif isinstance(prop, attributes.QueryableAttribute):
2019 if prop._of_type:
2020 from_entity = prop._of_type
2021 mapper_property = prop.property
2022 if mapper_property is None or not prop_is_relationship(
2023 mapper_property
2024 ):
2025 raise sa_exc.ArgumentError(
2026 f"Expected relationship property for with_parent(), "
2027 f"got {mapper_property}"
2028 )
2029 prop_t = mapper_property
2030 else:
2031 prop_t = prop
2033 return prop_t._with_parent(instance, from_entity=from_entity)
2036def has_identity(object_: object) -> bool:
2037 """Return True if the given object has a database
2038 identity.
2040 This typically corresponds to the object being
2041 in either the persistent or detached state.
2043 .. seealso::
2045 :func:`.was_deleted`
2047 """
2048 state = attributes.instance_state(object_)
2049 return state.has_identity
2052def was_deleted(object_: object) -> bool:
2053 """Return True if the given object was deleted
2054 within a session flush.
2056 This is regardless of whether or not the object is
2057 persistent or detached.
2059 .. seealso::
2061 :attr:`.InstanceState.was_deleted`
2063 """
2065 state = attributes.instance_state(object_)
2066 return state.was_deleted
2069def _entity_corresponds_to(
2070 given: _InternalEntityType[Any], entity: _InternalEntityType[Any]
2071) -> bool:
2072 """determine if 'given' corresponds to 'entity', in terms
2073 of an entity passed to Query that would match the same entity
2074 being referred to elsewhere in the query.
2076 """
2077 if insp_is_aliased_class(entity):
2078 if insp_is_aliased_class(given):
2079 if entity._base_alias() is given._base_alias():
2080 return True
2081 return False
2082 elif insp_is_aliased_class(given):
2083 if given._use_mapper_path:
2084 return entity in given.with_polymorphic_mappers
2085 else:
2086 return entity is given
2088 assert insp_is_mapper(given)
2089 return entity.common_parent(given)
2092def _entity_corresponds_to_use_path_impl(
2093 given: _InternalEntityType[Any], entity: _InternalEntityType[Any]
2094) -> bool:
2095 """determine if 'given' corresponds to 'entity', in terms
2096 of a path of loader options where a mapped attribute is taken to
2097 be a member of a parent entity.
2099 e.g.::
2101 someoption(A).someoption(A.b) # -> fn(A, A) -> True
2102 someoption(A).someoption(C.d) # -> fn(A, C) -> False
2104 a1 = aliased(A)
2105 someoption(a1).someoption(A.b) # -> fn(a1, A) -> False
2106 someoption(a1).someoption(a1.b) # -> fn(a1, a1) -> True
2108 wp = with_polymorphic(A, [A1, A2])
2109 someoption(wp).someoption(A1.foo) # -> fn(wp, A1) -> False
2110 someoption(wp).someoption(wp.A1.foo) # -> fn(wp, wp.A1) -> True
2112 """
2113 if insp_is_aliased_class(given):
2114 return (
2115 insp_is_aliased_class(entity)
2116 and not entity._use_mapper_path
2117 and (given is entity or entity in given._with_polymorphic_entities)
2118 )
2119 elif not insp_is_aliased_class(entity):
2120 return given.isa(entity.mapper)
2121 else:
2122 return (
2123 entity._use_mapper_path
2124 and given in entity.with_polymorphic_mappers
2125 )
2128def _entity_isa(given: _InternalEntityType[Any], mapper: Mapper[Any]) -> bool:
2129 """determine if 'given' "is a" mapper, in terms of the given
2130 would load rows of type 'mapper'.
2132 """
2133 if given.is_aliased_class:
2134 return mapper in given.with_polymorphic_mappers or given.mapper.isa(
2135 mapper
2136 )
2137 elif given.with_polymorphic_mappers:
2138 return mapper in given.with_polymorphic_mappers or given.isa(mapper)
2139 else:
2140 return given.isa(mapper)
2143def _getitem(iterable_query: Query[Any], item: Any) -> Any:
2144 """calculate __getitem__ in terms of an iterable query object
2145 that also has a slice() method.
2147 """
2149 def _no_negative_indexes():
2150 raise IndexError(
2151 "negative indexes are not accepted by SQL "
2152 "index / slice operators"
2153 )
2155 if isinstance(item, slice):
2156 start, stop, step = util.decode_slice(item)
2158 if (
2159 isinstance(stop, int)
2160 and isinstance(start, int)
2161 and stop - start <= 0
2162 ):
2163 return []
2165 elif (isinstance(start, int) and start < 0) or (
2166 isinstance(stop, int) and stop < 0
2167 ):
2168 _no_negative_indexes()
2170 res = iterable_query.slice(start, stop)
2171 if step is not None:
2172 return list(res)[None : None : item.step]
2173 else:
2174 return list(res)
2175 else:
2176 if item == -1:
2177 _no_negative_indexes()
2178 else:
2179 return list(iterable_query[item : item + 1])[0]
2182def _is_mapped_annotation(
2183 raw_annotation: _AnnotationScanType,
2184 cls: Type[Any],
2185 originating_cls: Type[Any],
2186) -> bool:
2187 try:
2188 annotated = de_stringify_annotation(
2189 cls, raw_annotation, originating_cls.__module__
2190 )
2191 except NameError:
2192 # in most cases, at least within our own tests, we can raise
2193 # here, which is more accurate as it prevents us from returning
2194 # false negatives. However, in the real world, try to avoid getting
2195 # involved with end-user annotations that have nothing to do with us.
2196 # see issue #8888 where we bypass using this function in the case
2197 # that we want to detect an unresolvable Mapped[] type.
2198 return False
2199 else:
2200 return is_origin_of_cls(annotated, _MappedAnnotationBase)
2203class _CleanupError(Exception):
2204 pass
2207def _cleanup_mapped_str_annotation(
2208 annotation: str, originating_module: str
2209) -> str:
2210 # fix up an annotation that comes in as the form:
2211 # 'Mapped[List[Address]]' so that it instead looks like:
2212 # 'Mapped[List["Address"]]' , which will allow us to get
2213 # "Address" as a string
2215 # additionally, resolve symbols for these names since this is where
2216 # we'd have to do it
2218 inner: Optional[Match[str]]
2220 mm = re.match(r"^([^ \|]+?)\[(.+)\]$", annotation)
2222 if not mm:
2223 return annotation
2225 # ticket #8759. Resolve the Mapped name to a real symbol.
2226 # originally this just checked the name.
2227 try:
2228 obj = eval_name_only(mm.group(1), originating_module)
2229 except NameError as ne:
2230 raise _CleanupError(
2231 f'For annotation "{annotation}", could not resolve '
2232 f'container type "{mm.group(1)}". '
2233 "Please ensure this type is imported at the module level "
2234 "outside of TYPE_CHECKING blocks"
2235 ) from ne
2237 if obj is typing.ClassVar:
2238 real_symbol = "ClassVar"
2239 else:
2240 try:
2241 if issubclass(obj, _MappedAnnotationBase):
2242 real_symbol = obj.__name__
2243 else:
2244 return annotation
2245 except TypeError:
2246 # avoid isinstance(obj, type) check, just catch TypeError
2247 return annotation
2249 # note: if one of the codepaths above didn't define real_symbol and
2250 # then didn't return, real_symbol raises UnboundLocalError
2251 # which is actually a NameError, and the calling routines don't
2252 # notice this since they are catching NameError anyway. Just in case
2253 # this is being modified in the future, something to be aware of.
2255 stack = []
2256 inner = mm
2257 while True:
2258 stack.append(real_symbol if mm is inner else inner.group(1))
2259 g2 = inner.group(2)
2260 inner = re.match(r"^([^ \|]+?)\[(.+)\]$", g2)
2261 if inner is None:
2262 stack.append(g2)
2263 break
2265 # stacks we want to rewrite, that is, quote the last entry which
2266 # we think is a relationship class name:
2267 #
2268 # ['Mapped', 'List', 'Address']
2269 # ['Mapped', 'A']
2270 #
2271 # stacks we dont want to rewrite, which are generally MappedColumn
2272 # use cases:
2273 #
2274 # ['Mapped', "'Optional[Dict[str, str]]'"]
2275 # ['Mapped', 'dict[str, str] | None']
2277 if (
2278 # avoid already quoted symbols such as
2279 # ['Mapped', "'Optional[Dict[str, str]]'"]
2280 not re.match(r"""^["'].*["']$""", stack[-1])
2281 # avoid further generics like Dict[] such as
2282 # ['Mapped', 'dict[str, str] | None'],
2283 # ['Mapped', 'list[int] | list[str]'],
2284 # ['Mapped', 'Union[list[int], list[str]]'],
2285 and not re.search(r"[\[\]]", stack[-1])
2286 ):
2287 stripchars = "\"' "
2288 stack[-1] = ", ".join(
2289 f'"{elem.strip(stripchars)}"' for elem in stack[-1].split(",")
2290 )
2292 annotation = "[".join(stack) + ("]" * (len(stack) - 1))
2294 return annotation
2297def _extract_mapped_subtype(
2298 raw_annotation: Optional[_AnnotationScanType],
2299 cls: type,
2300 originating_module: str,
2301 key: str,
2302 attr_cls: Type[Any],
2303 required: bool,
2304 is_dataclass_field: bool,
2305 expect_mapped: bool = True,
2306 raiseerr: bool = True,
2307) -> Optional[Tuple[Union[_AnnotationScanType, str], Optional[type]]]:
2308 """given an annotation, figure out if it's ``Mapped[something]`` and if
2309 so, return the ``something`` part.
2311 Includes error raise scenarios and other options.
2313 """
2315 if raw_annotation is None:
2316 if required:
2317 raise orm_exc.MappedAnnotationError(
2318 f"Python typing annotation is required for attribute "
2319 f'"{cls.__name__}.{key}" when primary argument(s) for '
2320 f'"{attr_cls.__name__}" construct are None or not present'
2321 )
2322 return None
2324 try:
2325 # destringify the "outside" of the annotation. note we are not
2326 # adding include_generic so it will *not* dig into generic contents,
2327 # which will remain as ForwardRef or plain str under future annotations
2328 # mode. The full destringify happens later when mapped_column goes
2329 # to do a full lookup in the registry type_annotations_map.
2330 annotated = de_stringify_annotation(
2331 cls,
2332 raw_annotation,
2333 originating_module,
2334 str_cleanup_fn=_cleanup_mapped_str_annotation,
2335 )
2336 except _CleanupError as ce:
2337 raise orm_exc.MappedAnnotationError(
2338 f"Could not interpret annotation {raw_annotation}. "
2339 "Check that it uses names that are correctly imported at the "
2340 "module level. See chained stack trace for more hints."
2341 ) from ce
2342 except NameError as ne:
2343 if raiseerr and "Mapped[" in raw_annotation: # type: ignore
2344 raise orm_exc.MappedAnnotationError(
2345 f"Could not interpret annotation {raw_annotation}. "
2346 "Check that it uses names that are correctly imported at the "
2347 "module level. See chained stack trace for more hints."
2348 ) from ne
2350 annotated = raw_annotation # type: ignore
2352 if is_dataclass_field:
2353 return annotated, None
2354 else:
2355 if not hasattr(annotated, "__origin__") or not is_origin_of_cls(
2356 annotated, _MappedAnnotationBase
2357 ):
2358 if expect_mapped:
2359 if not raiseerr:
2360 return None
2362 origin = getattr(annotated, "__origin__", None)
2363 if origin is typing.ClassVar:
2364 return None
2366 # check for other kind of ORM descriptor like AssociationProxy,
2367 # don't raise for that (issue #9957)
2368 elif isinstance(origin, type) and issubclass(
2369 origin, ORMDescriptor
2370 ):
2371 return None
2373 raise orm_exc.MappedAnnotationError(
2374 f'Type annotation for "{cls.__name__}.{key}" '
2375 "can't be correctly interpreted for "
2376 "Annotated Declarative Table form. ORM annotations "
2377 "should normally make use of the ``Mapped[]`` generic "
2378 "type, or other ORM-compatible generic type, as a "
2379 "container for the actual type, which indicates the "
2380 "intent that the attribute is mapped. "
2381 "Class variables that are not intended to be mapped "
2382 "by the ORM should use ClassVar[]. "
2383 "To allow Annotated Declarative to disregard legacy "
2384 "annotations which don't use Mapped[] to pass, set "
2385 '"__allow_unmapped__ = True" on the class or a '
2386 "superclass this class.",
2387 code="zlpr",
2388 )
2390 else:
2391 return annotated, None
2393 if len(annotated.__args__) != 1:
2394 raise orm_exc.MappedAnnotationError(
2395 "Expected sub-type for Mapped[] annotation"
2396 )
2398 return (
2399 # fix dict/list/set args to be ForwardRef, see #11814
2400 fixup_container_fwd_refs(annotated.__args__[0]),
2401 annotated.__origin__,
2402 )
2405def _mapper_property_as_plain_name(prop: Type[Any]) -> str:
2406 if hasattr(prop, "_mapper_property_name"):
2407 name = prop._mapper_property_name()
2408 else:
2409 name = None
2410 return util.clsname_as_plain_name(prop, name)