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