codekingpro/portable-devtools
115k
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
8
9from __future__ import annotations
10
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
35
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 ..util.langhelpers import MemoizedSlots
89from ..util.typing import de_stringify_annotation as _de_stringify_annotation
90from ..util.typing import eval_name_only as _eval_name_only
91from ..util.typing import fixup_container_fwd_refs
92from ..util.typing import get_origin
93from ..util.typing import is_origin_of_cls
94from ..util.typing import Literal
95from ..util.typing import Protocol
96
97if typing.TYPE_CHECKING:
98 from ._typing import _EntityType
99 from ._typing import _IdentityKeyType
100 from ._typing import _InternalEntityType
101 from ._typing import _ORMCOLEXPR
102 from .context import _MapperEntity
103 from .context import ORMCompileState
104 from .mapper import Mapper
105 from .path_registry import AbstractEntityRegistry
106 from .query import Query
107 from .relationships import RelationshipProperty
108 from ..engine import Row
109 from ..engine import RowMapping
110 from ..sql._typing import _CE
111 from ..sql._typing import _ColumnExpressionArgument
112 from ..sql._typing import _EquivalentColumnMap
113 from ..sql._typing import _FromClauseArgument
114 from ..sql._typing import _OnClauseArgument
115 from ..sql._typing import _PropagateAttrsType
116 from ..sql.annotation import _SA
117 from ..sql.base import ReadOnlyColumnCollection
118 from ..sql.elements import BindParameter
119 from ..sql.selectable import _ColumnsClauseElement
120 from ..sql.selectable import Select
121 from ..sql.selectable import Selectable
122 from ..sql.visitors import anon_map
123 from ..util.typing import _AnnotationScanType
124
125_T = TypeVar("_T", bound=Any)
126
127all_cascades = frozenset(
128 (
129 "delete",
130 "delete-orphan",
131 "all",
132 "merge",
133 "expunge",
134 "save-update",
135 "refresh-expire",
136 "none",
137 )
138)
139
140_de_stringify_partial = functools.partial(
141 functools.partial,
142 locals_=util.immutabledict(
143 {
144 "Mapped": Mapped,
145 "WriteOnlyMapped": WriteOnlyMapped,
146 "DynamicMapped": DynamicMapped,
147 }
148 ),
149)
150
151# partial is practically useless as we have to write out the whole
152# function and maintain the signature anyway
153
154
155class _DeStringifyAnnotation(Protocol):
156 def __call__(
157 self,
158 cls: Type[Any],
159 annotation: _AnnotationScanType,
160 originating_module: str,
161 *,
162 str_cleanup_fn: Optional[Callable[[str, str], str]] = None,
163 include_generic: bool = False,
164 ) -> Type[Any]: ...
165
166
167de_stringify_annotation = cast(
168 _DeStringifyAnnotation, _de_stringify_partial(_de_stringify_annotation)
169)
170
171
172class _EvalNameOnly(Protocol):
173 def __call__(self, name: str, module_name: str) -> Any: ...
174
175
176eval_name_only = cast(_EvalNameOnly, _de_stringify_partial(_eval_name_only))
177
178
179class CascadeOptions(FrozenSet[str]):
180 """Keeps track of the options sent to
181 :paramref:`.relationship.cascade`"""
182
183 _add_w_all_cascades = all_cascades.difference(
184 ["all", "none", "delete-orphan"]
185 )
186 _allowed_cascades = all_cascades
187
188 _viewonly_cascades = ["expunge", "all", "none", "refresh-expire", "merge"]
189
190 __slots__ = (
191 "save_update",
192 "delete",
193 "refresh_expire",
194 "merge",
195 "expunge",
196 "delete_orphan",
197 )
198
199 save_update: bool
200 delete: bool
201 refresh_expire: bool
202 merge: bool
203 expunge: bool
204 delete_orphan: bool
205
206 def __new__(
207 cls, value_list: Optional[Union[Iterable[str], str]]
208 ) -> CascadeOptions:
209 if isinstance(value_list, str) or value_list is None:
210 return cls.from_string(value_list) # type: ignore
211 values = set(value_list)
212 if values.difference(cls._allowed_cascades):
213 raise sa_exc.ArgumentError(
214 "Invalid cascade option(s): %s"
215 % ", ".join(
216 [
217 repr(x)
218 for x in sorted(
219 values.difference(cls._allowed_cascades)
220 )
221 ]
222 )
223 )
224
225 if "all" in values:
226 values.update(cls._add_w_all_cascades)
227 if "none" in values:
228 values.clear()
229 values.discard("all")
230
231 self = super().__new__(cls, values)
232 self.save_update = "save-update" in values
233 self.delete = "delete" in values
234 self.refresh_expire = "refresh-expire" in values
235 self.merge = "merge" in values
236 self.expunge = "expunge" in values
237 self.delete_orphan = "delete-orphan" in values
238
239 if self.delete_orphan and not self.delete:
240 util.warn("The 'delete-orphan' cascade option requires 'delete'.")
241 return self
242
243 def __repr__(self):
244 return "CascadeOptions(%r)" % (",".join([x for x in sorted(self)]))
245
246 @classmethod
247 def from_string(cls, arg):
248 values = [c for c in re.split(r"\s*,\s*", arg or "") if c]
249 return cls(values)
250
251
252def _validator_events(desc, key, validator, include_removes, include_backrefs):
253 """Runs a validation method on an attribute value to be set or
254 appended.
255 """
256
257 if not include_backrefs:
258
259 def detect_is_backref(state, initiator):
260 impl = state.manager[key].impl
261 return initiator.impl is not impl
262
263 if include_removes:
264
265 def append(state, value, initiator):
266 if initiator.op is not attributes.OP_BULK_REPLACE and (
267 include_backrefs or not detect_is_backref(state, initiator)
268 ):
269 return validator(state.obj(), key, value, False)
270 else:
271 return value
272
273 def bulk_set(state, values, initiator):
274 if include_backrefs or not detect_is_backref(state, initiator):
275 obj = state.obj()
276 values[:] = [
277 validator(obj, key, value, False) for value in values
278 ]
279
280 def set_(state, value, oldvalue, initiator):
281 if include_backrefs or not detect_is_backref(state, initiator):
282 return validator(state.obj(), key, value, False)
283 else:
284 return value
285
286 def remove(state, value, initiator):
287 if include_backrefs or not detect_is_backref(state, initiator):
288 validator(state.obj(), key, value, True)
289
290 else:
291
292 def append(state, value, initiator):
293 if initiator.op is not attributes.OP_BULK_REPLACE and (
294 include_backrefs or not detect_is_backref(state, initiator)
295 ):
296 return validator(state.obj(), key, value)
297 else:
298 return value
299
300 def bulk_set(state, values, initiator):
301 if include_backrefs or not detect_is_backref(state, initiator):
302 obj = state.obj()
303 values[:] = [validator(obj, key, value) for value in values]
304
305 def set_(state, value, oldvalue, initiator):
306 if include_backrefs or not detect_is_backref(state, initiator):
307 return validator(state.obj(), key, value)
308 else:
309 return value
310
311 event.listen(desc, "append", append, raw=True, retval=True)
312 event.listen(desc, "bulk_replace", bulk_set, raw=True)
313 event.listen(desc, "set", set_, raw=True, retval=True)
314 if include_removes:
315 event.listen(desc, "remove", remove, raw=True, retval=True)
316
317
318def polymorphic_union(
319 table_map, typecolname, aliasname="p_union", cast_nulls=True
320):
321 """Create a ``UNION`` statement used by a polymorphic mapper.
322
323 See :ref:`concrete_inheritance` for an example of how
324 this is used.
325
326 :param table_map: mapping of polymorphic identities to
327 :class:`_schema.Table` objects.
328 :param typecolname: string name of a "discriminator" column, which will be
329 derived from the query, producing the polymorphic identity for
330 each row. If ``None``, no polymorphic discriminator is generated.
331 :param aliasname: name of the :func:`~sqlalchemy.sql.expression.alias()`
332 construct generated.
333 :param cast_nulls: if True, non-existent columns, which are represented
334 as labeled NULLs, will be passed into CAST. This is a legacy behavior
335 that is problematic on some backends such as Oracle - in which case it
336 can be set to False.
337
338 """
339
340 colnames: util.OrderedSet[str] = util.OrderedSet()
341 colnamemaps = {}
342 types = {}
343 for key in table_map:
344 table = table_map[key]
345
346 table = coercions.expect(
347 roles.StrictFromClauseRole, table, allow_select=True
348 )
349 table_map[key] = table
350
351 m = {}
352 for c in table.c:
353 if c.key == typecolname:
354 raise sa_exc.InvalidRequestError(
355 "Polymorphic union can't use '%s' as the discriminator "
356 "column due to mapped column %r; please apply the "
357 "'typecolname' "
358 "argument; this is available on "
359 "ConcreteBase as '_concrete_discriminator_name'"
360 % (typecolname, c)
361 )
362 colnames.add(c.key)
363 m[c.key] = c
364 types[c.key] = c.type
365 colnamemaps[table] = m
366
367 def col(name, table):
368 try:
369 return colnamemaps[table][name]
370 except KeyError:
371 if cast_nulls:
372 return sql.cast(sql.null(), types[name]).label(name)
373 else:
374 return sql.type_coerce(sql.null(), types[name]).label(name)
375
376 result = []
377 for type_, table in table_map.items():
378 if typecolname is not None:
379 result.append(
380 sql.select(
381 *(
382 [col(name, table) for name in colnames]
383 + [
384 sql.literal_column(
385 sql_util._quote_ddl_expr(type_)
386 ).label(typecolname)
387 ]
388 )
389 ).select_from(table)
390 )
391 else:
392 result.append(
393 sql.select(
394 *[col(name, table) for name in colnames]
395 ).select_from(table)
396 )
397 return sql.union_all(*result).alias(aliasname)
398
399
400def identity_key(
401 class_: Optional[Type[_T]] = None,
402 ident: Union[Any, Tuple[Any, ...]] = None,
403 *,
404 instance: Optional[_T] = None,
405 row: Optional[Union[Row[Any], RowMapping]] = None,
406 identity_token: Optional[Any] = None,
407) -> _IdentityKeyType[_T]:
408 r"""Generate "identity key" tuples, as are used as keys in the
409 :attr:`.Session.identity_map` dictionary.
410
411 This function has several call styles:
412
413 * ``identity_key(class, ident, identity_token=token)``
414
415 This form receives a mapped class and a primary key scalar or
416 tuple as an argument.
417
418 E.g.::
419
420 >>> identity_key(MyClass, (1, 2))
421 (<class '__main__.MyClass'>, (1, 2), None)
422
423 :param class: mapped class (must be a positional argument)
424 :param ident: primary key, may be a scalar or tuple argument.
425 :param identity_token: optional identity token
426
427 .. versionadded:: 1.2 added identity_token
428
429
430 * ``identity_key(instance=instance)``
431
432 This form will produce the identity key for a given instance. The
433 instance need not be persistent, only that its primary key attributes
434 are populated (else the key will contain ``None`` for those missing
435 values).
436
437 E.g.::
438
439 >>> instance = MyClass(1, 2)
440 >>> identity_key(instance=instance)
441 (<class '__main__.MyClass'>, (1, 2), None)
442
443 In this form, the given instance is ultimately run though
444 :meth:`_orm.Mapper.identity_key_from_instance`, which will have the
445 effect of performing a database check for the corresponding row
446 if the object is expired.
447
448 :param instance: object instance (must be given as a keyword arg)
449
450 * ``identity_key(class, row=row, identity_token=token)``
451
452 This form is similar to the class/tuple form, except is passed a
453 database result row as a :class:`.Row` or :class:`.RowMapping` object.
454
455 E.g.::
456
457 >>> row = engine.execute(text("select * from table where a=1 and b=2")).first()
458 >>> identity_key(MyClass, row=row)
459 (<class '__main__.MyClass'>, (1, 2), None)
460
461 :param class: mapped class (must be a positional argument)
462 :param row: :class:`.Row` row returned by a :class:`_engine.CursorResult`
463 (must be given as a keyword arg)
464 :param identity_token: optional identity token
465
466 .. versionadded:: 1.2 added identity_token
467
468 """ # noqa: E501
469 if class_ is not None:
470 mapper = class_mapper(class_)
471 if row is None:
472 if ident is None:
473 raise sa_exc.ArgumentError("ident or row is required")
474 return mapper.identity_key_from_primary_key(
475 tuple(util.to_list(ident)), identity_token=identity_token
476 )
477 else:
478 return mapper.identity_key_from_row(
479 row, identity_token=identity_token
480 )
481 elif instance is not None:
482 mapper = object_mapper(instance)
483 return mapper.identity_key_from_instance(instance)
484 else:
485 raise sa_exc.ArgumentError("class or instance is required")
486
487
488class _TraceAdaptRole(enum.Enum):
489 """Enumeration of all the use cases for ORMAdapter.
490
491 ORMAdapter remains one of the most complicated aspects of the ORM, as it is
492 used for in-place adaption of column expressions to be applied to a SELECT,
493 replacing :class:`.Table` and other objects that are mapped to classes with
494 aliases of those tables in the case of joined eager loading, or in the case
495 of polymorphic loading as used with concrete mappings or other custom "with
496 polymorphic" parameters, with whole user-defined subqueries. The
497 enumerations provide an overview of all the use cases used by ORMAdapter, a
498 layer of formality as to the introduction of new ORMAdapter use cases (of
499 which none are anticipated), as well as a means to trace the origins of a
500 particular ORMAdapter within runtime debugging.
501
502 SQLAlchemy 2.0 has greatly scaled back ORM features which relied heavily on
503 open-ended statement adaption, including the ``Query.with_polymorphic()``
504 method and the ``Query.select_from_entity()`` methods, favoring
505 user-explicit aliasing schemes using the ``aliased()`` and
506 ``with_polymorphic()`` standalone constructs; these still use adaption,
507 however the adaption is applied in a narrower scope.
508
509 """
510
511 # aliased() use that is used to adapt individual attributes at query
512 # construction time
513 ALIASED_INSP = enum.auto()
514
515 # joinedload cases; typically adapt an ON clause of a relationship
516 # join
517 JOINEDLOAD_USER_DEFINED_ALIAS = enum.auto()
518 JOINEDLOAD_PATH_WITH_POLYMORPHIC = enum.auto()
519 JOINEDLOAD_MEMOIZED_ADAPTER = enum.auto()
520
521 # polymorphic cases - these are complex ones that replace FROM
522 # clauses, replacing tables with subqueries
523 MAPPER_POLYMORPHIC_ADAPTER = enum.auto()
524 WITH_POLYMORPHIC_ADAPTER = enum.auto()
525 WITH_POLYMORPHIC_ADAPTER_RIGHT_JOIN = enum.auto()
526 DEPRECATED_JOIN_ADAPT_RIGHT_SIDE = enum.auto()
527
528 # the from_statement() case, used only to adapt individual attributes
529 # from a given statement to local ORM attributes at result fetching
530 # time. assigned to ORMCompileState._from_obj_alias
531 ADAPT_FROM_STATEMENT = enum.auto()
532
533 # the joinedload for queries that have LIMIT/OFFSET/DISTINCT case;
534 # the query is placed inside of a subquery with the LIMIT/OFFSET/etc.,
535 # joinedloads are then placed on the outside.
536 # assigned to ORMCompileState.compound_eager_adapter
537 COMPOUND_EAGER_STATEMENT = enum.auto()
538
539 # the legacy Query._set_select_from() case.
540 # this is needed for Query's set operations (i.e. UNION, etc. )
541 # as well as "legacy from_self()", which while removed from 2.0 as
542 # public API, is used for the Query.count() method. this one
543 # still does full statement traversal
544 # assigned to ORMCompileState._from_obj_alias
545 LEGACY_SELECT_FROM_ALIAS = enum.auto()
546
547
548class ORMStatementAdapter(sql_util.ColumnAdapter):
549 """ColumnAdapter which includes a role attribute."""
550
551 __slots__ = ("role",)
552
553 def __init__(
554 self,
555 role: _TraceAdaptRole,
556 selectable: Selectable,
557 *,
558 equivalents: Optional[_EquivalentColumnMap] = None,
559 adapt_required: bool = False,
560 allow_label_resolve: bool = True,
561 anonymize_labels: bool = False,
562 adapt_on_names: bool = False,
563 adapt_from_selectables: Optional[AbstractSet[FromClause]] = None,
564 ):
565 self.role = role
566 super().__init__(
567 selectable,
568 equivalents=equivalents,
569 adapt_required=adapt_required,
570 allow_label_resolve=allow_label_resolve,
571 anonymize_labels=anonymize_labels,
572 adapt_on_names=adapt_on_names,
573 adapt_from_selectables=adapt_from_selectables,
574 )
575
576
577class ORMAdapter(sql_util.ColumnAdapter):
578 """ColumnAdapter subclass which excludes adaptation of entities from
579 non-matching mappers.
580
581 """
582
583 __slots__ = ("role", "mapper", "is_aliased_class", "aliased_insp")
584
585 is_aliased_class: bool
586 aliased_insp: Optional[AliasedInsp[Any]]
587
588 def __init__(
589 self,
590 role: _TraceAdaptRole,
591 entity: _InternalEntityType[Any],
592 *,
593 equivalents: Optional[_EquivalentColumnMap] = None,
594 adapt_required: bool = False,
595 allow_label_resolve: bool = True,
596 anonymize_labels: bool = False,
597 selectable: Optional[Selectable] = None,
598 limit_on_entity: bool = True,
599 adapt_on_names: bool = False,
600 adapt_from_selectables: Optional[AbstractSet[FromClause]] = None,
601 ):
602 self.role = role
603 self.mapper = entity.mapper
604 if selectable is None:
605 selectable = entity.selectable
606 if insp_is_aliased_class(entity):
607 self.is_aliased_class = True
608 self.aliased_insp = entity
609 else:
610 self.is_aliased_class = False
611 self.aliased_insp = None
612
613 super().__init__(
614 selectable,
615 equivalents,
616 adapt_required=adapt_required,
617 allow_label_resolve=allow_label_resolve,
618 anonymize_labels=anonymize_labels,
619 include_fn=self._include_fn if limit_on_entity else None,
620 adapt_on_names=adapt_on_names,
621 adapt_from_selectables=adapt_from_selectables,
622 )
623
624 def _include_fn(self, elem):
625 entity = elem._annotations.get("parentmapper", None)
626
627 return not entity or entity.isa(self.mapper) or self.mapper.isa(entity)
628
629
630class AliasedClass(
631 inspection.Inspectable["AliasedInsp[_O]"], ORMColumnsClauseRole[_O]
632):
633 r"""Represents an "aliased" form of a mapped class for usage with Query.
634
635 The ORM equivalent of a :func:`~sqlalchemy.sql.expression.alias`
636 construct, this object mimics the mapped class using a
637 ``__getattr__`` scheme and maintains a reference to a
638 real :class:`~sqlalchemy.sql.expression.Alias` object.
639
640 A primary purpose of :class:`.AliasedClass` is to serve as an alternate
641 within a SQL statement generated by the ORM, such that an existing
642 mapped entity can be used in multiple contexts. A simple example::
643
644 # find all pairs of users with the same name
645 user_alias = aliased(User)
646 session.query(User, user_alias).join(
647 (user_alias, User.id > user_alias.id)
648 ).filter(User.name == user_alias.name)
649
650 :class:`.AliasedClass` is also capable of mapping an existing mapped
651 class to an entirely new selectable, provided this selectable is column-
652 compatible with the existing mapped selectable, and it can also be
653 configured in a mapping as the target of a :func:`_orm.relationship`.
654 See the links below for examples.
655
656 The :class:`.AliasedClass` object is constructed typically using the
657 :func:`_orm.aliased` function. It also is produced with additional
658 configuration when using the :func:`_orm.with_polymorphic` function.
659
660 The resulting object is an instance of :class:`.AliasedClass`.
661 This object implements an attribute scheme which produces the
662 same attribute and method interface as the original mapped
663 class, allowing :class:`.AliasedClass` to be compatible
664 with any attribute technique which works on the original class,
665 including hybrid attributes (see :ref:`hybrids_toplevel`).
666
667 The :class:`.AliasedClass` can be inspected for its underlying
668 :class:`_orm.Mapper`, aliased selectable, and other information
669 using :func:`_sa.inspect`::
670
671 from sqlalchemy import inspect
672
673 my_alias = aliased(MyClass)
674 insp = inspect(my_alias)
675
676 The resulting inspection object is an instance of :class:`.AliasedInsp`.
677
678
679 .. seealso::
680
681 :func:`.aliased`
682
683 :func:`.with_polymorphic`
684
685 :ref:`relationship_aliased_class`
686
687 :ref:`relationship_to_window_function`
688
689
690 """
691
692 __name__: str
693
694 def __init__(
695 self,
696 mapped_class_or_ac: _EntityType[_O],
697 alias: Optional[FromClause] = None,
698 name: Optional[str] = None,
699 flat: bool = False,
700 adapt_on_names: bool = False,
701 with_polymorphic_mappers: Optional[Sequence[Mapper[Any]]] = None,
702 with_polymorphic_discriminator: Optional[ColumnElement[Any]] = None,
703 base_alias: Optional[AliasedInsp[Any]] = None,
704 use_mapper_path: bool = False,
705 represents_outer_join: bool = False,
706 ):
707 insp = cast(
708 "_InternalEntityType[_O]", inspection.inspect(mapped_class_or_ac)
709 )
710 mapper = insp.mapper
711
712 nest_adapters = False
713
714 if alias is None:
715 if insp.is_aliased_class and insp.selectable._is_subquery:
716 alias = insp.selectable.alias()
717 else:
718 alias = (
719 mapper._with_polymorphic_selectable._anonymous_fromclause(
720 name=name,
721 flat=flat,
722 )
723 )
724 elif insp.is_aliased_class:
725 nest_adapters = True
726
727 assert alias is not None
728 self._aliased_insp = AliasedInsp(
729 self,
730 insp,
731 alias,
732 name,
733 (
734 with_polymorphic_mappers
735 if with_polymorphic_mappers
736 else mapper.with_polymorphic_mappers
737 ),
738 (
739 with_polymorphic_discriminator
740 if with_polymorphic_discriminator is not None
741 else mapper.polymorphic_on
742 ),
743 base_alias,
744 use_mapper_path,
745 adapt_on_names,
746 represents_outer_join,
747 nest_adapters,
748 )
749
750 self.__name__ = f"aliased({mapper.class_.__name__})"
751
752 @classmethod
753 def _reconstitute_from_aliased_insp(
754 cls, aliased_insp: AliasedInsp[_O]
755 ) -> AliasedClass[_O]:
756 obj = cls.__new__(cls)
757 obj.__name__ = f"aliased({aliased_insp.mapper.class_.__name__})"
758 obj._aliased_insp = aliased_insp
759
760 if aliased_insp._is_with_polymorphic:
761 for sub_aliased_insp in aliased_insp._with_polymorphic_entities:
762 if sub_aliased_insp is not aliased_insp:
763 ent = AliasedClass._reconstitute_from_aliased_insp(
764 sub_aliased_insp
765 )
766 setattr(obj, sub_aliased_insp.class_.__name__, ent)
767
768 return obj
769
770 def __getattr__(self, key: str) -> Any:
771 try:
772 _aliased_insp = self.__dict__["_aliased_insp"]
773 except KeyError:
774 raise AttributeError()
775 else:
776 target = _aliased_insp._target
777 # maintain all getattr mechanics
778 attr = getattr(target, key)
779
780 # attribute is a method, that will be invoked against a
781 # "self"; so just return a new method with the same function and
782 # new self
783 if hasattr(attr, "__call__") and hasattr(attr, "__self__"):
784 return types.MethodType(attr.__func__, self)
785
786 # attribute is a descriptor, that will be invoked against a
787 # "self"; so invoke the descriptor against this self
788 if hasattr(attr, "__get__"):
789 attr = attr.__get__(None, self)
790
791 # attributes within the QueryableAttribute system will want this
792 # to be invoked so the object can be adapted
793 if hasattr(attr, "adapt_to_entity"):
794 attr = attr.adapt_to_entity(_aliased_insp)
795 setattr(self, key, attr)
796
797 return attr
798
799 def _get_from_serialized(
800 self, key: str, mapped_class: _O, aliased_insp: AliasedInsp[_O]
801 ) -> Any:
802 # this method is only used in terms of the
803 # sqlalchemy.ext.serializer extension
804 attr = getattr(mapped_class, key)
805 if hasattr(attr, "__call__") and hasattr(attr, "__self__"):
806 return types.MethodType(attr.__func__, self)
807
808 # attribute is a descriptor, that will be invoked against a
809 # "self"; so invoke the descriptor against this self
810 if hasattr(attr, "__get__"):
811 attr = attr.__get__(None, self)
812
813 # attributes within the QueryableAttribute system will want this
814 # to be invoked so the object can be adapted
815 if hasattr(attr, "adapt_to_entity"):
816 aliased_insp._weak_entity = weakref.ref(self)
817 attr = attr.adapt_to_entity(aliased_insp)
818 setattr(self, key, attr)
819
820 return attr
821
822 def __repr__(self) -> str:
823 return "<AliasedClass at 0x%x; %s>" % (
824 id(self),
825 self._aliased_insp._target.__name__,
826 )
827
828 def __str__(self) -> str:
829 return str(self._aliased_insp)
830
831
832@inspection._self_inspects
833class AliasedInsp(
834 ORMEntityColumnsClauseRole[_O],
835 ORMFromClauseRole,
836 HasCacheKey,
837 InspectionAttr,
838 MemoizedSlots,
839 inspection.Inspectable["AliasedInsp[_O]"],
840 Generic[_O],
841):
842 """Provide an inspection interface for an
843 :class:`.AliasedClass` object.
844
845 The :class:`.AliasedInsp` object is returned
846 given an :class:`.AliasedClass` using the
847 :func:`_sa.inspect` function::
848
849 from sqlalchemy import inspect
850 from sqlalchemy.orm import aliased
851
852 my_alias = aliased(MyMappedClass)
853 insp = inspect(my_alias)
854
855 Attributes on :class:`.AliasedInsp`
856 include:
857
858 * ``entity`` - the :class:`.AliasedClass` represented.
859 * ``mapper`` - the :class:`_orm.Mapper` mapping the underlying class.
860 * ``selectable`` - the :class:`_expression.Alias`
861 construct which ultimately
862 represents an aliased :class:`_schema.Table` or
863 :class:`_expression.Select`
864 construct.
865 * ``name`` - the name of the alias. Also is used as the attribute
866 name when returned in a result tuple from :class:`_query.Query`.
867 * ``with_polymorphic_mappers`` - collection of :class:`_orm.Mapper`
868 objects
869 indicating all those mappers expressed in the select construct
870 for the :class:`.AliasedClass`.
871 * ``polymorphic_on`` - an alternate column or SQL expression which
872 will be used as the "discriminator" for a polymorphic load.
873
874 .. seealso::
875
876 :ref:`inspection_toplevel`
877
878 """
879
880 __slots__ = (
881 "__weakref__",
882 "_weak_entity",
883 "mapper",
884 "selectable",
885 "name",
886 "_adapt_on_names",
887 "with_polymorphic_mappers",
888 "polymorphic_on",
889 "_use_mapper_path",
890 "_base_alias",
891 "represents_outer_join",
892 "persist_selectable",
893 "local_table",
894 "_is_with_polymorphic",
895 "_with_polymorphic_entities",
896 "_adapter",
897 "_target",
898 "__clause_element__",
899 "_memoized_values",
900 "_all_column_expressions",
901 "_nest_adapters",
902 )
903
904 _cache_key_traversal = [
905 ("name", visitors.ExtendedInternalTraversal.dp_string),
906 ("_adapt_on_names", visitors.ExtendedInternalTraversal.dp_boolean),
907 ("_use_mapper_path", visitors.ExtendedInternalTraversal.dp_boolean),
908 ("_target", visitors.ExtendedInternalTraversal.dp_inspectable),
909 ("selectable", visitors.ExtendedInternalTraversal.dp_clauseelement),
910 (
911 "with_polymorphic_mappers",
912 visitors.InternalTraversal.dp_has_cache_key_list,
913 ),
914 ("polymorphic_on", visitors.InternalTraversal.dp_clauseelement),
915 ]
916
917 mapper: Mapper[_O]
918 selectable: FromClause
919 _adapter: ORMAdapter
920 with_polymorphic_mappers: Sequence[Mapper[Any]]
921 _with_polymorphic_entities: Sequence[AliasedInsp[Any]]
922
923 _weak_entity: weakref.ref[AliasedClass[_O]]
924 """the AliasedClass that refers to this AliasedInsp"""
925
926 _target: Union[Type[_O], AliasedClass[_O]]
927 """the thing referenced by the AliasedClass/AliasedInsp.
928
929 In the vast majority of cases, this is the mapped class. However
930 it may also be another AliasedClass (alias of alias).
931
932 """
933
934 def __init__(
935 self,
936 entity: AliasedClass[_O],
937 inspected: _InternalEntityType[_O],
938 selectable: FromClause,
939 name: Optional[str],
940 with_polymorphic_mappers: Optional[Sequence[Mapper[Any]]],
941 polymorphic_on: Optional[ColumnElement[Any]],
942 _base_alias: Optional[AliasedInsp[Any]],
943 _use_mapper_path: bool,
944 adapt_on_names: bool,
945 represents_outer_join: bool,
946 nest_adapters: bool,
947 ):
948 mapped_class_or_ac = inspected.entity
949 mapper = inspected.mapper
950
951 self._weak_entity = weakref.ref(entity)
952 self.mapper = mapper
953 self.selectable = self.persist_selectable = self.local_table = (
954 selectable
955 )
956 self.name = name
957 self.polymorphic_on = polymorphic_on
958 self._base_alias = weakref.ref(_base_alias or self)
959 self._use_mapper_path = _use_mapper_path
960 self.represents_outer_join = represents_outer_join
961 self._nest_adapters = nest_adapters
962
963 if with_polymorphic_mappers:
964 self._is_with_polymorphic = True
965 self.with_polymorphic_mappers = with_polymorphic_mappers
966 self._with_polymorphic_entities = []
967 for poly in self.with_polymorphic_mappers:
968 if poly is not mapper:
969 ent = AliasedClass(
970 poly.class_,
971 selectable,
972 base_alias=self,
973 adapt_on_names=adapt_on_names,
974 use_mapper_path=_use_mapper_path,
975 )
976
977 setattr(self.entity, poly.class_.__name__, ent)
978 self._with_polymorphic_entities.append(ent._aliased_insp)
979
980 else:
981 self._is_with_polymorphic = False
982 self.with_polymorphic_mappers = [mapper]
983
984 self._adapter = ORMAdapter(
985 _TraceAdaptRole.ALIASED_INSP,
986 mapper,
987 selectable=selectable,
988 equivalents=mapper._equivalent_columns,
989 adapt_on_names=adapt_on_names,
990 anonymize_labels=True,
991 # make sure the adapter doesn't try to grab other tables that
992 # are not even the thing we are mapping, such as embedded
993 # selectables in subqueries or CTEs. See issue #6060
994 adapt_from_selectables={
995 m.selectable
996 for m in self.with_polymorphic_mappers
997 if not adapt_on_names
998 },
999 limit_on_entity=False,
1000 )
1001
1002 if nest_adapters:
1003 # supports "aliased class of aliased class" use case
1004 assert isinstance(inspected, AliasedInsp)
1005 self._adapter = inspected._adapter.wrap(self._adapter)
1006
1007 self._adapt_on_names = adapt_on_names
1008 self._target = mapped_class_or_ac
1009
1010 @classmethod
1011 def _alias_factory(
1012 cls,
1013 element: Union[_EntityType[_O], FromClause],
1014 alias: Optional[FromClause] = None,
1015 name: Optional[str] = None,
1016 flat: bool = False,
1017 adapt_on_names: bool = False,
1018 ) -> Union[AliasedClass[_O], FromClause]:
1019 if isinstance(element, FromClause):
1020 if adapt_on_names:
1021 raise sa_exc.ArgumentError(
1022 "adapt_on_names only applies to ORM elements"
1023 )
1024 if name:
1025 return element.alias(name=name, flat=flat)
1026 else:
1027 return coercions.expect(
1028 roles.AnonymizedFromClauseRole, element, flat=flat
1029 )
1030 else:
1031 return AliasedClass(
1032 element,
1033 alias=alias,
1034 flat=flat,
1035 name=name,
1036 adapt_on_names=adapt_on_names,
1037 )
1038
1039 @classmethod
1040 def _with_polymorphic_factory(
1041 cls,
1042 base: Union[Type[_O], Mapper[_O]],
1043 classes: Union[Literal["*"], Iterable[_EntityType[Any]]],
1044 selectable: Union[Literal[False, None], FromClause] = False,
1045 flat: bool = False,
1046 polymorphic_on: Optional[ColumnElement[Any]] = None,
1047 aliased: bool = False,
1048 innerjoin: bool = False,
1049 adapt_on_names: bool = False,
1050 name: Optional[str] = None,
1051 _use_mapper_path: bool = False,
1052 ) -> AliasedClass[_O]:
1053 primary_mapper = _class_to_mapper(base)
1054
1055 if selectable not in (None, False) and flat:
1056 raise sa_exc.ArgumentError(
1057 "the 'flat' and 'selectable' arguments cannot be passed "
1058 "simultaneously to with_polymorphic()"
1059 )
1060
1061 mappers, selectable = primary_mapper._with_polymorphic_args(
1062 classes, selectable, innerjoin=innerjoin
1063 )
1064 if aliased or flat:
1065 assert selectable is not None
1066 selectable = selectable._anonymous_fromclause(flat=flat)
1067
1068 return AliasedClass(
1069 base,
1070 selectable,
1071 name=name,
1072 with_polymorphic_mappers=mappers,
1073 adapt_on_names=adapt_on_names,
1074 with_polymorphic_discriminator=polymorphic_on,
1075 use_mapper_path=_use_mapper_path,
1076 represents_outer_join=not innerjoin,
1077 )
1078
1079 @property
1080 def entity(self) -> AliasedClass[_O]:
1081 # to eliminate reference cycles, the AliasedClass is held weakly.
1082 # this produces some situations where the AliasedClass gets lost,
1083 # particularly when one is created internally and only the AliasedInsp
1084 # is passed around.
1085 # to work around this case, we just generate a new one when we need
1086 # it, as it is a simple class with very little initial state on it.
1087 ent = self._weak_entity()
1088 if ent is None:
1089 ent = AliasedClass._reconstitute_from_aliased_insp(self)
1090 self._weak_entity = weakref.ref(ent)
1091 return ent
1092
1093 is_aliased_class = True
1094 "always returns True"
1095
1096 def _memoized_method___clause_element__(self) -> FromClause:
1097 return self.selectable._annotate(
1098 {
1099 "parentmapper": self.mapper,
1100 "parententity": self,
1101 "entity_namespace": self,
1102 }
1103 )._set_propagate_attrs(
1104 {"compile_state_plugin": "orm", "plugin_subject": self}
1105 )
1106
1107 @property
1108 def entity_namespace(self) -> AliasedClass[_O]:
1109 return self.entity
1110
1111 @property
1112 def class_(self) -> Type[_O]:
1113 """Return the mapped class ultimately represented by this
1114 :class:`.AliasedInsp`."""
1115 return self.mapper.class_
1116
1117 @property
1118 def _path_registry(self) -> AbstractEntityRegistry:
1119 if self._use_mapper_path:
1120 return self.mapper._path_registry
1121 else:
1122 return PathRegistry.per_mapper(self)
1123
1124 def __getstate__(self) -> Dict[str, Any]:
1125 return {
1126 "entity": self.entity,
1127 "mapper": self.mapper,
1128 "alias": self.selectable,
1129 "name": self.name,
1130 "adapt_on_names": self._adapt_on_names,
1131 "with_polymorphic_mappers": self.with_polymorphic_mappers,
1132 "with_polymorphic_discriminator": self.polymorphic_on,
1133 "base_alias": self._base_alias(),
1134 "use_mapper_path": self._use_mapper_path,
1135 "represents_outer_join": self.represents_outer_join,
1136 "nest_adapters": self._nest_adapters,
1137 }
1138
1139 def __setstate__(self, state: Dict[str, Any]) -> None:
1140 self.__init__( # type: ignore
1141 state["entity"],
1142 state["mapper"],
1143 state["alias"],
1144 state["name"],
1145 state["with_polymorphic_mappers"],
1146 state["with_polymorphic_discriminator"],
1147 state["base_alias"],
1148 state["use_mapper_path"],
1149 state["adapt_on_names"],
1150 state["represents_outer_join"],
1151 state["nest_adapters"],
1152 )
1153
1154 def _merge_with(self, other: AliasedInsp[_O]) -> AliasedInsp[_O]:
1155 # assert self._is_with_polymorphic
1156 # assert other._is_with_polymorphic
1157
1158 primary_mapper = other.mapper
1159
1160 assert self.mapper is primary_mapper
1161
1162 our_classes = util.to_set(
1163 mp.class_ for mp in self.with_polymorphic_mappers
1164 )
1165 new_classes = {mp.class_ for mp in other.with_polymorphic_mappers}
1166 if our_classes == new_classes:
1167 return other
1168 else:
1169 classes = our_classes.union(new_classes)
1170
1171 mappers, selectable = primary_mapper._with_polymorphic_args(
1172 classes, None, innerjoin=not other.represents_outer_join
1173 )
1174 selectable = selectable._anonymous_fromclause(flat=True)
1175 return AliasedClass(
1176 primary_mapper,
1177 selectable,
1178 with_polymorphic_mappers=mappers,
1179 with_polymorphic_discriminator=other.polymorphic_on,
1180 use_mapper_path=other._use_mapper_path,
1181 represents_outer_join=other.represents_outer_join,
1182 )._aliased_insp
1183
1184 def _adapt_element(
1185 self, expr: _ORMCOLEXPR, key: Optional[str] = None
1186 ) -> _ORMCOLEXPR:
1187 assert isinstance(expr, ColumnElement)
1188 d: Dict[str, Any] = {
1189 "parententity": self,
1190 "parentmapper": self.mapper,
1191 }
1192 if key:
1193 d["proxy_key"] = key
1194
1195 # IMO mypy should see this one also as returning the same type
1196 # we put into it, but it's not
1197 return (
1198 self._adapter.traverse(expr)
1199 ._annotate(d)
1200 ._set_propagate_attrs(