-
-
Notifications
You must be signed in to change notification settings - Fork 1.4k
/
context.py
3264 lines (2751 loc) · 110 KB
/
context.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
# orm/context.py
# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
from __future__ import annotations
import itertools
from typing import Any
from typing import cast
from typing import Dict
from typing import Iterable
from typing import List
from typing import Optional
from typing import Set
from typing import Tuple
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
from typing import Union
from . import attributes
from . import interfaces
from . import loading
from .base import _is_aliased_class
from .interfaces import ORMColumnDescription
from .interfaces import ORMColumnsClauseRole
from .path_registry import PathRegistry
from .util import _entity_corresponds_to
from .util import _ORMJoin
from .util import _TraceAdaptRole
from .util import AliasedClass
from .util import Bundle
from .util import ORMAdapter
from .util import ORMStatementAdapter
from .. import exc as sa_exc
from .. import future
from .. import inspect
from .. import sql
from .. import util
from ..sql import coercions
from ..sql import expression
from ..sql import roles
from ..sql import util as sql_util
from ..sql import visitors
from ..sql._typing import is_dml
from ..sql._typing import is_insert_update
from ..sql._typing import is_select_base
from ..sql.base import _select_iterables
from ..sql.base import CacheableOptions
from ..sql.base import CompileState
from ..sql.base import Executable
from ..sql.base import Generative
from ..sql.base import Options
from ..sql.dml import UpdateBase
from ..sql.elements import GroupedElement
from ..sql.elements import TextClause
from ..sql.selectable import CompoundSelectState
from ..sql.selectable import LABEL_STYLE_DISAMBIGUATE_ONLY
from ..sql.selectable import LABEL_STYLE_NONE
from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
from ..sql.selectable import Select
from ..sql.selectable import SelectLabelStyle
from ..sql.selectable import SelectState
from ..sql.selectable import TypedReturnsRows
from ..sql.visitors import InternalTraversal
from ..util.typing import TupleAny
from ..util.typing import TypeVarTuple
from ..util.typing import Unpack
if TYPE_CHECKING:
from ._typing import _InternalEntityType
from ._typing import OrmExecuteOptionsParameter
from .loading import PostLoad
from .mapper import Mapper
from .query import Query
from .session import _BindArguments
from .session import Session
from ..engine import Result
from ..engine.interfaces import _CoreSingleExecuteParams
from ..sql._typing import _ColumnsClauseArgument
from ..sql.compiler import SQLCompiler
from ..sql.dml import _DMLTableElement
from ..sql.elements import ColumnElement
from ..sql.selectable import _JoinTargetElement
from ..sql.selectable import _LabelConventionCallable
from ..sql.selectable import _SetupJoinsElement
from ..sql.selectable import ExecutableReturnsRows
from ..sql.selectable import SelectBase
from ..sql.type_api import TypeEngine
_T = TypeVar("_T", bound=Any)
_Ts = TypeVarTuple("_Ts")
_path_registry = PathRegistry.root
_EMPTY_DICT = util.immutabledict()
LABEL_STYLE_LEGACY_ORM = SelectLabelStyle.LABEL_STYLE_LEGACY_ORM
class QueryContext:
__slots__ = (
"top_level_context",
"compile_state",
"query",
"params",
"load_options",
"bind_arguments",
"execution_options",
"session",
"autoflush",
"populate_existing",
"invoke_all_eagers",
"version_check",
"refresh_state",
"create_eager_joins",
"propagated_loader_options",
"attributes",
"runid",
"partials",
"post_load_paths",
"identity_token",
"yield_per",
"loaders_require_buffering",
"loaders_require_uniquing",
)
runid: int
post_load_paths: Dict[PathRegistry, PostLoad]
compile_state: ORMCompileState
class default_load_options(Options):
_only_return_tuples = False
_populate_existing = False
_version_check = False
_invoke_all_eagers = True
_autoflush = True
_identity_token = None
_yield_per = None
_refresh_state = None
_lazy_loaded_from = None
_legacy_uniquing = False
_sa_top_level_orm_context = None
_is_user_refresh = False
def __init__(
self,
compile_state: CompileState,
statement: Union[
Select[Unpack[TupleAny]],
FromStatement[Unpack[TupleAny]],
],
params: _CoreSingleExecuteParams,
session: Session,
load_options: Union[
Type[QueryContext.default_load_options],
QueryContext.default_load_options,
],
execution_options: Optional[OrmExecuteOptionsParameter] = None,
bind_arguments: Optional[_BindArguments] = None,
):
self.load_options = load_options
self.execution_options = execution_options or _EMPTY_DICT
self.bind_arguments = bind_arguments or _EMPTY_DICT
self.compile_state = compile_state
self.query = statement
self.session = session
self.loaders_require_buffering = False
self.loaders_require_uniquing = False
self.params = params
self.top_level_context = load_options._sa_top_level_orm_context
cached_options = compile_state.select_statement._with_options
uncached_options = statement._with_options
# see issue #7447 , #8399 for some background
# propagated loader options will be present on loaded InstanceState
# objects under state.load_options and are typically used by
# LazyLoader to apply options to the SELECT statement it emits.
# For compile state options (i.e. loader strategy options), these
# need to line up with the ".load_path" attribute which in
# loader.py is pulled from context.compile_state.current_path.
# so, this means these options have to be the ones from the
# *cached* statement that's travelling with compile_state, not the
# *current* statement which won't match up for an ad-hoc
# AliasedClass
self.propagated_loader_options = tuple(
opt._adapt_cached_option_to_uncached_option(self, uncached_opt)
for opt, uncached_opt in zip(cached_options, uncached_options)
if opt.propagate_to_loaders
)
self.attributes = dict(compile_state.attributes)
self.autoflush = load_options._autoflush
self.populate_existing = load_options._populate_existing
self.invoke_all_eagers = load_options._invoke_all_eagers
self.version_check = load_options._version_check
self.refresh_state = load_options._refresh_state
self.yield_per = load_options._yield_per
self.identity_token = load_options._identity_token
def _get_top_level_context(self) -> QueryContext:
return self.top_level_context or self
_orm_load_exec_options = util.immutabledict(
{"_result_disable_adapt_to_context": True}
)
class AbstractORMCompileState(CompileState):
is_dml_returning = False
def _init_global_attributes(
self, statement, compiler, *, toplevel, process_criteria_for_toplevel
):
self.attributes = {}
if compiler is None:
# this is the legacy / testing only ORM _compile_state() use case.
# there is no need to apply criteria options for this.
self.global_attributes = ga = {}
assert toplevel
return
else:
self.global_attributes = ga = compiler._global_attributes
if toplevel:
ga["toplevel_orm"] = True
if process_criteria_for_toplevel:
for opt in statement._with_options:
if opt._is_criteria_option:
opt.process_compile_state(self)
return
elif ga.get("toplevel_orm", False):
return
stack_0 = compiler.stack[0]
try:
toplevel_stmt = stack_0["selectable"]
except KeyError:
pass
else:
for opt in toplevel_stmt._with_options:
if opt._is_compile_state and opt._is_criteria_option:
opt.process_compile_state(self)
ga["toplevel_orm"] = True
@classmethod
def create_for_statement(
cls,
statement: Union[Select, FromStatement],
compiler: Optional[SQLCompiler],
**kw: Any,
) -> AbstractORMCompileState:
"""Create a context for a statement given a :class:`.Compiler`.
This method is always invoked in the context of SQLCompiler.process().
For a Select object, this would be invoked from
SQLCompiler.visit_select(). For the special FromStatement object used
by Query to indicate "Query.from_statement()", this is called by
FromStatement._compiler_dispatch() that would be called by
SQLCompiler.process().
"""
return super().create_for_statement(statement, compiler, **kw)
@classmethod
def orm_pre_session_exec(
cls,
session,
statement,
params,
execution_options,
bind_arguments,
is_pre_event,
):
raise NotImplementedError()
@classmethod
def orm_execute_statement(
cls,
session,
statement,
params,
execution_options,
bind_arguments,
conn,
) -> Result:
result = conn.execute(
statement, params or {}, execution_options=execution_options
)
return cls.orm_setup_cursor_result(
session,
statement,
params,
execution_options,
bind_arguments,
result,
)
@classmethod
def orm_setup_cursor_result(
cls,
session,
statement,
params,
execution_options,
bind_arguments,
result,
):
raise NotImplementedError()
class AutoflushOnlyORMCompileState(AbstractORMCompileState):
"""ORM compile state that is a passthrough, except for autoflush."""
@classmethod
def orm_pre_session_exec(
cls,
session,
statement,
params,
execution_options,
bind_arguments,
is_pre_event,
):
# consume result-level load_options. These may have been set up
# in an ORMExecuteState hook
(
load_options,
execution_options,
) = QueryContext.default_load_options.from_execution_options(
"_sa_orm_load_options",
{
"autoflush",
},
execution_options,
statement._execution_options,
)
if not is_pre_event and load_options._autoflush:
session._autoflush()
return statement, execution_options
@classmethod
def orm_setup_cursor_result(
cls,
session,
statement,
params,
execution_options,
bind_arguments,
result,
):
return result
class ORMCompileState(AbstractORMCompileState):
class default_compile_options(CacheableOptions):
_cache_key_traversal = [
("_use_legacy_query_style", InternalTraversal.dp_boolean),
("_for_statement", InternalTraversal.dp_boolean),
("_bake_ok", InternalTraversal.dp_boolean),
("_current_path", InternalTraversal.dp_has_cache_key),
("_enable_single_crit", InternalTraversal.dp_boolean),
("_enable_eagerloads", InternalTraversal.dp_boolean),
("_only_load_props", InternalTraversal.dp_plain_obj),
("_set_base_alias", InternalTraversal.dp_boolean),
("_for_refresh_state", InternalTraversal.dp_boolean),
("_render_for_subquery", InternalTraversal.dp_boolean),
("_is_star", InternalTraversal.dp_boolean),
]
# set to True by default from Query._statement_20(), to indicate
# the rendered query should look like a legacy ORM query. right
# now this basically indicates we should use tablename_columnname
# style labels. Generally indicates the statement originated
# from a Query object.
_use_legacy_query_style = False
# set *only* when we are coming from the Query.statement
# accessor, or a Query-level equivalent such as
# query.subquery(). this supersedes "toplevel".
_for_statement = False
_bake_ok = True
_current_path = _path_registry
_enable_single_crit = True
_enable_eagerloads = True
_only_load_props = None
_set_base_alias = False
_for_refresh_state = False
_render_for_subquery = False
_is_star = False
attributes: Dict[Any, Any]
global_attributes: Dict[Any, Any]
statement: Union[Select[Unpack[TupleAny]], FromStatement[Unpack[TupleAny]]]
select_statement: Union[
Select[Unpack[TupleAny]], FromStatement[Unpack[TupleAny]]
]
_entities: List[_QueryEntity]
_polymorphic_adapters: Dict[_InternalEntityType, ORMAdapter]
compile_options: Union[
Type[default_compile_options], default_compile_options
]
_primary_entity: Optional[_QueryEntity]
use_legacy_query_style: bool
_label_convention: _LabelConventionCallable
primary_columns: List[ColumnElement[Any]]
secondary_columns: List[ColumnElement[Any]]
dedupe_columns: Set[ColumnElement[Any]]
create_eager_joins: List[
# TODO: this structure is set up by JoinedLoader
TupleAny
]
current_path: PathRegistry = _path_registry
_has_mapper_entities = False
def __init__(self, *arg, **kw):
raise NotImplementedError()
if TYPE_CHECKING:
@classmethod
def create_for_statement(
cls,
statement: Union[Select, FromStatement],
compiler: Optional[SQLCompiler],
**kw: Any,
) -> ORMCompileState: ...
def _append_dedupe_col_collection(self, obj, col_collection):
dedupe = self.dedupe_columns
if obj not in dedupe:
dedupe.add(obj)
col_collection.append(obj)
@classmethod
def _column_naming_convention(
cls, label_style: SelectLabelStyle, legacy: bool
) -> _LabelConventionCallable:
if legacy:
def name(col, col_name=None):
if col_name:
return col_name
else:
return getattr(col, "key")
return name
else:
return SelectState._column_naming_convention(label_style)
@classmethod
def get_column_descriptions(cls, statement):
return _column_descriptions(statement)
@classmethod
def orm_pre_session_exec(
cls,
session,
statement,
params,
execution_options,
bind_arguments,
is_pre_event,
):
# consume result-level load_options. These may have been set up
# in an ORMExecuteState hook
(
load_options,
execution_options,
) = QueryContext.default_load_options.from_execution_options(
"_sa_orm_load_options",
{
"populate_existing",
"autoflush",
"yield_per",
"identity_token",
"sa_top_level_orm_context",
},
execution_options,
statement._execution_options,
)
# default execution options for ORM results:
# 1. _result_disable_adapt_to_context=True
# this will disable the ResultSetMetadata._adapt_to_context()
# step which we don't need, as we have result processors cached
# against the original SELECT statement before caching.
if "sa_top_level_orm_context" in execution_options:
ctx = execution_options["sa_top_level_orm_context"]
execution_options = ctx.query._execution_options.merge_with(
ctx.execution_options, execution_options
)
if not execution_options:
execution_options = _orm_load_exec_options
else:
execution_options = execution_options.union(_orm_load_exec_options)
# would have been placed here by legacy Query only
if load_options._yield_per:
execution_options = execution_options.union(
{"yield_per": load_options._yield_per}
)
if (
getattr(statement._compile_options, "_current_path", None)
and len(statement._compile_options._current_path) > 10
and execution_options.get("compiled_cache", True) is not None
):
execution_options: util.immutabledict[str, Any] = (
execution_options.union(
{
"compiled_cache": None,
"_cache_disable_reason": "excess depth for "
"ORM loader options",
}
)
)
bind_arguments["clause"] = statement
# new in 1.4 - the coercions system is leveraged to allow the
# "subject" mapper of a statement be propagated to the top
# as the statement is built. "subject" mapper is the generally
# standard object used as an identifier for multi-database schemes.
# we are here based on the fact that _propagate_attrs contains
# "compile_state_plugin": "orm". The "plugin_subject"
# needs to be present as well.
try:
plugin_subject = statement._propagate_attrs["plugin_subject"]
except KeyError:
assert False, "statement had 'orm' plugin but no plugin_subject"
else:
if plugin_subject:
bind_arguments["mapper"] = plugin_subject.mapper
if not is_pre_event and load_options._autoflush:
session._autoflush()
return statement, execution_options
@classmethod
def orm_setup_cursor_result(
cls,
session,
statement,
params,
execution_options,
bind_arguments,
result,
):
execution_context = result.context
compile_state = execution_context.compiled.compile_state
# cover edge case where ORM entities used in legacy select
# were passed to session.execute:
# session.execute(legacy_select([User.id, User.name]))
# see test_query->test_legacy_tuple_old_select
load_options = execution_options.get(
"_sa_orm_load_options", QueryContext.default_load_options
)
if compile_state.compile_options._is_star:
return result
querycontext = QueryContext(
compile_state,
statement,
params,
session,
load_options,
execution_options,
bind_arguments,
)
return loading.instances(result, querycontext)
@property
def _lead_mapper_entities(self):
"""return all _MapperEntity objects in the lead entities collection.
Does **not** include entities that have been replaced by
with_entities(), with_only_columns()
"""
return [
ent for ent in self._entities if isinstance(ent, _MapperEntity)
]
def _create_with_polymorphic_adapter(self, ext_info, selectable):
"""given MapperEntity or ORMColumnEntity, setup polymorphic loading
if called for by the Mapper.
As of #8168 in 2.0.0rc1, polymorphic adapters, which greatly increase
the complexity of the query creation process, are not used at all
except in the quasi-legacy cases of with_polymorphic referring to an
alias and/or subquery. This would apply to concrete polymorphic
loading, and joined inheritance where a subquery is
passed to with_polymorphic (which is completely unnecessary in modern
use).
"""
if (
not ext_info.is_aliased_class
and ext_info.mapper.persist_selectable
not in self._polymorphic_adapters
):
for mp in ext_info.mapper.iterate_to_root():
self._mapper_loads_polymorphically_with(
mp,
ORMAdapter(
_TraceAdaptRole.WITH_POLYMORPHIC_ADAPTER,
mp,
equivalents=mp._equivalent_columns,
selectable=selectable,
),
)
def _mapper_loads_polymorphically_with(self, mapper, adapter):
for m2 in mapper._with_polymorphic_mappers or [mapper]:
self._polymorphic_adapters[m2] = adapter
for m in m2.iterate_to_root():
self._polymorphic_adapters[m.local_table] = adapter
@classmethod
def _create_entities_collection(cls, query, legacy):
raise NotImplementedError(
"this method only works for ORMSelectCompileState"
)
class DMLReturningColFilter:
"""an adapter used for the DML RETURNING case.
Has a subset of the interface used by
:class:`.ORMAdapter` and is used for :class:`._QueryEntity`
instances to set up their columns as used in RETURNING for a
DML statement.
"""
__slots__ = ("mapper", "columns", "__weakref__")
def __init__(self, target_mapper, immediate_dml_mapper):
if (
immediate_dml_mapper is not None
and target_mapper.local_table
is not immediate_dml_mapper.local_table
):
# joined inh, or in theory other kinds of multi-table mappings
self.mapper = immediate_dml_mapper
else:
# single inh, normal mappings, etc.
self.mapper = target_mapper
self.columns = self.columns = util.WeakPopulateDict(
self.adapt_check_present # type: ignore
)
def __call__(self, col, as_filter):
for cc in sql_util._find_columns(col):
c2 = self.adapt_check_present(cc)
if c2 is not None:
return col
else:
return None
def adapt_check_present(self, col):
mapper = self.mapper
prop = mapper._columntoproperty.get(col, None)
if prop is None:
return None
return mapper.local_table.c.corresponding_column(col)
@sql.base.CompileState.plugin_for("orm", "orm_from_statement")
class ORMFromStatementCompileState(ORMCompileState):
_from_obj_alias = None
_has_mapper_entities = False
statement_container: FromStatement
requested_statement: Union[SelectBase, TextClause, UpdateBase]
dml_table: Optional[_DMLTableElement] = None
_has_orm_entities = False
multi_row_eager_loaders = False
eager_adding_joins = False
compound_eager_adapter = None
extra_criteria_entities = _EMPTY_DICT
eager_joins = _EMPTY_DICT
@classmethod
def create_for_statement(
cls,
statement_container: Union[Select, FromStatement],
compiler: Optional[SQLCompiler],
**kw: Any,
) -> ORMFromStatementCompileState:
assert isinstance(statement_container, FromStatement)
if compiler is not None and compiler.stack:
raise sa_exc.CompileError(
"The ORM FromStatement construct only supports being "
"invoked as the topmost statement, as it is only intended to "
"define how result rows should be returned."
)
self = cls.__new__(cls)
self._primary_entity = None
self.use_legacy_query_style = (
statement_container._compile_options._use_legacy_query_style
)
self.statement_container = self.select_statement = statement_container
self.requested_statement = statement = statement_container.element
if statement.is_dml:
self.dml_table = statement.table
self.is_dml_returning = True
self._entities = []
self._polymorphic_adapters = {}
self.compile_options = statement_container._compile_options
if (
self.use_legacy_query_style
and isinstance(statement, expression.SelectBase)
and not statement._is_textual
and not statement.is_dml
and statement._label_style is LABEL_STYLE_NONE
):
self.statement = statement.set_label_style(
LABEL_STYLE_TABLENAME_PLUS_COL
)
else:
self.statement = statement
self._label_convention = self._column_naming_convention(
(
statement._label_style
if not statement._is_textual and not statement.is_dml
else LABEL_STYLE_NONE
),
self.use_legacy_query_style,
)
_QueryEntity.to_compile_state(
self,
statement_container._raw_columns,
self._entities,
is_current_entities=True,
)
self.current_path = statement_container._compile_options._current_path
self._init_global_attributes(
statement_container,
compiler,
process_criteria_for_toplevel=False,
toplevel=True,
)
if statement_container._with_options:
for opt in statement_container._with_options:
if opt._is_compile_state:
opt.process_compile_state(self)
if statement_container._with_context_options:
for fn, key in statement_container._with_context_options:
fn(self)
self.primary_columns = []
self.secondary_columns = []
self.dedupe_columns = set()
self.create_eager_joins = []
self._fallback_from_clauses = []
self.order_by = None
if isinstance(self.statement, expression.TextClause):
# TextClause has no "column" objects at all. for this case,
# we generate columns from our _QueryEntity objects, then
# flip on all the "please match no matter what" parameters.
self.extra_criteria_entities = {}
for entity in self._entities:
entity.setup_compile_state(self)
compiler._ordered_columns = compiler._textual_ordered_columns = (
False
)
# enable looser result column matching. this is shown to be
# needed by test_query.py::TextTest
compiler._loose_column_name_matching = True
for c in self.primary_columns:
compiler.process(
c,
within_columns_clause=True,
add_to_result_map=compiler._add_to_result_map,
)
else:
# for everyone else, Select, Insert, Update, TextualSelect, they
# have column objects already. After much
# experimentation here, the best approach seems to be, use
# those columns completely, don't interfere with the compiler
# at all; just in ORM land, use an adapter to convert from
# our ORM columns to whatever columns are in the statement,
# before we look in the result row. Adapt on names
# to accept cases such as issue #9217, however also allow
# this to be overridden for cases such as #9273.
self._from_obj_alias = ORMStatementAdapter(
_TraceAdaptRole.ADAPT_FROM_STATEMENT,
self.statement,
adapt_on_names=statement_container._adapt_on_names,
)
return self
def _adapt_col_list(self, cols, current_adapter):
return cols
def _get_current_adapter(self):
return None
def setup_dml_returning_compile_state(self, dml_mapper):
"""used by BulkORMInsert (and Update / Delete?) to set up a handler
for RETURNING to return ORM objects and expressions
"""
target_mapper = self.statement._propagate_attrs.get(
"plugin_subject", None
)
adapter = DMLReturningColFilter(target_mapper, dml_mapper)
if self.compile_options._is_star and (len(self._entities) != 1):
raise sa_exc.CompileError(
"Can't generate ORM query that includes multiple expressions "
"at the same time as '*'; query for '*' alone if present"
)
for entity in self._entities:
entity.setup_dml_returning_compile_state(self, adapter)
class FromStatement(GroupedElement, Generative, TypedReturnsRows[Unpack[_Ts]]):
"""Core construct that represents a load of ORM objects from various
:class:`.ReturnsRows` and other classes including:
:class:`.Select`, :class:`.TextClause`, :class:`.TextualSelect`,
:class:`.CompoundSelect`, :class`.Insert`, :class:`.Update`,
and in theory, :class:`.Delete`.
"""
__visit_name__ = "orm_from_statement"
_compile_options = ORMFromStatementCompileState.default_compile_options
_compile_state_factory = ORMFromStatementCompileState.create_for_statement
_for_update_arg = None
element: Union[ExecutableReturnsRows, TextClause]
_adapt_on_names: bool
_traverse_internals = [
("_raw_columns", InternalTraversal.dp_clauseelement_list),
("element", InternalTraversal.dp_clauseelement),
] + Executable._executable_traverse_internals
_cache_key_traversal = _traverse_internals + [
("_compile_options", InternalTraversal.dp_has_cache_key)
]
is_from_statement = True
def __init__(
self,
entities: Iterable[_ColumnsClauseArgument[Any]],
element: Union[ExecutableReturnsRows, TextClause],
_adapt_on_names: bool = True,
):
self._raw_columns = [
coercions.expect(
roles.ColumnsClauseRole,
ent,
apply_propagate_attrs=self,
post_inspect=True,
)
for ent in util.to_list(entities)
]
self.element = element
self.is_dml = element.is_dml
self.is_select = element.is_select
self.is_delete = element.is_delete
self.is_insert = element.is_insert
self.is_update = element.is_update
self._label_style = (
element._label_style if is_select_base(element) else None
)
self._adapt_on_names = _adapt_on_names
def _compiler_dispatch(self, compiler, **kw):
"""provide a fixed _compiler_dispatch method.
This is roughly similar to using the sqlalchemy.ext.compiler
``@compiles`` extension.
"""
compile_state = self._compile_state_factory(self, compiler, **kw)
toplevel = not compiler.stack
if toplevel:
compiler.compile_state = compile_state
return compiler.process(compile_state.statement, **kw)
@property
def column_descriptions(self):
"""Return a :term:`plugin-enabled` 'column descriptions' structure
referring to the columns which are SELECTed by this statement.
See the section :ref:`queryguide_inspection` for an overview
of this feature.
.. seealso::
:ref:`queryguide_inspection` - ORM background
"""
meth = cast(
ORMSelectCompileState, SelectState.get_plugin_class(self)
).get_column_descriptions
return meth(self)
def _ensure_disambiguated_names(self):
return self
def get_children(self, **kw):
yield from itertools.chain.from_iterable(
element._from_objects for element in self._raw_columns
)
yield from super().get_children(**kw)
@property
def _all_selected_columns(self):
return self.element._all_selected_columns
@property
def _return_defaults(self):
return self.element._return_defaults if is_dml(self.element) else None
@property
def _returning(self):
return self.element._returning if is_dml(self.element) else None
@property
def _inline(self):
return self.element._inline if is_insert_update(self.element) else None
@sql.base.CompileState.plugin_for("orm", "compound_select")
class CompoundSelectCompileState(
AutoflushOnlyORMCompileState, CompoundSelectState
):
pass
@sql.base.CompileState.plugin_for("orm", "select")
class ORMSelectCompileState(ORMCompileState, SelectState):
_already_joined_edges = ()
_memoized_entities = _EMPTY_DICT