Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# orm/bulk_persistence.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: ignore-errors
10"""additional ORM persistence classes related to "bulk" operations,
11specifically outside of the flush() process.
13"""
15from __future__ import annotations
17from typing import Any
18from typing import cast
19from typing import Dict
20from typing import Iterable
21from typing import Literal
22from typing import Optional
23from typing import overload
24from typing import TYPE_CHECKING
25from typing import TypeVar
26from typing import Union
28from . import attributes
29from . import context
30from . import evaluator
31from . import exc as orm_exc
32from . import loading
33from . import persistence
34from .base import NO_VALUE
35from .context import _AbstractORMCompileState
36from .context import _ORMFromStatementCompileState
37from .context import FromStatement
38from .context import QueryContext
39from .interfaces import PropComparator
40from .. import exc as sa_exc
41from .. import util
42from ..engine import Dialect
43from ..engine import result as _result
44from ..sql import coercions
45from ..sql import dml
46from ..sql import expression
47from ..sql import roles
48from ..sql import select
49from ..sql import sqltypes
50from ..sql.base import _entity_namespace_key
51from ..sql.base import CompileState
52from ..sql.base import Options
53from ..sql.dml import DeleteDMLState
54from ..sql.dml import InsertDMLState
55from ..sql.dml import UpdateDMLState
56from ..util import EMPTY_DICT
57from ..util.typing import TupleAny
58from ..util.typing import Unpack
60if TYPE_CHECKING:
61 from ._typing import DMLStrategyArgument
62 from ._typing import OrmExecuteOptionsParameter
63 from ._typing import SynchronizeSessionArgument
64 from .mapper import Mapper
65 from .session import _BindArguments
66 from .session import ORMExecuteState
67 from .session import Session
68 from .session import SessionTransaction
69 from .state import InstanceState
70 from ..engine import Connection
71 from ..engine import cursor
72 from ..engine.interfaces import _CoreAnyExecuteParams
74_O = TypeVar("_O", bound=object)
77@overload
78def _bulk_insert(
79 mapper: Mapper[_O],
80 mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],
81 session_transaction: SessionTransaction,
82 *,
83 isstates: bool,
84 return_defaults: bool,
85 render_nulls: bool,
86 use_orm_insert_stmt: Literal[None] = ...,
87 execution_options: Optional[OrmExecuteOptionsParameter] = ...,
88) -> None: ...
91@overload
92def _bulk_insert(
93 mapper: Mapper[_O],
94 mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],
95 session_transaction: SessionTransaction,
96 *,
97 isstates: bool,
98 return_defaults: bool,
99 render_nulls: bool,
100 use_orm_insert_stmt: Optional[dml.Insert] = ...,
101 execution_options: Optional[OrmExecuteOptionsParameter] = ...,
102) -> cursor.CursorResult[Any]: ...
105def _bulk_insert(
106 mapper: Mapper[_O],
107 mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],
108 session_transaction: SessionTransaction,
109 *,
110 isstates: bool,
111 return_defaults: bool,
112 render_nulls: bool,
113 use_orm_insert_stmt: Optional[dml.Insert] = None,
114 execution_options: Optional[OrmExecuteOptionsParameter] = None,
115) -> Optional[cursor.CursorResult[Any]]:
116 base_mapper = mapper.base_mapper
118 if session_transaction.session.connection_callable:
119 raise NotImplementedError(
120 "connection_callable / per-instance sharding "
121 "not supported in bulk_insert()"
122 )
124 if isstates:
125 if TYPE_CHECKING:
126 mappings = cast(Iterable[InstanceState[_O]], mappings)
128 if return_defaults:
129 # list of states allows us to attach .key for return_defaults case
130 states = [(state, state.dict) for state in mappings]
131 mappings = [dict_ for (state, dict_) in states]
132 else:
133 mappings = [state.dict for state in mappings]
134 else:
135 if TYPE_CHECKING:
136 mappings = cast(Iterable[Dict[str, Any]], mappings)
138 if return_defaults:
139 # use dictionaries given, so that newly populated defaults
140 # can be delivered back to the caller (see #11661). This is **not**
141 # compatible with other use cases such as a session-executed
142 # insert() construct, as this will confuse the case of
143 # insert-per-subclass for joined inheritance cases (see
144 # test_bulk_statements.py::BulkDMLReturningJoinedInhTest).
145 #
146 # So in this conditional, we have **only** called
147 # session.bulk_insert_mappings() which does not have this
148 # requirement
149 mappings = list(mappings)
150 else:
151 # for all other cases we need to establish a local dictionary
152 # so that the incoming dictionaries aren't mutated
153 mappings = [dict(m) for m in mappings]
154 _expand_other_attrs(mapper, mappings)
156 connection = session_transaction.connection(base_mapper)
158 return_result: Optional[cursor.CursorResult[Any]] = None
160 mappers_to_run = [
161 (table, mp)
162 for table, mp in base_mapper._sorted_tables.items()
163 if table in mapper._pks_by_table
164 ]
166 if return_defaults:
167 # not used by new-style bulk inserts, only used for legacy
168 bookkeeping = True
169 elif len(mappers_to_run) > 1:
170 # if we have more than one table, mapper to run where we will be
171 # either horizontally splicing, or copying values between tables,
172 # we need the "bookkeeping" / deterministic returning order
173 bookkeeping = True
174 else:
175 bookkeeping = False
177 for table, super_mapper in mappers_to_run:
178 # find bindparams in the statement. For bulk, we don't really know if
179 # a key in the params applies to a different table since we are
180 # potentially inserting for multiple tables here; looking at the
181 # bindparam() is a lot more direct. in most cases this will
182 # use _generate_cache_key() which is memoized, although in practice
183 # the ultimate statement that's executed is probably not the same
184 # object so that memoization might not matter much.
185 extra_bp_names = (
186 [
187 b.key
188 for b in use_orm_insert_stmt._get_embedded_bindparams()
189 if b.key in mappings[0]
190 ]
191 if use_orm_insert_stmt is not None
192 else ()
193 )
195 records = (
196 (
197 None,
198 state_dict,
199 params,
200 mapper,
201 connection,
202 value_params,
203 has_all_pks,
204 has_all_defaults,
205 )
206 for (
207 state,
208 state_dict,
209 params,
210 mp,
211 conn,
212 value_params,
213 has_all_pks,
214 has_all_defaults,
215 ) in persistence._collect_insert_commands(
216 table,
217 ((None, mapping, mapper, connection) for mapping in mappings),
218 bulk=True,
219 return_defaults=bookkeeping,
220 render_nulls=render_nulls,
221 include_bulk_keys=extra_bp_names,
222 )
223 )
225 result = persistence._emit_insert_statements(
226 base_mapper,
227 None,
228 super_mapper,
229 table,
230 records,
231 bookkeeping=bookkeeping,
232 use_orm_insert_stmt=use_orm_insert_stmt,
233 execution_options=execution_options,
234 )
235 if use_orm_insert_stmt is not None:
236 if not use_orm_insert_stmt._returning or return_result is None:
237 return_result = result
238 elif result.returns_rows:
239 assert bookkeeping
240 return_result = return_result.splice_horizontally(result)
242 if return_defaults and isstates:
243 identity_cls = mapper._identity_class
244 identity_props = [p.key for p in mapper._identity_key_props]
245 for state, dict_ in states:
246 state.key = (
247 identity_cls,
248 tuple([dict_[key] for key in identity_props]),
249 None,
250 )
252 if use_orm_insert_stmt is not None:
253 assert return_result is not None
254 return return_result
257@overload
258def _bulk_update(
259 mapper: Mapper[Any],
260 mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],
261 session_transaction: SessionTransaction,
262 *,
263 isstates: bool,
264 update_changed_only: bool,
265 use_orm_update_stmt: Literal[None] = ...,
266 enable_check_rowcount: bool = True,
267) -> None: ...
270@overload
271def _bulk_update(
272 mapper: Mapper[Any],
273 mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],
274 session_transaction: SessionTransaction,
275 *,
276 isstates: bool,
277 update_changed_only: bool,
278 use_orm_update_stmt: Optional[dml.Update] = ...,
279 enable_check_rowcount: bool = True,
280) -> _result.Result[Unpack[TupleAny]]: ...
283def _bulk_update(
284 mapper: Mapper[Any],
285 mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],
286 session_transaction: SessionTransaction,
287 *,
288 isstates: bool,
289 update_changed_only: bool,
290 use_orm_update_stmt: Optional[dml.Update] = None,
291 enable_check_rowcount: bool = True,
292) -> Optional[_result.Result[Unpack[TupleAny]]]:
293 base_mapper = mapper.base_mapper
295 search_keys = mapper._primary_key_propkeys
296 if mapper._version_id_prop:
297 search_keys = {mapper._version_id_prop.key}.union(search_keys)
299 def _changed_dict(mapper, state):
300 return {
301 k: v
302 for k, v in state.dict.items()
303 if k in state.committed_state or k in search_keys
304 }
306 if isstates:
307 if update_changed_only:
308 mappings = [_changed_dict(mapper, state) for state in mappings]
309 else:
310 mappings = [state.dict for state in mappings]
311 else:
312 mappings = [dict(m) for m in mappings]
313 _expand_other_attrs(mapper, mappings)
315 if session_transaction.session.connection_callable:
316 raise NotImplementedError(
317 "connection_callable / per-instance sharding "
318 "not supported in bulk_update()"
319 )
321 connection = session_transaction.connection(base_mapper)
323 # find bindparams in the statement. see _bulk_insert for similar
324 # notes for the insert case
325 extra_bp_names = (
326 [
327 b.key
328 for b in use_orm_update_stmt._get_embedded_bindparams()
329 if b.key in mappings[0]
330 ]
331 if use_orm_update_stmt is not None
332 else ()
333 )
335 for table, super_mapper in base_mapper._sorted_tables.items():
336 if not mapper.isa(super_mapper) or table not in mapper._pks_by_table:
337 continue
339 records = persistence._collect_update_commands(
340 None,
341 table,
342 (
343 (
344 None,
345 mapping,
346 mapper,
347 connection,
348 (
349 mapping[mapper._version_id_prop.key]
350 if mapper._version_id_prop
351 else None
352 ),
353 )
354 for mapping in mappings
355 ),
356 bulk=True,
357 use_orm_update_stmt=use_orm_update_stmt,
358 include_bulk_keys=extra_bp_names,
359 )
360 persistence._emit_update_statements(
361 base_mapper,
362 None,
363 super_mapper,
364 table,
365 records,
366 bookkeeping=False,
367 use_orm_update_stmt=use_orm_update_stmt,
368 enable_check_rowcount=enable_check_rowcount,
369 )
371 if use_orm_update_stmt is not None:
372 return _result.null_result()
375def _expand_other_attrs(
376 mapper: Mapper[Any], mappings: Iterable[Dict[str, Any]]
377) -> None:
378 all_attrs = mapper.all_orm_descriptors
380 attr_keys = set(all_attrs.keys())
382 bulk_dml_setters = {
383 key: setter
384 for key, setter in (
385 (key, attr._bulk_dml_setter(key))
386 for key, attr in (
387 (key, _entity_namespace_key(mapper, key, default=NO_VALUE))
388 for key in attr_keys
389 )
390 if attr is not NO_VALUE and isinstance(attr, PropComparator)
391 )
392 if setter is not None
393 }
394 setters_todo = set(bulk_dml_setters)
395 if not setters_todo:
396 return
398 for mapping in mappings:
399 for key in setters_todo.intersection(mapping):
400 bulk_dml_setters[key](mapping)
403class _ORMDMLState(_AbstractORMCompileState):
404 is_dml_returning = True
405 from_statement_ctx: Optional[_ORMFromStatementCompileState] = None
407 @classmethod
408 def _get_orm_crud_kv_pairs(
409 cls, mapper, statement, kv_iterator, needs_to_be_cacheable
410 ):
411 core_get_crud_kv_pairs = UpdateDMLState._get_crud_kv_pairs
413 for k, v in kv_iterator:
414 k = coercions.expect(roles.DMLColumnRole, k)
416 if isinstance(k, str):
417 desc = _entity_namespace_key(mapper, k, default=NO_VALUE)
418 if not isinstance(desc, PropComparator):
419 yield (
420 coercions.expect(roles.DMLColumnRole, k),
421 (
422 coercions.expect(
423 roles.ExpressionElementRole,
424 v,
425 type_=sqltypes.NullType(),
426 is_crud=True,
427 )
428 if needs_to_be_cacheable
429 else v
430 ),
431 )
432 else:
433 yield from core_get_crud_kv_pairs(
434 statement,
435 desc._bulk_update_tuples(v),
436 needs_to_be_cacheable,
437 )
438 elif "entity_namespace" in k._annotations:
439 k_anno = k._annotations
440 attr = _entity_namespace_key(
441 k_anno["entity_namespace"], k_anno["proxy_key"]
442 )
443 assert isinstance(attr, PropComparator)
444 yield from core_get_crud_kv_pairs(
445 statement,
446 attr._bulk_update_tuples(v),
447 needs_to_be_cacheable,
448 )
449 else:
450 yield (
451 k,
452 (
453 v
454 if not needs_to_be_cacheable
455 else coercions.expect(
456 roles.ExpressionElementRole,
457 v,
458 type_=sqltypes.NullType(),
459 is_crud=True,
460 )
461 ),
462 )
464 @classmethod
465 def _get_dml_plugin_subject(cls, statement):
466 plugin_subject = statement.table._propagate_attrs.get("plugin_subject")
468 if (
469 not plugin_subject
470 or not plugin_subject.mapper
471 or plugin_subject
472 is not statement._propagate_attrs["plugin_subject"]
473 ):
474 return None
475 return plugin_subject
477 @classmethod
478 def _get_multi_crud_kv_pairs(cls, statement, kv_iterator):
479 plugin_subject = cls._get_dml_plugin_subject(statement)
481 if not plugin_subject:
482 return UpdateDMLState._get_multi_crud_kv_pairs(
483 statement, kv_iterator
484 )
486 return [
487 dict(
488 cls._get_orm_crud_kv_pairs(
489 plugin_subject.mapper, statement, value_dict.items(), False
490 )
491 )
492 for value_dict in kv_iterator
493 ]
495 @classmethod
496 def _get_crud_kv_pairs(cls, statement, kv_iterator, needs_to_be_cacheable):
497 assert (
498 needs_to_be_cacheable
499 ), "no test coverage for needs_to_be_cacheable=False"
501 plugin_subject = cls._get_dml_plugin_subject(statement)
503 if not plugin_subject:
504 return UpdateDMLState._get_crud_kv_pairs(
505 statement, kv_iterator, needs_to_be_cacheable
506 )
507 return list(
508 cls._get_orm_crud_kv_pairs(
509 plugin_subject.mapper,
510 statement,
511 kv_iterator,
512 needs_to_be_cacheable,
513 )
514 )
516 @classmethod
517 def get_entity_description(cls, statement):
518 ext_info = statement.table._annotations["parententity"]
519 mapper = ext_info.mapper
520 if ext_info.is_aliased_class:
521 _label_name = ext_info.name
522 else:
523 _label_name = mapper.class_.__name__
525 return {
526 "name": _label_name,
527 "type": mapper.class_,
528 "expr": ext_info.entity,
529 "entity": ext_info.entity,
530 "table": mapper.local_table,
531 }
533 @classmethod
534 def get_returning_column_descriptions(cls, statement):
535 def _ent_for_col(c):
536 return c._annotations.get("parententity", None)
538 def _attr_for_col(c, ent):
539 if ent is None:
540 return c
541 proxy_key = c._annotations.get("proxy_key", None)
542 if not proxy_key:
543 return c
544 else:
545 return getattr(ent.entity, proxy_key, c)
547 return [
548 {
549 "name": c.key,
550 "type": c.type,
551 "expr": _attr_for_col(c, ent),
552 "aliased": ent.is_aliased_class,
553 "entity": ent.entity,
554 }
555 for c, ent in [
556 (c, _ent_for_col(c)) for c in statement._all_selected_columns
557 ]
558 ]
560 def _setup_orm_returning(
561 self,
562 compiler,
563 orm_level_statement,
564 dml_level_statement,
565 dml_mapper,
566 *,
567 use_supplemental_cols=True,
568 ):
569 """establish ORM column handlers for an INSERT, UPDATE, or DELETE
570 which uses explicit returning().
572 called within compilation level create_for_statement.
574 The _return_orm_returning() method then receives the Result
575 after the statement was executed, and applies ORM loading to the
576 state that we first established here.
578 """
580 if orm_level_statement._returning:
581 fs = FromStatement(
582 orm_level_statement._returning,
583 dml_level_statement,
584 _adapt_on_names=False,
585 )
586 fs = fs.execution_options(**orm_level_statement._execution_options)
587 fs = fs.options(*orm_level_statement._with_options)
588 self.select_statement = fs
589 self.from_statement_ctx = fsc = (
590 _ORMFromStatementCompileState.create_for_statement(
591 fs, compiler
592 )
593 )
594 fsc.setup_dml_returning_compile_state(dml_mapper)
596 dml_level_statement = dml_level_statement._generate()
597 dml_level_statement._returning = ()
599 cols_to_return = [c for c in fsc.primary_columns if c is not None]
601 # since we are splicing result sets together, make sure there
602 # are columns of some kind returned in each result set
603 if not cols_to_return:
604 cols_to_return.extend(dml_mapper.primary_key)
606 if use_supplemental_cols:
607 dml_level_statement = dml_level_statement.return_defaults(
608 # this is a little weird looking, but by passing
609 # primary key as the main list of cols, this tells
610 # return_defaults to omit server-default cols (and
611 # actually all cols, due to some weird thing we should
612 # clean up in crud.py).
613 # Since we have cols_to_return, just return what we asked
614 # for (plus primary key, which ORM persistence needs since
615 # we likely set bookkeeping=True here, which is another
616 # whole thing...). We dont want to clutter the
617 # statement up with lots of other cols the user didn't
618 # ask for. see #9685
619 *dml_mapper.primary_key,
620 supplemental_cols=cols_to_return,
621 )
622 else:
623 dml_level_statement = dml_level_statement.returning(
624 *cols_to_return
625 )
627 return dml_level_statement
629 @classmethod
630 def _return_orm_returning(
631 cls,
632 session,
633 statement,
634 params,
635 execution_options,
636 bind_arguments,
637 result,
638 ):
639 execution_context = result.context
640 compile_state = execution_context.compiled.compile_state
642 if (
643 compile_state.from_statement_ctx
644 and not compile_state.from_statement_ctx.compile_options._is_star
645 ):
646 load_options = execution_options.get(
647 "_sa_orm_load_options", QueryContext.default_load_options
648 )
650 querycontext = QueryContext(
651 compile_state.from_statement_ctx,
652 compile_state.select_statement,
653 statement,
654 params,
655 session,
656 load_options,
657 execution_options,
658 bind_arguments,
659 )
660 return loading.instances(result, querycontext)
661 else:
662 return result
665class _BulkUDCompileState(_ORMDMLState):
666 class default_update_options(Options):
667 _dml_strategy: DMLStrategyArgument = "auto"
668 _synchronize_session: SynchronizeSessionArgument = "auto"
669 _can_use_returning: bool = False
670 _is_delete_using: bool = False
671 _is_update_from: bool = False
672 _autoflush: bool = True
673 _subject_mapper: Optional[Mapper[Any]] = None
674 _resolved_values = EMPTY_DICT
675 _eval_condition = None
676 _matched_rows = None
677 _identity_token = None
678 _populate_existing: bool = False
680 @classmethod
681 def can_use_returning(
682 cls,
683 dialect: Dialect,
684 mapper: Mapper[Any],
685 *,
686 is_multitable: bool = False,
687 is_update_from: bool = False,
688 is_delete_using: bool = False,
689 is_executemany: bool = False,
690 ) -> bool:
691 raise NotImplementedError()
693 @classmethod
694 def orm_pre_session_exec(
695 cls,
696 session,
697 statement,
698 params,
699 execution_options,
700 bind_arguments,
701 is_pre_event,
702 ):
703 (
704 update_options,
705 execution_options,
706 ) = _BulkUDCompileState.default_update_options.from_execution_options(
707 "_sa_orm_update_options",
708 {
709 "synchronize_session",
710 "autoflush",
711 "populate_existing",
712 "identity_token",
713 "is_delete_using",
714 "is_update_from",
715 "dml_strategy",
716 },
717 execution_options,
718 statement._execution_options,
719 )
720 bind_arguments["clause"] = statement
721 try:
722 plugin_subject = statement._propagate_attrs["plugin_subject"]
723 except KeyError:
724 assert False, "statement had 'orm' plugin but no plugin_subject"
725 else:
726 if plugin_subject:
727 bind_arguments["mapper"] = plugin_subject.mapper
728 update_options += {"_subject_mapper": plugin_subject.mapper}
730 if "parententity" not in statement.table._annotations:
731 update_options += {"_dml_strategy": "core_only"}
732 elif not isinstance(params, list):
733 if update_options._dml_strategy == "auto":
734 update_options += {"_dml_strategy": "orm"}
735 elif update_options._dml_strategy == "bulk":
736 raise sa_exc.InvalidRequestError(
737 'Can\'t use "bulk" ORM insert strategy without '
738 "passing separate parameters"
739 )
740 else:
741 if update_options._dml_strategy == "auto":
742 update_options += {"_dml_strategy": "bulk"}
744 sync = update_options._synchronize_session
745 if sync is not None:
746 if sync not in ("auto", "evaluate", "fetch", False):
747 raise sa_exc.ArgumentError(
748 "Valid strategies for session synchronization "
749 "are 'auto', 'evaluate', 'fetch', False"
750 )
751 if update_options._dml_strategy == "bulk" and sync == "fetch":
752 raise sa_exc.InvalidRequestError(
753 "The 'fetch' synchronization strategy is not available "
754 "for 'bulk' ORM updates (i.e. multiple parameter sets)"
755 )
757 if not is_pre_event:
758 if update_options._autoflush:
759 session._autoflush()
761 if update_options._dml_strategy == "orm":
762 if update_options._synchronize_session == "auto":
763 update_options = cls._do_pre_synchronize_auto(
764 session,
765 statement,
766 params,
767 execution_options,
768 bind_arguments,
769 update_options,
770 )
771 elif update_options._synchronize_session == "evaluate":
772 update_options = cls._do_pre_synchronize_evaluate(
773 session,
774 statement,
775 params,
776 execution_options,
777 bind_arguments,
778 update_options,
779 )
780 elif update_options._synchronize_session == "fetch":
781 update_options = cls._do_pre_synchronize_fetch(
782 session,
783 statement,
784 params,
785 execution_options,
786 bind_arguments,
787 update_options,
788 )
789 elif update_options._dml_strategy == "bulk":
790 if update_options._synchronize_session == "auto":
791 update_options += {"_synchronize_session": "evaluate"}
793 # indicators from the "pre exec" step that are then
794 # added to the DML statement, which will also be part of the cache
795 # key. The compile level create_for_statement() method will then
796 # consume these at compiler time.
797 statement = statement._annotate(
798 {
799 "synchronize_session": update_options._synchronize_session,
800 "is_delete_using": update_options._is_delete_using,
801 "is_update_from": update_options._is_update_from,
802 "dml_strategy": update_options._dml_strategy,
803 "can_use_returning": update_options._can_use_returning,
804 }
805 )
807 # disable result-level adapt_to_context. ORM UPDATE/DELETE with
808 # RETURNING uses a "two level" statement: the invoked ORM statement
809 # holds the user's returning() columns while the cached Core
810 # statement returns primary key + supplemental columns, so the two
811 # are not positionally aligned. adapt_to_context() would remap
812 # result keys positionally between them and mislabel columns on a
813 # cache hit; the ORM instead interprets rows via its own
814 # from_statement context, so the Core adaptation would interfere
815 # here.
816 if not execution_options:
817 execution_options = context._orm_load_exec_options
818 else:
819 execution_options = execution_options.union(
820 context._orm_load_exec_options
821 )
823 return (
824 statement,
825 util.immutabledict(execution_options).union(
826 {"_sa_orm_update_options": update_options}
827 ),
828 params,
829 )
831 @classmethod
832 def orm_setup_cursor_result(
833 cls,
834 session,
835 statement,
836 params,
837 execution_options,
838 bind_arguments,
839 result,
840 ):
841 # this stage of the execution is called after the
842 # do_orm_execute event hook. meaning for an extension like
843 # horizontal sharding, this step happens *within* the horizontal
844 # sharding event handler which calls session.execute() re-entrantly
845 # and will occur for each backend individually.
846 # the sharding extension then returns its own merged result from the
847 # individual ones we return here.
849 update_options = execution_options["_sa_orm_update_options"]
850 if update_options._dml_strategy == "orm":
851 if update_options._synchronize_session == "evaluate":
852 cls._do_post_synchronize_evaluate(
853 session, statement, result, update_options
854 )
855 elif update_options._synchronize_session == "fetch":
856 cls._do_post_synchronize_fetch(
857 session, statement, result, update_options
858 )
859 elif update_options._dml_strategy == "bulk":
860 if update_options._synchronize_session == "evaluate":
861 cls._do_post_synchronize_bulk_evaluate(
862 session, params, result, update_options
863 )
864 return result
866 return cls._return_orm_returning(
867 session,
868 statement,
869 params,
870 execution_options,
871 bind_arguments,
872 result,
873 )
875 @classmethod
876 def _adjust_for_extra_criteria(cls, global_attributes, ext_info):
877 """Apply extra criteria filtering.
879 For all distinct single-table-inheritance mappers represented in the
880 table being updated or deleted, produce additional WHERE criteria such
881 that only the appropriate subtypes are selected from the total results.
883 Additionally, add WHERE criteria originating from LoaderCriteriaOptions
884 collected from the statement.
886 """
888 return_crit = ()
890 adapter = ext_info._adapter if ext_info.is_aliased_class else None
892 if (
893 "additional_entity_criteria",
894 ext_info.mapper,
895 ) in global_attributes:
896 return_crit += tuple(
897 ae._resolve_where_criteria(ext_info)
898 for ae in global_attributes[
899 ("additional_entity_criteria", ext_info.mapper)
900 ]
901 if ae.include_aliases or ae.entity is ext_info
902 )
904 if ext_info.mapper._single_table_criterion is not None:
905 return_crit += (ext_info.mapper._single_table_criterion,)
907 if adapter:
908 return_crit = tuple(adapter.traverse(crit) for crit in return_crit)
910 return return_crit
912 @classmethod
913 def _interpret_returning_rows(cls, result, mapper, rows):
914 """return rows that indicate PK cols in mapper.primary_key position
915 for RETURNING rows.
917 Prior to 2.0.36, this method seemed to be written for some kind of
918 inheritance scenario but the scenario was unused for actual joined
919 inheritance, and the function instead seemed to perform some kind of
920 partial translation that would remove non-PK cols if the PK cols
921 happened to be first in the row, but not otherwise. The joined
922 inheritance walk feature here seems to have never been used as it was
923 always skipped by the "local_table" check.
925 As of 2.0.36 the function strips away non-PK cols and provides the
926 PK cols for the table in mapper PK order.
928 """
930 try:
931 if mapper.local_table is not mapper.base_mapper.local_table:
932 # TODO: dive more into how a local table PK is used for fetch
933 # sync, not clear if this is correct as it depends on the
934 # downstream routine to fetch rows using
935 # local_table.primary_key order
936 pk_keys = result._tuple_getter(mapper.local_table.primary_key)
937 else:
938 pk_keys = result._tuple_getter(mapper.primary_key)
939 except KeyError:
940 # can't use these rows, they don't have PK cols in them
941 # this is an unusual case where the user would have used
942 # .return_defaults()
943 return []
945 return [pk_keys(row) for row in rows]
947 @classmethod
948 def _get_matched_objects_on_criteria(cls, update_options, states):
949 mapper = update_options._subject_mapper
950 eval_condition = update_options._eval_condition
952 raw_data = [
953 (state.obj(), state, state.dict)
954 for state in states
955 if state.mapper.isa(mapper) and not state.expired
956 ]
958 identity_token = update_options._identity_token
959 if identity_token is not None:
960 raw_data = [
961 (obj, state, dict_)
962 for obj, state, dict_ in raw_data
963 if state.identity_token == identity_token
964 ]
966 result = []
967 for obj, state, dict_ in raw_data:
968 evaled_condition = eval_condition(obj)
970 # caution: don't use "in ()" or == here, _EXPIRE_OBJECT
971 # evaluates as True for all comparisons
972 if (
973 evaled_condition is True
974 or evaled_condition is evaluator._EXPIRED_OBJECT
975 ):
976 result.append(
977 (
978 obj,
979 state,
980 dict_,
981 evaled_condition is evaluator._EXPIRED_OBJECT,
982 )
983 )
984 return result
986 @classmethod
987 def _eval_condition_from_statement(cls, update_options, statement):
988 mapper = update_options._subject_mapper
989 target_cls = mapper.class_
991 evaluator_compiler = evaluator._EvaluatorCompiler(target_cls)
992 crit = ()
993 if statement._where_criteria:
994 crit += statement._where_criteria
996 global_attributes = {}
997 for opt in statement._with_options:
998 if opt._is_criteria_option:
999 opt.get_global_criteria(global_attributes)
1001 if global_attributes:
1002 crit += cls._adjust_for_extra_criteria(global_attributes, mapper)
1004 if crit:
1005 eval_condition = evaluator_compiler.process(*crit)
1006 else:
1007 # workaround for mypy https://github.com/python/mypy/issues/14027
1008 def _eval_condition(obj):
1009 return True
1011 eval_condition = _eval_condition
1013 return eval_condition
1015 @classmethod
1016 def _do_pre_synchronize_auto(
1017 cls,
1018 session,
1019 statement,
1020 params,
1021 execution_options,
1022 bind_arguments,
1023 update_options,
1024 ):
1025 """setup auto sync strategy
1028 "auto" checks if we can use "evaluate" first, then falls back
1029 to "fetch"
1031 evaluate is vastly more efficient for the common case
1032 where session is empty, only has a few objects, and the UPDATE
1033 statement can potentially match thousands/millions of rows.
1035 OTOH more complex criteria that fails to work with "evaluate"
1036 we would hope usually correlates with fewer net rows.
1038 """
1040 try:
1041 eval_condition = cls._eval_condition_from_statement(
1042 update_options, statement
1043 )
1045 except evaluator.UnevaluatableError:
1046 pass
1047 else:
1048 return update_options + {
1049 "_eval_condition": eval_condition,
1050 "_synchronize_session": "evaluate",
1051 }
1053 update_options += {"_synchronize_session": "fetch"}
1054 return cls._do_pre_synchronize_fetch(
1055 session,
1056 statement,
1057 params,
1058 execution_options,
1059 bind_arguments,
1060 update_options,
1061 )
1063 @classmethod
1064 def _do_pre_synchronize_evaluate(
1065 cls,
1066 session,
1067 statement,
1068 params,
1069 execution_options,
1070 bind_arguments,
1071 update_options,
1072 ):
1073 try:
1074 eval_condition = cls._eval_condition_from_statement(
1075 update_options, statement
1076 )
1078 except evaluator.UnevaluatableError as err:
1079 raise sa_exc.InvalidRequestError(
1080 'Could not evaluate current criteria in Python: "%s". '
1081 "Specify 'fetch' or False for the "
1082 "synchronize_session execution option." % err
1083 ) from err
1085 return update_options + {
1086 "_eval_condition": eval_condition,
1087 }
1089 @classmethod
1090 def _get_resolved_values(cls, mapper, statement):
1091 if statement._multi_values:
1092 return []
1093 elif statement._values:
1094 return list(statement._values.items())
1095 else:
1096 return []
1098 @classmethod
1099 def _resolved_keys_as_propnames(cls, mapper, resolved_values):
1100 values = []
1101 for k, v in resolved_values:
1102 if mapper and isinstance(k, expression.ColumnElement):
1103 try:
1104 attr = mapper._columntoproperty[k]
1105 except orm_exc.UnmappedColumnError:
1106 pass
1107 else:
1108 values.append((attr.key, v))
1109 else:
1110 raise sa_exc.InvalidRequestError(
1111 "Attribute name not found, can't be "
1112 "synchronized back to objects: %r" % k
1113 )
1114 return values
1116 @classmethod
1117 def _do_pre_synchronize_fetch(
1118 cls,
1119 session,
1120 statement,
1121 params,
1122 execution_options,
1123 bind_arguments,
1124 update_options,
1125 ):
1126 mapper = update_options._subject_mapper
1128 select_stmt = (
1129 select(*(mapper.primary_key + (mapper.select_identity_token,)))
1130 .select_from(mapper)
1131 .options(*statement._with_options)
1132 )
1133 select_stmt._where_criteria = statement._where_criteria
1135 # conditionally run the SELECT statement for pre-fetch, testing the
1136 # "bind" for if we can use RETURNING or not using the do_orm_execute
1137 # event. If RETURNING is available, the do_orm_execute event
1138 # will cancel the SELECT from being actually run.
1139 #
1140 # The way this is organized seems strange, why don't we just
1141 # call can_use_returning() before invoking the statement and get
1142 # answer?, why does this go through the whole execute phase using an
1143 # event? Answer: because we are integrating with extensions such
1144 # as the horizontal sharding extension that "multiplexes" an individual
1145 # statement run through multiple engines, and it uses
1146 # do_orm_execute() to do that.
1148 can_use_returning = None
1150 def skip_for_returning(orm_context: ORMExecuteState) -> Any:
1151 bind = orm_context.session.get_bind(**orm_context.bind_arguments)
1152 nonlocal can_use_returning
1154 per_bind_result = cls.can_use_returning(
1155 bind.dialect,
1156 mapper,
1157 is_update_from=update_options._is_update_from,
1158 is_delete_using=update_options._is_delete_using,
1159 is_executemany=orm_context.is_executemany,
1160 )
1162 if can_use_returning is not None:
1163 if can_use_returning != per_bind_result:
1164 raise sa_exc.InvalidRequestError(
1165 "For synchronize_session='fetch', can't mix multiple "
1166 "backends where some support RETURNING and others "
1167 "don't"
1168 )
1169 elif orm_context.is_executemany and not per_bind_result:
1170 raise sa_exc.InvalidRequestError(
1171 "For synchronize_session='fetch', can't use multiple "
1172 "parameter sets in ORM mode, which this backend does not "
1173 "support with RETURNING"
1174 )
1175 else:
1176 can_use_returning = per_bind_result
1178 if per_bind_result:
1179 return _result.null_result()
1180 else:
1181 return None
1183 result = session.execute(
1184 select_stmt,
1185 params,
1186 execution_options=execution_options,
1187 bind_arguments=bind_arguments,
1188 _add_event=skip_for_returning,
1189 )
1190 matched_rows = result.fetchall()
1192 return update_options + {
1193 "_matched_rows": matched_rows,
1194 "_can_use_returning": can_use_returning,
1195 }
1198@CompileState.plugin_for("orm", "insert")
1199class _BulkORMInsert(_ORMDMLState, InsertDMLState):
1200 class default_insert_options(Options):
1201 _dml_strategy: DMLStrategyArgument = "auto"
1202 _render_nulls: bool = False
1203 _return_defaults: bool = False
1204 _subject_mapper: Optional[Mapper[Any]] = None
1205 _autoflush: bool = True
1206 _populate_existing: bool = False
1208 select_statement: Optional[FromStatement] = None
1210 @classmethod
1211 def orm_pre_session_exec(
1212 cls,
1213 session,
1214 statement,
1215 params,
1216 execution_options,
1217 bind_arguments,
1218 is_pre_event,
1219 ):
1220 (
1221 insert_options,
1222 execution_options,
1223 ) = _BulkORMInsert.default_insert_options.from_execution_options(
1224 "_sa_orm_insert_options",
1225 {"dml_strategy", "autoflush", "populate_existing", "render_nulls"},
1226 execution_options,
1227 statement._execution_options,
1228 )
1229 bind_arguments["clause"] = statement
1230 try:
1231 plugin_subject = statement._propagate_attrs["plugin_subject"]
1232 except KeyError:
1233 assert False, "statement had 'orm' plugin but no plugin_subject"
1234 else:
1235 if plugin_subject:
1236 bind_arguments["mapper"] = plugin_subject.mapper
1237 insert_options += {"_subject_mapper": plugin_subject.mapper}
1239 if not params:
1240 if insert_options._dml_strategy == "auto":
1241 insert_options += {"_dml_strategy": "orm"}
1242 elif insert_options._dml_strategy == "bulk":
1243 raise sa_exc.InvalidRequestError(
1244 'Can\'t use "bulk" ORM insert strategy without '
1245 "passing separate parameters"
1246 )
1247 else:
1248 if insert_options._dml_strategy == "auto":
1249 insert_options += {"_dml_strategy": "bulk"}
1251 if insert_options._dml_strategy != "raw":
1252 # for ORM object loading, like ORMContext, we have to disable
1253 # result set adapt_to_context, because we will be generating a
1254 # new statement with specific columns that's cached inside of
1255 # an ORMFromStatementCompileState, which we will reuse for
1256 # each result.
1257 if not execution_options:
1258 execution_options = context._orm_load_exec_options
1259 else:
1260 execution_options = execution_options.union(
1261 context._orm_load_exec_options
1262 )
1264 if not is_pre_event and insert_options._autoflush:
1265 session._autoflush()
1267 statement = statement._annotate(
1268 {"dml_strategy": insert_options._dml_strategy}
1269 )
1271 return (
1272 statement,
1273 util.immutabledict(execution_options).union(
1274 {"_sa_orm_insert_options": insert_options}
1275 ),
1276 params,
1277 )
1279 @classmethod
1280 def orm_execute_statement(
1281 cls,
1282 session: Session,
1283 statement: dml.Insert,
1284 params: _CoreAnyExecuteParams,
1285 execution_options: OrmExecuteOptionsParameter,
1286 bind_arguments: _BindArguments,
1287 conn: Connection,
1288 ) -> _result.Result:
1289 insert_options = execution_options.get(
1290 "_sa_orm_insert_options", cls.default_insert_options
1291 )
1293 if insert_options._dml_strategy not in (
1294 "raw",
1295 "bulk",
1296 "orm",
1297 "auto",
1298 ):
1299 raise sa_exc.ArgumentError(
1300 "Valid strategies for ORM insert strategy "
1301 "are 'raw', 'orm', 'bulk', 'auto"
1302 )
1304 result: _result.Result[Unpack[TupleAny]]
1306 if insert_options._dml_strategy == "raw":
1307 result = conn.execute(
1308 statement, params or {}, execution_options=execution_options
1309 )
1310 return result
1312 if insert_options._dml_strategy == "bulk":
1313 mapper = insert_options._subject_mapper
1315 if (
1316 statement._post_values_clause is not None
1317 and mapper._multiple_persistence_tables
1318 ):
1319 raise sa_exc.InvalidRequestError(
1320 "bulk INSERT with a 'post values' clause "
1321 "(typically upsert) not supported for multi-table "
1322 f"mapper {mapper}"
1323 )
1325 assert mapper is not None
1326 assert session._transaction is not None
1327 result = _bulk_insert(
1328 mapper,
1329 cast(
1330 "Iterable[Dict[str, Any]]",
1331 [params] if isinstance(params, dict) else params,
1332 ),
1333 session._transaction,
1334 isstates=False,
1335 return_defaults=insert_options._return_defaults,
1336 render_nulls=insert_options._render_nulls,
1337 use_orm_insert_stmt=statement,
1338 execution_options=execution_options,
1339 )
1340 elif insert_options._dml_strategy == "orm":
1341 result = conn.execute(
1342 statement, params or {}, execution_options=execution_options
1343 )
1344 else:
1345 raise AssertionError()
1347 if not bool(statement._returning):
1348 return result
1350 if insert_options._populate_existing:
1351 load_options = execution_options.get(
1352 "_sa_orm_load_options", QueryContext.default_load_options
1353 )
1354 load_options += {"_populate_existing": True}
1355 execution_options = execution_options.union(
1356 {"_sa_orm_load_options": load_options}
1357 )
1359 return cls._return_orm_returning(
1360 session,
1361 statement,
1362 params,
1363 execution_options,
1364 bind_arguments,
1365 result,
1366 )
1368 @classmethod
1369 def create_for_statement(cls, statement, compiler, **kw) -> _BulkORMInsert:
1370 self = cast(
1371 _BulkORMInsert,
1372 super().create_for_statement(statement, compiler, **kw),
1373 )
1375 if compiler is not None:
1376 toplevel = not compiler.stack
1377 else:
1378 toplevel = True
1379 if not toplevel:
1380 return self
1382 mapper = statement._propagate_attrs["plugin_subject"]
1383 dml_strategy = statement._annotations.get("dml_strategy", "raw")
1384 if dml_strategy == "bulk":
1385 self._setup_for_bulk_insert(compiler)
1386 elif dml_strategy == "orm":
1387 self._setup_for_orm_insert(compiler, mapper)
1389 return self
1391 @classmethod
1392 def _resolved_keys_as_col_keys(cls, mapper, resolved_value_dict):
1393 return {
1394 col.key if col is not None else k: v
1395 for col, k, v in (
1396 (mapper.c.get(k), k, v) for k, v in resolved_value_dict.items()
1397 )
1398 }
1400 def _setup_for_orm_insert(self, compiler, mapper):
1401 statement = orm_level_statement = cast(dml.Insert, self.statement)
1403 statement = self._setup_orm_returning(
1404 compiler,
1405 orm_level_statement,
1406 statement,
1407 dml_mapper=mapper,
1408 use_supplemental_cols=False,
1409 )
1410 self.statement = statement
1412 def _setup_for_bulk_insert(self, compiler):
1413 """establish an INSERT statement within the context of
1414 bulk insert.
1416 This method will be within the "conn.execute()" call that is invoked
1417 by persistence._emit_insert_statement().
1419 """
1420 statement = orm_level_statement = cast(dml.Insert, self.statement)
1421 an = statement._annotations
1423 emit_insert_table, emit_insert_mapper = (
1424 an["_emit_insert_table"],
1425 an["_emit_insert_mapper"],
1426 )
1428 statement = statement._clone()
1430 statement.table = emit_insert_table
1431 if self._dict_parameters:
1432 self._dict_parameters = {
1433 col: val
1434 for col, val in self._dict_parameters.items()
1435 if col.table is emit_insert_table
1436 }
1438 statement = self._setup_orm_returning(
1439 compiler,
1440 orm_level_statement,
1441 statement,
1442 dml_mapper=emit_insert_mapper,
1443 use_supplemental_cols=True,
1444 )
1446 if (
1447 self.from_statement_ctx is not None
1448 and self.from_statement_ctx.compile_options._is_star
1449 ):
1450 raise sa_exc.CompileError(
1451 "Can't use RETURNING * with bulk ORM INSERT. "
1452 "Please use a different INSERT form, such as INSERT..VALUES "
1453 "or INSERT with a Core Connection"
1454 )
1456 self.statement = statement
1459@CompileState.plugin_for("orm", "update")
1460class _BulkORMUpdate(_BulkUDCompileState, UpdateDMLState):
1461 @classmethod
1462 def create_for_statement(cls, statement, compiler, **kw):
1463 self = cls.__new__(cls)
1465 dml_strategy = statement._annotations.get(
1466 "dml_strategy", "unspecified"
1467 )
1469 toplevel = not compiler.stack
1471 if toplevel and dml_strategy == "bulk":
1472 self._setup_for_bulk_update(statement, compiler)
1473 elif (
1474 dml_strategy == "core_only"
1475 or dml_strategy == "unspecified"
1476 and "parententity" not in statement.table._annotations
1477 ):
1478 UpdateDMLState.__init__(self, statement, compiler, **kw)
1479 elif not toplevel or dml_strategy in ("orm", "unspecified"):
1480 self._setup_for_orm_update(statement, compiler)
1482 return self
1484 def _setup_for_orm_update(self, statement, compiler, **kw):
1485 orm_level_statement = statement
1487 toplevel = not compiler.stack
1489 ext_info = statement.table._annotations["parententity"]
1491 self.mapper = mapper = ext_info.mapper
1493 self._resolved_values = self._get_resolved_values(mapper, statement)
1495 self._init_global_attributes(
1496 statement,
1497 compiler,
1498 toplevel=toplevel,
1499 process_criteria_for_toplevel=toplevel,
1500 )
1502 if statement._values:
1503 self._resolved_values = dict(self._resolved_values)
1505 new_stmt = statement._clone()
1507 if new_stmt.table._annotations["parententity"] is mapper:
1508 new_stmt.table = mapper.local_table
1510 # note if the statement has _multi_values, these
1511 # are passed through to the new statement, which will then raise
1512 # InvalidRequestError because UPDATE doesn't support multi_values
1513 # right now.
1514 if statement._values:
1515 new_stmt._values = self._resolved_values
1517 new_crit = self._adjust_for_extra_criteria(
1518 self.global_attributes, mapper
1519 )
1520 if new_crit:
1521 new_stmt = new_stmt.where(*new_crit)
1523 # if we are against a lambda statement we might not be the
1524 # topmost object that received per-execute annotations
1526 # do this first as we need to determine if there is
1527 # UPDATE..FROM
1529 UpdateDMLState.__init__(self, new_stmt, compiler, **kw)
1531 use_supplemental_cols = False
1533 if not toplevel:
1534 synchronize_session = None
1535 else:
1536 synchronize_session = compiler._annotations.get(
1537 "synchronize_session", None
1538 )
1539 can_use_returning = compiler._annotations.get(
1540 "can_use_returning", None
1541 )
1542 if can_use_returning is not False:
1543 # even though pre_exec has determined basic
1544 # can_use_returning for the dialect, if we are to use
1545 # RETURNING we need to run can_use_returning() at this level
1546 # unconditionally because is_delete_using was not known
1547 # at the pre_exec level
1548 can_use_returning = (
1549 synchronize_session == "fetch"
1550 and self.can_use_returning(
1551 compiler.dialect, mapper, is_multitable=self.is_multitable
1552 )
1553 )
1555 if synchronize_session == "fetch" and can_use_returning:
1556 use_supplemental_cols = True
1558 # NOTE: we might want to RETURNING the actual columns to be
1559 # synchronized also. however this is complicated and difficult
1560 # to align against the behavior of "evaluate". Additionally,
1561 # in a large number (if not the majority) of cases, we have the
1562 # "evaluate" answer, usually a fixed value, in memory already and
1563 # there's no need to re-fetch the same value
1564 # over and over again. so perhaps if it could be RETURNING just
1565 # the elements that were based on a SQL expression and not
1566 # a constant. For now it doesn't quite seem worth it
1567 new_stmt = new_stmt.return_defaults(*new_stmt.table.primary_key)
1569 if toplevel:
1570 new_stmt = self._setup_orm_returning(
1571 compiler,
1572 orm_level_statement,
1573 new_stmt,
1574 dml_mapper=mapper,
1575 use_supplemental_cols=use_supplemental_cols,
1576 )
1578 self.statement = new_stmt
1580 def _setup_for_bulk_update(self, statement, compiler, **kw):
1581 """establish an UPDATE statement within the context of
1582 bulk insert.
1584 This method will be within the "conn.execute()" call that is invoked
1585 by persistence._emit_update_statement().
1587 """
1588 statement = cast(dml.Update, statement)
1589 an = statement._annotations
1591 emit_update_table, _ = (
1592 an["_emit_update_table"],
1593 an["_emit_update_mapper"],
1594 )
1596 statement = statement._clone()
1597 statement.table = emit_update_table
1599 UpdateDMLState.__init__(self, statement, compiler, **kw)
1601 if self._maintain_values_ordering:
1602 raise sa_exc.InvalidRequestError(
1603 "bulk ORM UPDATE does not support ordered_values() for "
1604 "custom UPDATE statements with bulk parameter sets. Use a "
1605 "non-bulk UPDATE statement or use values()."
1606 )
1608 if self._dict_parameters:
1609 self._dict_parameters = {
1610 col: val
1611 for col, val in self._dict_parameters.items()
1612 if col.table is emit_update_table
1613 }
1614 self.statement = statement
1616 @classmethod
1617 def orm_execute_statement(
1618 cls,
1619 session: Session,
1620 statement: dml.Update,
1621 params: _CoreAnyExecuteParams,
1622 execution_options: OrmExecuteOptionsParameter,
1623 bind_arguments: _BindArguments,
1624 conn: Connection,
1625 ) -> _result.Result:
1627 update_options = execution_options.get(
1628 "_sa_orm_update_options", cls.default_update_options
1629 )
1631 if update_options._populate_existing:
1632 load_options = execution_options.get(
1633 "_sa_orm_load_options", QueryContext.default_load_options
1634 )
1635 load_options += {"_populate_existing": True}
1636 execution_options = execution_options.union(
1637 {"_sa_orm_load_options": load_options}
1638 )
1640 if update_options._dml_strategy not in (
1641 "orm",
1642 "auto",
1643 "bulk",
1644 "core_only",
1645 ):
1646 raise sa_exc.ArgumentError(
1647 "Valid strategies for ORM UPDATE strategy "
1648 "are 'orm', 'auto', 'bulk', 'core_only'"
1649 )
1651 result: _result.Result[Unpack[TupleAny]]
1653 if update_options._dml_strategy == "bulk":
1654 enable_check_rowcount = not statement._where_criteria
1656 assert update_options._synchronize_session != "fetch"
1658 if (
1659 statement._where_criteria
1660 and update_options._synchronize_session == "evaluate"
1661 ):
1662 raise sa_exc.InvalidRequestError(
1663 "bulk synchronize of persistent objects not supported "
1664 "when using bulk update with additional WHERE "
1665 "criteria right now. add synchronize_session=None "
1666 "execution option to bypass synchronize of persistent "
1667 "objects."
1668 )
1669 mapper = update_options._subject_mapper
1670 assert mapper is not None
1671 assert session._transaction is not None
1672 result = _bulk_update(
1673 mapper,
1674 cast(
1675 "Iterable[Dict[str, Any]]",
1676 [params] if isinstance(params, dict) else params,
1677 ),
1678 session._transaction,
1679 isstates=False,
1680 update_changed_only=False,
1681 use_orm_update_stmt=statement,
1682 enable_check_rowcount=enable_check_rowcount,
1683 )
1684 return cls.orm_setup_cursor_result(
1685 session,
1686 statement,
1687 params,
1688 execution_options,
1689 bind_arguments,
1690 result,
1691 )
1692 else:
1693 return super().orm_execute_statement(
1694 session,
1695 statement,
1696 params,
1697 execution_options,
1698 bind_arguments,
1699 conn,
1700 )
1702 @classmethod
1703 def can_use_returning(
1704 cls,
1705 dialect: Dialect,
1706 mapper: Mapper[Any],
1707 *,
1708 is_multitable: bool = False,
1709 is_update_from: bool = False,
1710 is_delete_using: bool = False,
1711 is_executemany: bool = False,
1712 ) -> bool:
1713 # normal answer for "should we use RETURNING" at all.
1714 normal_answer = (
1715 dialect.update_returning and mapper.local_table.implicit_returning
1716 )
1717 if not normal_answer:
1718 return False
1720 if is_executemany:
1721 return dialect.update_executemany_returning
1723 # these workarounds are currently hypothetical for UPDATE,
1724 # unlike DELETE where they impact MariaDB
1725 if is_update_from:
1726 return dialect.update_returning_multifrom
1728 elif is_multitable and not dialect.update_returning_multifrom:
1729 raise sa_exc.CompileError(
1730 f'Dialect "{dialect.name}" does not support RETURNING '
1731 "with UPDATE..FROM; for synchronize_session='fetch', "
1732 "please add the additional execution option "
1733 "'is_update_from=True' to the statement to indicate that "
1734 "a separate SELECT should be used for this backend."
1735 )
1737 return True
1739 @classmethod
1740 def _do_post_synchronize_bulk_evaluate(
1741 cls, session, params, result, update_options
1742 ):
1743 if not params:
1744 return
1746 mapper = update_options._subject_mapper
1747 pk_keys = [prop.key for prop in mapper._identity_key_props]
1749 identity_map = session.identity_map
1751 for param in params:
1752 identity_key = mapper.identity_key_from_primary_key(
1753 (param[key] for key in pk_keys),
1754 update_options._identity_token,
1755 )
1756 state = identity_map.fast_get_state(identity_key)
1757 if not state:
1758 continue
1760 evaluated_keys = set(param).difference(pk_keys)
1762 dict_ = state.dict
1763 # only evaluate unmodified attributes
1764 to_evaluate = state.unmodified.intersection(evaluated_keys)
1765 for key in to_evaluate:
1766 if key in dict_:
1767 dict_[key] = param[key]
1769 state.manager.dispatch.refresh(state, None, to_evaluate)
1771 state._commit(dict_, list(to_evaluate))
1773 # attributes that were formerly modified instead get expired.
1774 # this only gets hit if the session had pending changes
1775 # and autoflush were set to False.
1776 to_expire = evaluated_keys.intersection(dict_).difference(
1777 to_evaluate
1778 )
1779 if to_expire:
1780 state._expire_attributes(dict_, to_expire)
1782 @classmethod
1783 def _do_post_synchronize_evaluate(
1784 cls, session, statement, result, update_options
1785 ):
1786 matched_objects = cls._get_matched_objects_on_criteria(
1787 update_options,
1788 session.identity_map.all_states(),
1789 )
1791 cls._apply_update_set_values_to_objects(
1792 session,
1793 update_options,
1794 statement,
1795 result.context.compiled_parameters[0],
1796 [(obj, state, dict_) for obj, state, dict_, _ in matched_objects],
1797 result.prefetch_cols(),
1798 result.postfetch_cols(),
1799 )
1801 @classmethod
1802 def _do_post_synchronize_fetch(
1803 cls, session, statement, result, update_options
1804 ):
1805 target_mapper = update_options._subject_mapper
1807 returned_defaults_rows = result.returned_defaults_rows
1808 if returned_defaults_rows:
1809 pk_rows = cls._interpret_returning_rows(
1810 result, target_mapper, returned_defaults_rows
1811 )
1812 matched_rows = [
1813 tuple(row) + (update_options._identity_token,)
1814 for row in pk_rows
1815 ]
1816 else:
1817 matched_rows = update_options._matched_rows
1819 objs = [
1820 session.identity_map[identity_key]
1821 for identity_key in [
1822 target_mapper.identity_key_from_primary_key(
1823 list(primary_key),
1824 identity_token=identity_token,
1825 )
1826 for primary_key, identity_token in [
1827 (row[0:-1], row[-1]) for row in matched_rows
1828 ]
1829 if update_options._identity_token is None
1830 or identity_token == update_options._identity_token
1831 ]
1832 if identity_key in session.identity_map
1833 ]
1835 if not objs:
1836 return
1838 cls._apply_update_set_values_to_objects(
1839 session,
1840 update_options,
1841 statement,
1842 result.context.compiled_parameters[0],
1843 [
1844 (
1845 obj,
1846 attributes.instance_state(obj),
1847 attributes.instance_dict(obj),
1848 )
1849 for obj in objs
1850 ],
1851 result.prefetch_cols(),
1852 result.postfetch_cols(),
1853 )
1855 @classmethod
1856 def _apply_update_set_values_to_objects(
1857 cls,
1858 session,
1859 update_options,
1860 statement,
1861 effective_params,
1862 matched_objects,
1863 prefetch_cols,
1864 postfetch_cols,
1865 ):
1866 """apply values to objects derived from an update statement, e.g.
1867 UPDATE..SET <values>
1869 """
1871 mapper = update_options._subject_mapper
1872 target_cls = mapper.class_
1873 evaluator_compiler = evaluator._EvaluatorCompiler(target_cls)
1874 resolved_values = cls._get_resolved_values(mapper, statement)
1875 resolved_keys_as_propnames = cls._resolved_keys_as_propnames(
1876 mapper, resolved_values
1877 )
1878 value_evaluators = {}
1879 for key, value in resolved_keys_as_propnames:
1880 try:
1881 _evaluator = evaluator_compiler.process(
1882 coercions.expect(roles.ExpressionElementRole, value)
1883 )
1884 except evaluator.UnevaluatableError:
1885 pass
1886 else:
1887 value_evaluators[key] = _evaluator
1889 evaluated_keys = list(value_evaluators.keys())
1890 attrib = {k for k, v in resolved_keys_as_propnames}
1892 states = set()
1894 to_prefetch = {
1895 c
1896 for c in prefetch_cols
1897 if c.key in effective_params
1898 and c in mapper._columntoproperty
1899 and c.key not in evaluated_keys
1900 }
1901 to_expire = {
1902 mapper._columntoproperty[c].key
1903 for c in postfetch_cols
1904 if c in mapper._columntoproperty
1905 }.difference(evaluated_keys)
1907 prefetch_transfer = [
1908 (mapper._columntoproperty[c].key, c.key) for c in to_prefetch
1909 ]
1911 for obj, state, dict_ in matched_objects:
1913 dict_.update(
1914 {
1915 col_to_prop: effective_params[c_key]
1916 for col_to_prop, c_key in prefetch_transfer
1917 }
1918 )
1920 state._expire_attributes(state.dict, to_expire)
1922 to_evaluate = state.unmodified.intersection(evaluated_keys)
1924 for key in to_evaluate:
1925 if key in dict_:
1926 # only run eval for attributes that are present.
1927 dict_[key] = value_evaluators[key](obj)
1929 state.manager.dispatch.refresh(state, None, to_evaluate)
1931 state._commit(dict_, list(to_evaluate))
1933 # attributes that were formerly modified instead get expired.
1934 # this only gets hit if the session had pending changes
1935 # and autoflush were set to False.
1936 to_expire = attrib.intersection(dict_).difference(to_evaluate)
1937 if to_expire:
1938 state._expire_attributes(dict_, to_expire)
1940 states.add(state)
1941 session._register_altered(states)
1944@CompileState.plugin_for("orm", "delete")
1945class _BulkORMDelete(_BulkUDCompileState, DeleteDMLState):
1946 @classmethod
1947 def create_for_statement(cls, statement, compiler, **kw):
1948 self = cls.__new__(cls)
1950 dml_strategy = statement._annotations.get(
1951 "dml_strategy", "unspecified"
1952 )
1954 if (
1955 dml_strategy == "core_only"
1956 or dml_strategy == "unspecified"
1957 and "parententity" not in statement.table._annotations
1958 ):
1959 DeleteDMLState.__init__(self, statement, compiler, **kw)
1960 return self
1962 toplevel = not compiler.stack
1964 orm_level_statement = statement
1966 ext_info = statement.table._annotations["parententity"]
1967 self.mapper = mapper = ext_info.mapper
1969 self._init_global_attributes(
1970 statement,
1971 compiler,
1972 toplevel=toplevel,
1973 process_criteria_for_toplevel=toplevel,
1974 )
1976 new_stmt = statement._clone()
1978 if new_stmt.table._annotations["parententity"] is mapper:
1979 new_stmt.table = mapper.local_table
1981 new_crit = cls._adjust_for_extra_criteria(
1982 self.global_attributes, mapper
1983 )
1984 if new_crit:
1985 new_stmt = new_stmt.where(*new_crit)
1987 # do this first as we need to determine if there is
1988 # DELETE..FROM
1989 DeleteDMLState.__init__(self, new_stmt, compiler, **kw)
1991 use_supplemental_cols = False
1993 if not toplevel:
1994 synchronize_session = None
1995 else:
1996 synchronize_session = compiler._annotations.get(
1997 "synchronize_session", None
1998 )
1999 can_use_returning = compiler._annotations.get(
2000 "can_use_returning", None
2001 )
2002 if can_use_returning is not False:
2003 # even though pre_exec has determined basic
2004 # can_use_returning for the dialect, if we are to use
2005 # RETURNING we need to run can_use_returning() at this level
2006 # unconditionally because is_delete_using was not known
2007 # at the pre_exec level
2008 can_use_returning = (
2009 synchronize_session == "fetch"
2010 and self.can_use_returning(
2011 compiler.dialect,
2012 mapper,
2013 is_multitable=self.is_multitable,
2014 is_delete_using=compiler._annotations.get(
2015 "is_delete_using", False
2016 ),
2017 )
2018 )
2020 if can_use_returning:
2021 use_supplemental_cols = True
2023 new_stmt = new_stmt.return_defaults(*new_stmt.table.primary_key)
2025 if toplevel:
2026 new_stmt = self._setup_orm_returning(
2027 compiler,
2028 orm_level_statement,
2029 new_stmt,
2030 dml_mapper=mapper,
2031 use_supplemental_cols=use_supplemental_cols,
2032 )
2034 self.statement = new_stmt
2036 return self
2038 @classmethod
2039 def orm_execute_statement(
2040 cls,
2041 session: Session,
2042 statement: dml.Delete,
2043 params: _CoreAnyExecuteParams,
2044 execution_options: OrmExecuteOptionsParameter,
2045 bind_arguments: _BindArguments,
2046 conn: Connection,
2047 ) -> _result.Result:
2048 update_options = execution_options.get(
2049 "_sa_orm_update_options", cls.default_update_options
2050 )
2052 if update_options._dml_strategy == "bulk":
2053 raise sa_exc.InvalidRequestError(
2054 "Bulk ORM DELETE not supported right now. "
2055 "Statement may be invoked at the "
2056 "Core level using "
2057 "session.connection().execute(stmt, parameters)"
2058 )
2060 if update_options._dml_strategy not in ("orm", "auto", "core_only"):
2061 raise sa_exc.ArgumentError(
2062 "Valid strategies for ORM DELETE strategy are 'orm', 'auto', "
2063 "'core_only'"
2064 )
2066 return super().orm_execute_statement(
2067 session, statement, params, execution_options, bind_arguments, conn
2068 )
2070 @classmethod
2071 def can_use_returning(
2072 cls,
2073 dialect: Dialect,
2074 mapper: Mapper[Any],
2075 *,
2076 is_multitable: bool = False,
2077 is_update_from: bool = False,
2078 is_delete_using: bool = False,
2079 is_executemany: bool = False,
2080 ) -> bool:
2081 # normal answer for "should we use RETURNING" at all.
2082 normal_answer = (
2083 dialect.delete_returning and mapper.local_table.implicit_returning
2084 )
2085 if not normal_answer:
2086 return False
2088 # now get into special workarounds because MariaDB supports
2089 # DELETE...RETURNING but not DELETE...USING...RETURNING.
2090 if is_delete_using:
2091 # is_delete_using hint was passed. use
2092 # additional dialect feature (True for PG, False for MariaDB)
2093 return dialect.delete_returning_multifrom
2095 elif is_multitable and not dialect.delete_returning_multifrom:
2096 # is_delete_using hint was not passed, but we determined
2097 # at compile time that this is in fact a DELETE..USING.
2098 # it's too late to continue since we did not pre-SELECT.
2099 # raise that we need that hint up front.
2101 raise sa_exc.CompileError(
2102 f'Dialect "{dialect.name}" does not support RETURNING '
2103 "with DELETE..USING; for synchronize_session='fetch', "
2104 "please add the additional execution option "
2105 "'is_delete_using=True' to the statement to indicate that "
2106 "a separate SELECT should be used for this backend."
2107 )
2109 return True
2111 @classmethod
2112 def _do_post_synchronize_evaluate(
2113 cls, session, statement, result, update_options
2114 ):
2115 matched_objects = cls._get_matched_objects_on_criteria(
2116 update_options,
2117 session.identity_map.all_states(),
2118 )
2120 to_delete = []
2122 for _, state, dict_, is_partially_expired in matched_objects:
2123 if is_partially_expired:
2124 state._expire(dict_, session.identity_map._modified)
2125 else:
2126 to_delete.append(state)
2128 if to_delete:
2129 session._remove_newly_deleted(to_delete)
2131 @classmethod
2132 def _do_post_synchronize_fetch(
2133 cls, session, statement, result, update_options
2134 ):
2135 target_mapper = update_options._subject_mapper
2137 returned_defaults_rows = result.returned_defaults_rows
2139 if returned_defaults_rows:
2140 pk_rows = cls._interpret_returning_rows(
2141 result, target_mapper, returned_defaults_rows
2142 )
2144 matched_rows = [
2145 tuple(row) + (update_options._identity_token,)
2146 for row in pk_rows
2147 ]
2148 else:
2149 matched_rows = update_options._matched_rows
2151 for row in matched_rows:
2152 primary_key = row[0:-1]
2153 identity_token = row[-1]
2155 # TODO: inline this and call remove_newly_deleted
2156 # once
2157 identity_key = target_mapper.identity_key_from_primary_key(
2158 list(primary_key),
2159 identity_token=identity_token,
2160 )
2161 if identity_key in session.identity_map:
2162 session._remove_newly_deleted(
2163 [
2164 attributes.instance_state(
2165 session.identity_map[identity_key]
2166 )
2167 ]
2168 )