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