codekingpro/portable-devtools
115k
1# orm/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
8
9
10"""private module containing functions used to emit INSERT, UPDATE
11and DELETE statements on behalf of a :class:`_orm.Mapper` and its descending
12mappers.
13
14The functions here are called only by the unit of work functions
15in unitofwork.py.
16
17"""
18from __future__ import annotations
19
20from itertools import chain
21from itertools import groupby
22from itertools import zip_longest
23import operator
24
25from . import attributes
26from . import exc as orm_exc
27from . import loading
28from . import sync
29from .base import state_str
30from .. import exc as sa_exc
31from .. import future
32from .. import sql
33from .. import util
34from ..engine import cursor as _cursor
35from ..sql import operators
36from ..sql.elements import BooleanClauseList
37from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
38
39
40def save_obj(base_mapper, states, uowtransaction, single=False):
41 """Issue ``INSERT`` and/or ``UPDATE`` statements for a list
42 of objects.
43
44 This is called within the context of a UOWTransaction during a
45 flush operation, given a list of states to be flushed. The
46 base mapper in an inheritance hierarchy handles the inserts/
47 updates for all descendant mappers.
48
49 """
50
51 # if batch=false, call _save_obj separately for each object
52 if not single and not base_mapper.batch:
53 for state in _sort_states(base_mapper, states):
54 save_obj(base_mapper, [state], uowtransaction, single=True)
55 return
56
57 states_to_update = []
58 states_to_insert = []
59
60 for (
61 state,
62 dict_,
63 mapper,
64 connection,
65 has_identity,
66 row_switch,
67 update_version_id,
68 ) in _organize_states_for_save(base_mapper, states, uowtransaction):
69 if has_identity or row_switch:
70 states_to_update.append(
71 (state, dict_, mapper, connection, update_version_id)
72 )
73 else:
74 states_to_insert.append((state, dict_, mapper, connection))
75
76 for table, mapper in base_mapper._sorted_tables.items():
77 if table not in mapper._pks_by_table:
78 continue
79 insert = _collect_insert_commands(table, states_to_insert)
80
81 update = _collect_update_commands(
82 uowtransaction, table, states_to_update
83 )
84
85 _emit_update_statements(
86 base_mapper,
87 uowtransaction,
88 mapper,
89 table,
90 update,
91 )
92
93 _emit_insert_statements(
94 base_mapper,
95 uowtransaction,
96 mapper,
97 table,
98 insert,
99 )
100
101 _finalize_insert_update_commands(
102 base_mapper,
103 uowtransaction,
104 chain(
105 (
106 (state, state_dict, mapper, connection, False)
107 for (state, state_dict, mapper, connection) in states_to_insert
108 ),
109 (
110 (state, state_dict, mapper, connection, True)
111 for (
112 state,
113 state_dict,
114 mapper,
115 connection,
116 update_version_id,
117 ) in states_to_update
118 ),
119 ),
120 )
121
122
123def post_update(base_mapper, states, uowtransaction, post_update_cols):
124 """Issue UPDATE statements on behalf of a relationship() which
125 specifies post_update.
126
127 """
128
129 states_to_update = list(
130 _organize_states_for_post_update(base_mapper, states, uowtransaction)
131 )
132
133 for table, mapper in base_mapper._sorted_tables.items():
134 if table not in mapper._pks_by_table:
135 continue
136
137 update = (
138 (
139 state,
140 state_dict,
141 sub_mapper,
142 connection,
143 (
144 mapper._get_committed_state_attr_by_column(
145 state, state_dict, mapper.version_id_col
146 )
147 if mapper.version_id_col is not None
148 else None
149 ),
150 )
151 for state, state_dict, sub_mapper, connection in states_to_update
152 if table in sub_mapper._pks_by_table
153 )
154
155 update = _collect_post_update_commands(
156 base_mapper, uowtransaction, table, update, post_update_cols
157 )
158
159 _emit_post_update_statements(
160 base_mapper,
161 uowtransaction,
162 mapper,
163 table,
164 update,
165 )
166
167
168def delete_obj(base_mapper, states, uowtransaction):
169 """Issue ``DELETE`` statements for a list of objects.
170
171 This is called within the context of a UOWTransaction during a
172 flush operation.
173
174 """
175
176 states_to_delete = list(
177 _organize_states_for_delete(base_mapper, states, uowtransaction)
178 )
179
180 table_to_mapper = base_mapper._sorted_tables
181
182 for table in reversed(list(table_to_mapper.keys())):
183 mapper = table_to_mapper[table]
184 if table not in mapper._pks_by_table:
185 continue
186 elif mapper.inherits and mapper.passive_deletes:
187 continue
188
189 delete = _collect_delete_commands(
190 base_mapper, uowtransaction, table, states_to_delete
191 )
192
193 _emit_delete_statements(
194 base_mapper,
195 uowtransaction,
196 mapper,
197 table,
198 delete,
199 )
200
201 for (
202 state,
203 state_dict,
204 mapper,
205 connection,
206 update_version_id,
207 ) in states_to_delete:
208 mapper.dispatch.after_delete(mapper, connection, state)
209
210
211def _organize_states_for_save(base_mapper, states, uowtransaction):
212 """Make an initial pass across a set of states for INSERT or
213 UPDATE.
214
215 This includes splitting out into distinct lists for
216 each, calling before_insert/before_update, obtaining
217 key information for each state including its dictionary,
218 mapper, the connection to use for the execution per state,
219 and the identity flag.
220
221 """
222
223 for state, dict_, mapper, connection in _connections_for_states(
224 base_mapper, uowtransaction, states
225 ):
226 has_identity = bool(state.key)
227
228 instance_key = state.key or mapper._identity_key_from_state(state)
229
230 row_switch = update_version_id = None
231
232 # call before_XXX extensions
233 if not has_identity:
234 mapper.dispatch.before_insert(mapper, connection, state)
235 else:
236 mapper.dispatch.before_update(mapper, connection, state)
237
238 if mapper._validate_polymorphic_identity:
239 mapper._validate_polymorphic_identity(mapper, state, dict_)
240
241 # detect if we have a "pending" instance (i.e. has
242 # no instance_key attached to it), and another instance
243 # with the same identity key already exists as persistent.
244 # convert to an UPDATE if so.
245 if (
246 not has_identity
247 and instance_key in uowtransaction.session.identity_map
248 ):
249 instance = uowtransaction.session.identity_map[instance_key]
250 existing = attributes.instance_state(instance)
251
252 if not uowtransaction.was_already_deleted(existing):
253 if not uowtransaction.is_deleted(existing):
254 util.warn(
255 "New instance %s with identity key %s conflicts "
256 "with persistent instance %s"
257 % (state_str(state), instance_key, state_str(existing))
258 )
259 else:
260 base_mapper._log_debug(
261 "detected row switch for identity %s. "
262 "will update %s, remove %s from "
263 "transaction",
264 instance_key,
265 state_str(state),
266 state_str(existing),
267 )
268
269 # remove the "delete" flag from the existing element
270 uowtransaction.remove_state_actions(existing)
271 row_switch = existing
272
273 if (has_identity or row_switch) and mapper.version_id_col is not None:
274 update_version_id = mapper._get_committed_state_attr_by_column(
275 row_switch if row_switch else state,
276 row_switch.dict if row_switch else dict_,
277 mapper.version_id_col,
278 )
279
280 yield (
281 state,
282 dict_,
283 mapper,
284 connection,
285 has_identity,
286 row_switch,
287 update_version_id,
288 )
289
290
291def _organize_states_for_post_update(base_mapper, states, uowtransaction):
292 """Make an initial pass across a set of states for UPDATE
293 corresponding to post_update.
294
295 This includes obtaining key information for each state
296 including its dictionary, mapper, the connection to use for
297 the execution per state.
298
299 """
300 return _connections_for_states(base_mapper, uowtransaction, states)
301
302
303def _organize_states_for_delete(base_mapper, states, uowtransaction):
304 """Make an initial pass across a set of states for DELETE.
305
306 This includes calling out before_delete and obtaining
307 key information for each state including its dictionary,
308 mapper, the connection to use for the execution per state.
309
310 """
311 for state, dict_, mapper, connection in _connections_for_states(
312 base_mapper, uowtransaction, states
313 ):
314 mapper.dispatch.before_delete(mapper, connection, state)
315
316 if mapper.version_id_col is not None:
317 update_version_id = mapper._get_committed_state_attr_by_column(
318 state, dict_, mapper.version_id_col
319 )
320 else:
321 update_version_id = None
322
323 yield (state, dict_, mapper, connection, update_version_id)
324
325
326def _collect_insert_commands(
327 table,
328 states_to_insert,
329 *,
330 bulk=False,
331 return_defaults=False,
332 render_nulls=False,
333 include_bulk_keys=(),
334):
335 """Identify sets of values to use in INSERT statements for a
336 list of states.
337
338 """
339 for state, state_dict, mapper, connection in states_to_insert:
340 if table not in mapper._pks_by_table:
341 continue
342
343 params = {}
344 value_params = {}
345
346 propkey_to_col = mapper._propkey_to_col[table]
347
348 eval_none = mapper._insert_cols_evaluating_none[table]
349
350 for propkey in set(propkey_to_col).intersection(state_dict):
351 value = state_dict[propkey]
352 col = propkey_to_col[propkey]
353 if value is None and col not in eval_none and not render_nulls:
354 continue
355 elif not bulk and (
356 hasattr(value, "__clause_element__")
357 or isinstance(value, sql.ClauseElement)
358 ):
359 value_params[col] = (
360 value.__clause_element__()
361 if hasattr(value, "__clause_element__")
362 else value
363 )
364 else:
365 params[col.key] = value
366
367 if not bulk:
368 # for all the columns that have no default and we don't have
369 # a value and where "None" is not a special value, add
370 # explicit None to the INSERT. This is a legacy behavior
371 # which might be worth removing, as it should not be necessary
372 # and also produces confusion, given that "missing" and None
373 # now have distinct meanings
374 for colkey in (
375 mapper._insert_cols_as_none[table]
376 .difference(params)
377 .difference([c.key for c in value_params])
378 ):
379 params[colkey] = None
380
381 if not bulk or return_defaults:
382 # params are in terms of Column key objects, so
383 # compare to pk_keys_by_table
384 has_all_pks = mapper._pk_keys_by_table[table].issubset(params)
385
386 if mapper.base_mapper._prefer_eager_defaults(
387 connection.dialect, table
388 ):
389 has_all_defaults = mapper._server_default_col_keys[
390 table
391 ].issubset(params)
392 else:
393 has_all_defaults = True
394 else:
395 has_all_defaults = has_all_pks = True
396
397 if (
398 mapper.version_id_generator is not False
399 and mapper.version_id_col is not None
400 and mapper.version_id_col in mapper._cols_by_table[table]
401 ):
402 params[mapper.version_id_col.key] = mapper.version_id_generator(
403 None
404 )
405
406 if bulk:
407 if mapper._set_polymorphic_identity:
408 params.setdefault(
409 mapper._polymorphic_attr_key, mapper.polymorphic_identity
410 )
411
412 if include_bulk_keys:
413 params.update((k, state_dict[k]) for k in include_bulk_keys)
414
415 yield (
416 state,
417 state_dict,
418 params,
419 mapper,
420 connection,
421 value_params,
422 has_all_pks,
423 has_all_defaults,
424 )
425
426
427def _collect_update_commands(
428 uowtransaction,
429 table,
430 states_to_update,
431 *,
432 bulk=False,
433 use_orm_update_stmt=None,
434 include_bulk_keys=(),
435):
436 """Identify sets of values to use in UPDATE statements for a
437 list of states.
438
439 This function works intricately with the history system
440 to determine exactly what values should be updated
441 as well as how the row should be matched within an UPDATE
442 statement. Includes some tricky scenarios where the primary
443 key of an object might have been changed.
444
445 """
446
447 for (
448 state,
449 state_dict,
450 mapper,
451 connection,
452 update_version_id,
453 ) in states_to_update:
454 if table not in mapper._pks_by_table:
455 continue
456
457 pks = mapper._pks_by_table[table]
458
459 if use_orm_update_stmt is not None:
460 # TODO: ordered values, etc
461 value_params = use_orm_update_stmt._values
462 else:
463 value_params = {}
464
465 propkey_to_col = mapper._propkey_to_col[table]
466
467 if bulk:
468 # keys here are mapped attribute keys, so
469 # look at mapper attribute keys for pk
470 params = {
471 propkey_to_col[propkey].key: state_dict[propkey]
472 for propkey in set(propkey_to_col)
473 .intersection(state_dict)
474 .difference(mapper._pk_attr_keys_by_table[table])
475 }
476 has_all_defaults = True
477 else:
478 params = {}
479 for propkey in set(propkey_to_col).intersection(
480 state.committed_state
481 ):
482 value = state_dict[propkey]
483 col = propkey_to_col[propkey]
484
485 if hasattr(value, "__clause_element__") or isinstance(
486 value, sql.ClauseElement
487 ):
488 value_params[col] = (
489 value.__clause_element__()
490 if hasattr(value, "__clause_element__")
491 else value
492 )
493 # guard against values that generate non-__nonzero__
494 # objects for __eq__()
495 elif (
496 state.manager[propkey].impl.is_equal(
497 value, state.committed_state[propkey]
498 )
499 is not True
500 ):
501 params[col.key] = value
502
503 if mapper.base_mapper.eager_defaults is True:
504 has_all_defaults = (
505 mapper._server_onupdate_default_col_keys[table]
506 ).issubset(params)
507 else:
508 has_all_defaults = True
509
510 if (
511 update_version_id is not None
512 and mapper.version_id_col in mapper._cols_by_table[table]
513 ):
514 if not bulk and not (params or value_params):
515 # HACK: check for history in other tables, in case the
516 # history is only in a different table than the one
517 # where the version_id_col is. This logic was lost
518 # from 0.9 -> 1.0.0 and restored in 1.0.6.
519 for prop in mapper._columntoproperty.values():
520 history = state.manager[prop.key].impl.get_history(
521 state, state_dict, attributes.PASSIVE_NO_INITIALIZE
522 )
523 if history.added:
524 break
525 else:
526 # no net change, break
527 continue
528
529 col = mapper.version_id_col
530 no_params = not params and not value_params
531 params[col._label] = update_version_id
532
533 if (
534 bulk or col.key not in params
535 ) and mapper.version_id_generator is not False:
536 val = mapper.version_id_generator(update_version_id)
537 params[col.key] = val
538 elif mapper.version_id_generator is False and no_params:
539 # no version id generator, no values set on the table,
540 # and version id wasn't manually incremented.
541 # set version id to itself so we get an UPDATE
542 # statement
543 params[col.key] = update_version_id
544
545 elif not (params or value_params):
546 continue
547
548 has_all_pks = True
549 expect_pk_cascaded = False
550 if bulk:
551 # keys here are mapped attribute keys, so
552 # look at mapper attribute keys for pk
553 pk_params = {
554 propkey_to_col[propkey]._label: state_dict.get(propkey)
555 for propkey in set(propkey_to_col).intersection(
556 mapper._pk_attr_keys_by_table[table]
557 )
558 }
559 if util.NONE_SET.intersection(pk_params.values()):
560 raise sa_exc.InvalidRequestError(
561 f"No primary key value supplied for column(s) "
562 f"""{
563 ', '.join(
564 str(c) for c in pks if pk_params[c._label] is None
565 )
566 }; """
567 "per-row ORM Bulk UPDATE by Primary Key requires that "
568 "records contain primary key values",
569 code="bupq",
570 )
571
572 else:
573 pk_params = {}
574 for col in pks:
575 propkey = mapper._columntoproperty[col].key
576
577 history = state.manager[propkey].impl.get_history(
578 state, state_dict, attributes.PASSIVE_OFF
579 )
580
581 if history.added:
582 if (
583 not history.deleted
584 or ("pk_cascaded", state, col)
585 in uowtransaction.attributes
586 ):
587 expect_pk_cascaded = True
588 pk_params[col._label] = history.added[0]
589 params.pop(col.key, None)
590 else:
591 # else, use the old value to locate the row
592 pk_params[col._label] = history.deleted[0]
593 if col in value_params:
594 has_all_pks = False
595 else:
596 pk_params[col._label] = history.unchanged[0]
597 if pk_params[col._label] is None:
598 raise orm_exc.FlushError(
599 "Can't update table %s using NULL for primary "
600 "key value on column %s" % (table, col)
601 )
602
603 if include_bulk_keys:
604 params.update((k, state_dict[k]) for k in include_bulk_keys)
605
606 if params or value_params:
607 params.update(pk_params)
608 yield (
609 state,
610 state_dict,
611 params,
612 mapper,
613 connection,
614 value_params,
615 has_all_defaults,
616 has_all_pks,
617 )
618 elif expect_pk_cascaded:
619 # no UPDATE occurs on this table, but we expect that CASCADE rules
620 # have changed the primary key of the row; propagate this event to
621 # other columns that expect to have been modified. this normally
622 # occurs after the UPDATE is emitted however we invoke it here
623 # explicitly in the absence of our invoking an UPDATE
624 for m, equated_pairs in mapper._table_to_equated[table]:
625 sync.populate(
626 state,
627 m,
628 state,
629 m,
630 equated_pairs,
631 uowtransaction,
632 mapper.passive_updates,
633 )
634
635
636def _collect_post_update_commands(
637 base_mapper, uowtransaction, table, states_to_update, post_update_cols
638):
639 """Identify sets of values to use in UPDATE statements for a
640 list of states within a post_update operation.
641
642 """
643
644 for (
645 state,
646 state_dict,
647 mapper,
648 connection,
649 update_version_id,
650 ) in states_to_update:
651 # assert table in mapper._pks_by_table
652
653 pks = mapper._pks_by_table[table]
654 params = {}
655 hasdata = False
656
657 for col in mapper._cols_by_table[table]:
658 if col in pks:
659 params[col._label] = mapper._get_state_attr_by_column(
660 state, state_dict, col, passive=attributes.PASSIVE_OFF
661 )
662
663 elif col in post_update_cols or col.onupdate is not None:
664 prop = mapper._columntoproperty[col]
665 history = state.manager[prop.key].impl.get_history(
666 state, state_dict, attributes.PASSIVE_NO_INITIALIZE
667 )
668 if history.added:
669 value = history.added[0]
670 params[col.key] = value
671 hasdata = True
672 if hasdata:
673 if (
674 update_version_id is not None
675 and mapper.version_id_col in mapper._cols_by_table[table]
676 ):
677 col = mapper.version_id_col
678 params[col._label] = update_version_id
679
680 if (
681 bool(state.key)
682 and col.key not in params
683 and mapper.version_id_generator is not False
684 ):
685 val = mapper.version_id_generator(update_version_id)
686 params[col.key] = val
687 yield state, state_dict, mapper, connection, params
688
689
690def _collect_delete_commands(
691 base_mapper, uowtransaction, table, states_to_delete
692):
693 """Identify values to use in DELETE statements for a list of
694 states to be deleted."""
695
696 for (
697 state,
698 state_dict,
699 mapper,
700 connection,
701 update_version_id,
702 ) in states_to_delete:
703 if table not in mapper._pks_by_table:
704 continue
705
706 params = {}
707 for col in mapper._pks_by_table[table]:
708 params[col.key] = value = (
709 mapper._get_committed_state_attr_by_column(
710 state, state_dict, col
711 )
712 )
713 if value is None:
714 raise orm_exc.FlushError(
715 "Can't delete from table %s "
716 "using NULL for primary "
717 "key value on column %s" % (table, col)
718 )
719
720 if (
721 update_version_id is not None
722 and mapper.version_id_col in mapper._cols_by_table[table]
723 ):
724 params[mapper.version_id_col.key] = update_version_id
725 yield params, connection
726
727
728def _emit_update_statements(
729 base_mapper,
730 uowtransaction,
731 mapper,
732 table,
733 update,
734 *,
735 bookkeeping=True,
736 use_orm_update_stmt=None,
737 enable_check_rowcount=True,
738):
739 """Emit UPDATE statements corresponding to value lists collected
740 by _collect_update_commands()."""
741
742 needs_version_id = (
743 mapper.version_id_col is not None
744 and mapper.version_id_col in mapper._cols_by_table[table]
745 )
746
747 execution_options = {"compiled_cache": base_mapper._compiled_cache}
748
749 def update_stmt(existing_stmt=None):
750 clauses = BooleanClauseList._construct_raw(operators.and_)
751
752 for col in mapper._pks_by_table[table]:
753 clauses._append_inplace(
754 col == sql.bindparam(col._label, type_=col.type)
755 )
756
757 if needs_version_id:
758 clauses._append_inplace(
759 mapper.version_id_col
760 == sql.bindparam(
761 mapper.version_id_col._label,
762 type_=mapper.version_id_col.type,
763 )
764 )
765
766 if existing_stmt is not None:
767 stmt = existing_stmt.where(clauses)
768 else:
769 stmt = table.update().where(clauses)
770 return stmt
771
772 if use_orm_update_stmt is not None:
773 cached_stmt = update_stmt(use_orm_update_stmt)
774
775 else:
776 cached_stmt = base_mapper._memo(("update", table), update_stmt)
777
778 for (
779 (connection, paramkeys, hasvalue, has_all_defaults, has_all_pks),
780 records,
781 ) in groupby(
782 update,
783 lambda rec: (
784 rec[4], # connection
785 set(rec[2]), # set of parameter keys
786 bool(rec[5]), # whether or not we have "value" parameters
787 rec[6], # has_all_defaults
788 rec[7], # has all pks
789 ),
790 ):
791 rows = 0
792 records = list(records)
793
794 statement = cached_stmt
795
796 if use_orm_update_stmt is not None:
797 statement = statement._annotate(
798 {
799 "_emit_update_table": table,
800 "_emit_update_mapper": mapper,
801 }
802 )
803
804 return_defaults = False
805
806 if not has_all_pks:
807 statement = statement.return_defaults(*mapper._pks_by_table[table])
808 return_defaults = True
809
810 if (
811 bookkeeping
812 and not has_all_defaults
813 and mapper.base_mapper.eager_defaults is True
814 # change as of #8889 - if RETURNING is not going to be used anyway,
815 # (applies to MySQL, MariaDB which lack UPDATE RETURNING) ensure
816 # we can do an executemany UPDATE which is more efficient
817 and table.implicit_returning
818 and connection.dialect.update_returning
819 ):
820 statement = statement.return_defaults(
821 *mapper._server_onupdate_default_cols[table]
822 )
823 return_defaults = True
824
825 if mapper._version_id_has_server_side_value:
826 statement = statement.return_defaults(mapper.version_id_col)
827 return_defaults = True
828
829 assert_singlerow = connection.dialect.supports_sane_rowcount
830
831 assert_multirow = (
832 assert_singlerow
833 and connection.dialect.supports_sane_multi_rowcount
834 )
835
836 # change as of #8889 - if RETURNING is not going to be used anyway,
837 # (applies to MySQL, MariaDB which lack UPDATE RETURNING) ensure
838 # we can do an executemany UPDATE which is more efficient
839 allow_executemany = not return_defaults and not needs_version_id
840
841 if hasvalue:
842 for (
843 state,
844 state_dict,
845 params,
846 mapper,
847 connection,
848 value_params,
849 has_all_defaults,
850 has_all_pks,
851 ) in records:
852 c = connection.execute(
853 statement.values(value_params),
854 params,
855 execution_options=execution_options,
856 )
857 if bookkeeping:
858 _postfetch(
859 mapper,
860 uowtransaction,
861 table,
862 state,
863 state_dict,
864 c,
865 c.context.compiled_parameters[0],
866 value_params,
867 True,
868 c.returned_defaults,
869 )
870 rows += c.rowcount
871 check_rowcount = enable_check_rowcount and assert_singlerow
872 else:
873 if not allow_executemany:
874 check_rowcount = enable_check_rowcount and assert_singlerow
875 for (
876 state,
877 state_dict,
878 params,
879 mapper,
880 connection,
881 value_params,
882 has_all_defaults,
883 has_all_pks,
884 ) in records:
885 c = connection.execute(
886 statement, params, execution_options=execution_options
887 )
888
889 # TODO: why with bookkeeping=False?
890 if bookkeeping:
891 _postfetch(
892 mapper,
893 uowtransaction,
894 table,
895 state,
896 state_dict,
897 c,
898 c.context.compiled_parameters[0],
899 value_params,
900 True,
901 c.returned_defaults,
902 )
903 rows += c.rowcount
904 else:
905 multiparams = [rec[2] for rec in records]
906
907 check_rowcount = enable_check_rowcount and (
908 assert_multirow
909 or (assert_singlerow and len(multiparams) == 1)
910 )
911
912 c = connection.execute(
913 statement, multiparams, execution_options=execution_options
914 )
915
916 rows += c.rowcount
917
918 for (
919 state,
920 state_dict,
921 params,
922 mapper,
923 connection,
924 value_params,
925 has_all_defaults,
926 has_all_pks,
927 ) in records:
928 if bookkeeping:
929 _postfetch(
930 mapper,
931 uowtransaction,
932 table,
933 state,
934 state_dict,
935 c,
936 c.context.compiled_parameters[0],
937 value_params,
938 True,
939 (
940 c.returned_defaults
941 if not c.context.executemany
942 else None
943 ),
944 )
945
946 if check_rowcount:
947 if rows != len(records):
948 raise orm_exc.StaleDataError(
949 "UPDATE statement on table '%s' expected to "
950 "update %d row(s); %d were matched."
951 % (table.description, len(records), rows)
952 )
953
954 elif needs_version_id:
955 util.warn(
956 "Dialect %s does not support updated rowcount "
957 "- versioning cannot be verified."
958 % c.dialect.dialect_description
959 )
960
961
962def _emit_insert_statements(
963 base_mapper,
964 uowtransaction,
965 mapper,
966 table,
967 insert,
968 *,
969 bookkeeping=True,
970 use_orm_insert_stmt=None,
971 execution_options=None,
972):
973 """Emit INSERT statements corresponding to value lists collected
974 by _collect_insert_commands()."""
975
976 if use_orm_insert_stmt is not None:
977 cached_stmt = use_orm_insert_stmt
978 exec_opt = util.EMPTY_DICT
979
980 # if a user query with RETURNING was passed, we definitely need
981 # to use RETURNING.
982 returning_is_required_anyway = bool(use_orm_insert_stmt._returning)
983 deterministic_results_reqd = (
984 returning_is_required_anyway
985 and use_orm_insert_stmt._sort_by_parameter_order
986 ) or bookkeeping
987 else:
988 returning_is_required_anyway = False
989 deterministic_results_reqd = bookkeeping
990 cached_stmt = base_mapper._memo(("insert", table), table.insert)
991 exec_opt = {"compiled_cache": base_mapper._compiled_cache}
992
993 if execution_options:
994 execution_options = util.EMPTY_DICT.merge_with(
995 exec_opt, execution_options
996 )
997 else:
998 execution_options = exec_opt
999
1000 return_result = None
1001
1002 for (
1003 (connection, _, hasvalue, has_all_pks, has_all_defaults),
1004 records,
1005 ) in groupby(
1006 insert,
1007 lambda rec: (
1008 rec[4], # connection
1009 set(rec[2]), # parameter keys
1010 bool(rec[5]), # whether we have "value" parameters
1011 rec[6],
1012 rec[7],
1013 ),
1014 ):
1015 statement = cached_stmt
1016
1017 if use_orm_insert_stmt is not None:
1018 statement = statement._annotate(
1019 {
1020 "_emit_insert_table": table,
1021 "_emit_insert_mapper": mapper,
1022 }
1023 )
1024
1025 if (
1026 (
1027 not bookkeeping
1028 or (
1029 has_all_defaults
1030 or not base_mapper._prefer_eager_defaults(
1031 connection.dialect, table
1032 )
1033 or not table.implicit_returning
1034 or not connection.dialect.insert_returning
1035 )
1036 )
1037 and not returning_is_required_anyway
1038 and has_all_pks
1039 and not hasvalue
1040 ):
1041 # the "we don't need newly generated values back" section.
1042 # here we have all the PKs, all the defaults or we don't want
1043 # to fetch them, or the dialect doesn't support RETURNING at all
1044 # so we have to post-fetch / use lastrowid anyway.
1045 records = list(records)
1046 multiparams = [rec[2] for rec in records]
1047
1048 result = connection.execute(
1049 statement, multiparams, execution_options=execution_options
1050 )
1051 if bookkeeping:
1052 for (
1053 (
1054 state,
1055 state_dict,
1056 params,
1057 mapper_rec,
1058 conn,
1059 value_params,
1060 has_all_pks,
1061 has_all_defaults,
1062 ),
1063 last_inserted_params,
1064 ) in zip(records, result.context.compiled_parameters):
1065 if state:
1066 _postfetch(
1067 mapper_rec,
1068 uowtransaction,
1069 table,
1070 state,
1071 state_dict,
1072 result,
1073 last_inserted_params,
1074 value_params,
1075 False,
1076 (
1077 result.returned_defaults
1078 if not result.context.executemany
1079 else None
1080 ),
1081 )
1082 else:
1083 _postfetch_bulk_save(mapper_rec, state_dict, table)
1084
1085 else:
1086 # here, we need defaults and/or pk values back or we otherwise
1087 # know that we are using RETURNING in any case
1088
1089 records = list(records)
1090
1091 if returning_is_required_anyway or (
1092 table.implicit_returning and not hasvalue and len(records) > 1
1093 ):
1094 if (
1095 deterministic_results_reqd
1096 and connection.dialect.insert_executemany_returning_sort_by_parameter_order # noqa: E501
1097 ) or (
1098 not deterministic_results_reqd
1099 and connection.dialect.insert_executemany_returning
1100 ):
1101 do_executemany = True
1102 elif returning_is_required_anyway:
1103 if deterministic_results_reqd:
1104 dt = " with RETURNING and sort by parameter order"
1105 else:
1106 dt = " with RETURNING"
1107 raise sa_exc.InvalidRequestError(
1108 f"Can't use explicit RETURNING for bulk INSERT "
1109 f"operation with "
1110 f"{connection.dialect.dialect_description} backend; "
1111 f"executemany{dt} is not enabled for this dialect."
1112 )
1113 else:
1114 do_executemany = False
1115 else:
1116 do_executemany = False
1117
1118 if use_orm_insert_stmt is None:
1119 if (
1120 not has_all_defaults
1121 and base_mapper._prefer_eager_defaults(
1122 connection.dialect, table
1123 )
1124 ):
1125 statement = statement.return_defaults(
1126 *mapper._server_default_cols[table],
1127 sort_by_parameter_order=bookkeeping,
1128 )
1129
1130 if mapper.version_id_col is not None:
1131 statement = statement.return_defaults(
1132 mapper.version_id_col,
1133 sort_by_parameter_order=bookkeeping,
1134 )
1135 elif do_executemany:
1136 statement = statement.return_defaults(
1137 *table.primary_key, sort_by_parameter_order=bookkeeping
1138 )
1139
1140 if do_executemany:
1141 multiparams = [rec[2] for rec in records]
1142
1143 result = connection.execute(
1144 statement, multiparams, execution_options=execution_options
1145 )
1146
1147 if use_orm_insert_stmt is not None:
1148 if return_result is None:
1149 return_result = result
1150 else:
1151 return_result = return_result.splice_vertically(result)
1152
1153 if bookkeeping:
1154 for (
1155 (
1156 state,
1157 state_dict,
1158 params,
1159 mapper_rec,
1160 conn,
1161 value_params,
1162 has_all_pks,
1163 has_all_defaults,
1164 ),
1165 last_inserted_params,
1166 inserted_primary_key,
1167 returned_defaults,
1168 ) in zip_longest(
1169 records,
1170 result.context.compiled_parameters,
1171 result.inserted_primary_key_rows,
1172 result.returned_defaults_rows or (),
1173 ):
1174 if inserted_primary_key is None:
1175 # this is a real problem and means that we didn't
1176 # get back as many PK rows. we can't continue
1177 # since this indicates PK rows were missing, which
1178 # means we likely mis-populated records starting
1179 # at that point with incorrectly matched PK
1180 # values.
1181 raise orm_exc.FlushError(
1182 "Multi-row INSERT statement for %s did not "
1183 "produce "
1184 "the correct number of INSERTed rows for "
1185 "RETURNING. Ensure there are no triggers or "
1186 "special driver issues preventing INSERT from "
1187 "functioning properly." % mapper_rec
1188 )
1189
1190 for pk, col in zip(
1191 inserted_primary_key,
1192 mapper._pks_by_table[table],
1193 ):
1194 prop = mapper_rec._columntoproperty[col]
1195 if state_dict.get(prop.key) is None:
1196 state_dict[prop.key] = pk
1197
1198 if state:
1199 _postfetch(
1200 mapper_rec,