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/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"""private module containing functions used to emit INSERT, UPDATE
11and DELETE statements on behalf of a :class:`_orm.Mapper` and its descending
12mappers.
14The functions here are called only by the unit of work functions
15in unitofwork.py.
17"""
19from __future__ import annotations
21from itertools import chain
22from itertools import groupby
23from itertools import zip_longest
24import operator
26from . import attributes
27from . import exc as orm_exc
28from . import loading
29from . import sync
30from .base import state_str
31from .. import exc as sa_exc
32from .. import sql
33from .. import util
34from ..engine import cursor as _cursor
35from ..sql import operators
36from ..sql.elements import BooleanClauseList
39def _save_obj(base_mapper, states, uowtransaction, single=False):
40 """Issue ``INSERT`` and/or ``UPDATE`` statements for a list
41 of objects.
43 This is called within the context of a UOWTransaction during a
44 flush operation, given a list of states to be flushed. The
45 base mapper in an inheritance hierarchy handles the inserts/
46 updates for all descendant mappers.
48 """
50 # if batch=false, call _save_obj separately for each object
51 if not single and not base_mapper.batch:
52 for state in _sort_states(base_mapper, states):
53 _save_obj(base_mapper, [state], uowtransaction, single=True)
54 return
56 states_to_update = []
57 states_to_insert = []
59 for (
60 state,
61 dict_,
62 mapper,
63 connection,
64 has_identity,
65 row_switch,
66 update_version_id,
67 ) in _organize_states_for_save(base_mapper, states, uowtransaction):
68 if has_identity or row_switch:
69 states_to_update.append(
70 (state, dict_, mapper, connection, update_version_id)
71 )
72 else:
73 states_to_insert.append((state, dict_, mapper, connection))
75 for table, mapper in base_mapper._sorted_tables.items():
76 if table not in mapper._pks_by_table:
77 continue
78 insert = _collect_insert_commands(table, states_to_insert)
80 update = _collect_update_commands(
81 uowtransaction, table, states_to_update
82 )
84 _emit_update_statements(
85 base_mapper,
86 uowtransaction,
87 mapper,
88 table,
89 update,
90 )
92 _emit_insert_statements(
93 base_mapper,
94 uowtransaction,
95 mapper,
96 table,
97 insert,
98 )
100 _finalize_insert_update_commands(
101 base_mapper,
102 uowtransaction,
103 chain(
104 (
105 (state, state_dict, mapper, connection, False)
106 for (state, state_dict, mapper, connection) in states_to_insert
107 ),
108 (
109 (state, state_dict, mapper, connection, True)
110 for (
111 state,
112 state_dict,
113 mapper,
114 connection,
115 update_version_id,
116 ) in states_to_update
117 ),
118 ),
119 )
122def _post_update(base_mapper, states, uowtransaction, post_update_cols):
123 """Issue UPDATE statements on behalf of a relationship() which
124 specifies post_update.
126 """
128 states_to_update = list(
129 _organize_states_for_post_update(base_mapper, states, uowtransaction)
130 )
132 for table, mapper in base_mapper._sorted_tables.items():
133 if table not in mapper._pks_by_table:
134 continue
136 update = (
137 (
138 state,
139 state_dict,
140 sub_mapper,
141 connection,
142 (
143 mapper._get_committed_state_attr_by_column(
144 state, state_dict, mapper.version_id_col
145 )
146 if mapper.version_id_col is not None
147 else None
148 ),
149 )
150 for state, state_dict, sub_mapper, connection in states_to_update
151 if table in sub_mapper._pks_by_table
152 )
154 update = _collect_post_update_commands(
155 base_mapper, uowtransaction, table, update, post_update_cols
156 )
158 _emit_post_update_statements(
159 base_mapper,
160 uowtransaction,
161 mapper,
162 table,
163 update,
164 )
167def _delete_obj(base_mapper, states, uowtransaction):
168 """Issue ``DELETE`` statements for a list of objects.
170 This is called within the context of a UOWTransaction during a
171 flush operation.
173 """
175 states_to_delete = list(
176 _organize_states_for_delete(base_mapper, states, uowtransaction)
177 )
179 table_to_mapper = base_mapper._sorted_tables
181 for table in reversed(list(table_to_mapper.keys())):
182 mapper = table_to_mapper[table]
183 if table not in mapper._pks_by_table:
184 continue
185 elif mapper.inherits and mapper.passive_deletes:
186 continue
188 delete = _collect_delete_commands(
189 base_mapper, uowtransaction, table, states_to_delete
190 )
192 _emit_delete_statements(
193 base_mapper,
194 uowtransaction,
195 mapper,
196 table,
197 delete,
198 )
200 for (
201 state,
202 state_dict,
203 mapper,
204 connection,
205 update_version_id,
206 ) in states_to_delete:
207 mapper.dispatch.after_delete(mapper, connection, state)
210def _organize_states_for_save(base_mapper, states, uowtransaction):
211 """Make an initial pass across a set of states for INSERT or
212 UPDATE.
214 This includes splitting out into distinct lists for
215 each, calling before_insert/before_update, obtaining
216 key information for each state including its dictionary,
217 mapper, the connection to use for the execution per state,
218 and the identity flag.
220 """
222 for state, dict_, mapper, connection in _connections_for_states(
223 base_mapper, uowtransaction, states
224 ):
225 has_identity = bool(state.key)
227 instance_key = state.key or mapper._identity_key_from_state(state)
229 row_switch = update_version_id = None
231 # call before_XXX extensions
232 if not has_identity:
233 mapper.dispatch.before_insert(mapper, connection, state)
234 else:
235 mapper.dispatch.before_update(mapper, connection, state)
237 if mapper._validate_polymorphic_identity:
238 mapper._validate_polymorphic_identity(mapper, state, dict_)
240 # detect if we have a "pending" instance (i.e. has
241 # no instance_key attached to it), and another instance
242 # with the same identity key already exists as persistent.
243 # convert to an UPDATE if so.
244 if (
245 not has_identity
246 and instance_key in uowtransaction.session.identity_map
247 ):
248 instance = uowtransaction.session.identity_map[instance_key]
249 existing = attributes.instance_state(instance)
251 if not uowtransaction.was_already_deleted(existing):
252 if not uowtransaction.is_deleted(existing):
253 util.warn(
254 "New instance %s with identity key %s conflicts "
255 "with persistent instance %s"
256 % (state_str(state), instance_key, state_str(existing))
257 )
258 else:
259 base_mapper._log_debug(
260 "detected row switch for identity %s. "
261 "will update %s, remove %s from "
262 "transaction",
263 instance_key,
264 state_str(state),
265 state_str(existing),
266 )
268 # remove the "delete" flag from the existing element
269 uowtransaction.remove_state_actions(existing)
270 row_switch = existing
272 if (has_identity or row_switch) and mapper.version_id_col is not None:
273 update_version_id = mapper._get_committed_state_attr_by_column(
274 row_switch if row_switch else state,
275 row_switch.dict if row_switch else dict_,
276 mapper.version_id_col,
277 )
279 yield (
280 state,
281 dict_,
282 mapper,
283 connection,
284 has_identity,
285 row_switch,
286 update_version_id,
287 )
290def _organize_states_for_post_update(base_mapper, states, uowtransaction):
291 """Make an initial pass across a set of states for UPDATE
292 corresponding to post_update.
294 This includes obtaining key information for each state
295 including its dictionary, mapper, the connection to use for
296 the execution per state.
298 """
299 return _connections_for_states(base_mapper, uowtransaction, states)
302def _organize_states_for_delete(base_mapper, states, uowtransaction):
303 """Make an initial pass across a set of states for DELETE.
305 This includes calling out before_delete and obtaining
306 key information for each state including its dictionary,
307 mapper, the connection to use for the execution per state.
309 """
310 for state, dict_, mapper, connection in _connections_for_states(
311 base_mapper, uowtransaction, states
312 ):
313 mapper.dispatch.before_delete(mapper, connection, state)
315 if mapper.version_id_col is not None:
316 update_version_id = mapper._get_committed_state_attr_by_column(
317 state, dict_, mapper.version_id_col
318 )
319 else:
320 update_version_id = None
322 yield (state, dict_, mapper, connection, update_version_id)
325def _set_insert_value(col, value, params, value_params):
326 """Place a value for an INSERT into either ``value_params``, if it is a
327 SQL expression, or ``params``, if it is a plain value.
329 """
330 if hasattr(value, "__clause_element__"):
331 value_params[col] = value.__clause_element__()
332 elif isinstance(value, sql.ClauseElement):
333 value_params[col] = value
334 else:
335 params[col.key] = value
338def _collect_insert_commands(
339 table,
340 states_to_insert,
341 *,
342 bulk=False,
343 return_defaults=False,
344 render_nulls=False,
345 include_bulk_keys=(),
346):
347 """Identify sets of values to use in INSERT statements for a
348 list of states.
350 """
351 for state, state_dict, mapper, connection in states_to_insert:
352 if table not in mapper._pks_by_table:
353 continue
355 params = {}
356 value_params = {}
358 propkey_to_col = mapper._propkey_to_col[table]
360 eval_none = mapper._insert_cols_evaluating_none[table]
362 for propkey in set(propkey_to_col).intersection(state_dict):
363 value = state_dict[propkey]
364 col = propkey_to_col[propkey]
365 if value is None and col not in eval_none and not render_nulls:
366 continue
367 elif bulk:
368 params[col.key] = value
369 else:
370 _set_insert_value(col, value, params, value_params)
372 if not bulk:
373 # ORM-level defaults, i.e. dataclass defaults, for columns that
374 # have no value, where the column's own default does not
375 # deliver the value, #13748
376 for propkey, col, value in mapper._insert_orm_defaults[table]:
377 if col.key not in params and col not in value_params:
378 _set_insert_value(col, value, params, value_params)
380 # for all the columns that have no default and we don't have
381 # a value and where "None" is not a special value, add
382 # explicit None to the INSERT. This is a legacy behavior
383 # which might be worth removing, as it should not be necessary
384 # and also produces confusion, given that "missing" and None
385 # now have distinct meanings
386 for colkey in (
387 mapper._insert_cols_as_none[table]
388 .difference(params)
389 .difference([c.key for c in value_params])
390 ):
391 params[colkey] = None
393 if not bulk or return_defaults:
394 # params are in terms of Column key objects, so
395 # compare to pk_keys_by_table
396 has_all_pks = mapper._pk_keys_by_table[table].issubset(params)
398 if mapper.base_mapper._prefer_eager_defaults(
399 connection.dialect, table
400 ):
401 has_all_defaults = mapper._server_default_col_keys[
402 table
403 ].issubset(params)
404 else:
405 has_all_defaults = True
406 else:
407 has_all_defaults = has_all_pks = True
409 if (
410 mapper.version_id_generator is not False
411 and mapper.version_id_col is not None
412 and mapper.version_id_col in mapper._cols_by_table[table]
413 ):
414 params[mapper.version_id_col.key] = mapper.version_id_generator(
415 None
416 )
418 if bulk:
419 if mapper._set_polymorphic_identity:
420 params.setdefault(
421 mapper._polymorphic_attr_key, mapper.polymorphic_identity
422 )
424 if include_bulk_keys:
425 params.update((k, state_dict[k]) for k in include_bulk_keys)
427 yield (
428 state,
429 state_dict,
430 params,
431 mapper,
432 connection,
433 value_params,
434 has_all_pks,
435 has_all_defaults,
436 )
439def _collect_update_commands(
440 uowtransaction,
441 table,
442 states_to_update,
443 *,
444 bulk=False,
445 use_orm_update_stmt=None,
446 include_bulk_keys=(),
447):
448 """Identify sets of values to use in UPDATE statements for a
449 list of states.
451 This function works intricately with the history system
452 to determine exactly what values should be updated
453 as well as how the row should be matched within an UPDATE
454 statement. Includes some tricky scenarios where the primary
455 key of an object might have been changed.
457 """
459 for (
460 state,
461 state_dict,
462 mapper,
463 connection,
464 update_version_id,
465 ) in states_to_update:
466 if table not in mapper._pks_by_table:
467 continue
469 pks = mapper._pks_by_table[table]
471 if (
472 use_orm_update_stmt is not None
473 and not use_orm_update_stmt._maintain_values_ordering
474 ):
475 # TODO: ordered values, etc
476 # ORM bulk_persistence will raise for the maintain_values_ordering
477 # case right now
478 value_params = use_orm_update_stmt._values
479 else:
480 value_params = {}
482 propkey_to_col = mapper._propkey_to_col[table]
484 if bulk:
485 # keys here are mapped attribute keys, so
486 # look at mapper attribute keys for pk
487 params = {
488 propkey_to_col[propkey].key: state_dict[propkey]
489 for propkey in set(propkey_to_col)
490 .intersection(state_dict)
491 .difference(mapper._pk_attr_keys_by_table[table])
492 }
493 has_all_defaults = True
494 else:
495 params = {}
496 for propkey in set(propkey_to_col).intersection(
497 state.committed_state
498 ):
499 value = state_dict[propkey]
500 col = propkey_to_col[propkey]
502 if hasattr(value, "__clause_element__") or isinstance(
503 value, sql.ClauseElement
504 ):
505 value_params[col] = (
506 value.__clause_element__()
507 if hasattr(value, "__clause_element__")
508 else value
509 )
510 # guard against values that generate non-__nonzero__
511 # objects for __eq__()
512 elif (
513 state.manager[propkey].impl.is_equal(
514 value, state.committed_state[propkey]
515 )
516 is not True
517 ):
518 params[col.key] = value
520 if mapper.base_mapper.eager_defaults is True:
521 has_all_defaults = (
522 mapper._server_onupdate_default_col_keys[table]
523 ).issubset(params)
524 else:
525 has_all_defaults = True
527 if (
528 update_version_id is not None
529 and mapper.version_id_col in mapper._cols_by_table[table]
530 ):
531 if not bulk and not (params or value_params):
532 # HACK: check for history in other tables, in case the
533 # history is only in a different table than the one
534 # where the version_id_col is. This logic was lost
535 # from 0.9 -> 1.0.0 and restored in 1.0.6.
536 for prop in mapper._columntoproperty.values():
537 history = state.manager[prop.key].impl.get_history(
538 state, state_dict, attributes.PASSIVE_NO_INITIALIZE
539 )
540 if history.added:
541 break
542 else:
543 # no net change, break
544 continue
546 col = mapper.version_id_col
547 no_params = not params and not value_params
548 params[col._label] = update_version_id
550 if (
551 bulk or col.key not in params
552 ) and mapper.version_id_generator is not False:
553 val = mapper.version_id_generator(update_version_id)
554 params[col.key] = val
555 elif mapper.version_id_generator is False and no_params:
556 # no version id generator, no values set on the table,
557 # and version id wasn't manually incremented.
558 # set version id to itself so we get an UPDATE
559 # statement
560 params[col.key] = update_version_id
562 elif not (params or value_params):
563 continue
565 has_all_pks = True
566 expect_pk_cascaded = False
567 if bulk:
568 # keys here are mapped attribute keys, so
569 # look at mapper attribute keys for pk
570 pk_params = {
571 propkey_to_col[propkey]._label: state_dict.get(propkey)
572 for propkey in set(propkey_to_col).intersection(
573 mapper._pk_attr_keys_by_table[table]
574 )
575 }
576 if util.NONE_SET.intersection(pk_params.values()):
577 raise sa_exc.InvalidRequestError(
578 f"No primary key value supplied for column(s) "
579 f"""{
580 ', '.join(
581 str(c) for c in pks if pk_params[c._label] is None
582 )
583 }; """
584 "per-row ORM Bulk UPDATE by Primary Key requires that "
585 "records contain primary key values",
586 code="bupq",
587 )
589 else:
590 pk_params = {}
591 for col in pks:
592 propkey = mapper._columntoproperty[col].key
594 history = state.manager[propkey].impl.get_history(
595 state, state_dict, attributes.PASSIVE_OFF
596 )
598 if history.added:
599 if (
600 not history.deleted
601 or ("pk_cascaded", state, col)
602 in uowtransaction.attributes
603 ):
604 expect_pk_cascaded = True
605 pk_params[col._label] = history.added[0]
606 params.pop(col.key, None)
607 else:
608 # else, use the old value to locate the row
609 pk_params[col._label] = history.deleted[0]
610 if col in value_params:
611 has_all_pks = False
612 else:
613 pk_params[col._label] = history.unchanged[0]
614 if pk_params[col._label] is None:
615 raise orm_exc.FlushError(
616 "Can't update table %s using NULL for primary "
617 "key value on column %s" % (table, col)
618 )
620 if include_bulk_keys:
621 params.update((k, state_dict[k]) for k in include_bulk_keys)
623 if params or value_params:
624 params.update(pk_params)
625 yield (
626 state,
627 state_dict,
628 params,
629 mapper,
630 connection,
631 value_params,
632 has_all_defaults,
633 has_all_pks,
634 )
635 elif expect_pk_cascaded:
636 # no UPDATE occurs on this table, but we expect that CASCADE rules
637 # have changed the primary key of the row; propagate this event to
638 # other columns that expect to have been modified. this normally
639 # occurs after the UPDATE is emitted however we invoke it here
640 # explicitly in the absence of our invoking an UPDATE
641 for m, equated_pairs in mapper._table_to_equated[table]:
642 sync._populate(
643 state,
644 m,
645 state,
646 m,
647 equated_pairs,
648 uowtransaction,
649 mapper.passive_updates,
650 )
653def _collect_post_update_commands(
654 base_mapper, uowtransaction, table, states_to_update, post_update_cols
655):
656 """Identify sets of values to use in UPDATE statements for a
657 list of states within a post_update operation.
659 """
661 for (
662 state,
663 state_dict,
664 mapper,
665 connection,
666 update_version_id,
667 ) in states_to_update:
668 # assert table in mapper._pks_by_table
670 pks = mapper._pks_by_table[table]
671 params = {}
672 hasdata = False
674 for col in mapper._cols_by_table[table]:
675 if col in pks:
676 params[col._label] = mapper._get_state_attr_by_column(
677 state, state_dict, col, passive=attributes.PASSIVE_OFF
678 )
680 elif col in post_update_cols or col.onupdate is not None:
681 prop = mapper._columntoproperty[col]
682 history = state.manager[prop.key].impl.get_history(
683 state, state_dict, attributes.PASSIVE_NO_INITIALIZE
684 )
685 if history.added:
686 value = history.added[0]
687 params[col.key] = value
688 hasdata = True
689 if hasdata:
690 if (
691 update_version_id is not None
692 and mapper.version_id_col in mapper._cols_by_table[table]
693 ):
694 col = mapper.version_id_col
695 params[col._label] = update_version_id
697 if (
698 bool(state.key)
699 and col.key not in params
700 and mapper.version_id_generator is not False
701 ):
702 val = mapper.version_id_generator(update_version_id)
703 params[col.key] = val
704 yield state, state_dict, mapper, connection, params
707def _collect_delete_commands(
708 base_mapper, uowtransaction, table, states_to_delete
709):
710 """Identify values to use in DELETE statements for a list of
711 states to be deleted."""
713 for (
714 state,
715 state_dict,
716 mapper,
717 connection,
718 update_version_id,
719 ) in states_to_delete:
720 if table not in mapper._pks_by_table:
721 continue
723 params = {}
724 for col in mapper._pks_by_table[table]:
725 params[col.key] = value = (
726 mapper._get_committed_state_attr_by_column(
727 state, state_dict, col
728 )
729 )
730 if value is None:
731 raise orm_exc.FlushError(
732 "Can't delete from table %s "
733 "using NULL for primary "
734 "key value on column %s" % (table, col)
735 )
737 if (
738 update_version_id is not None
739 and mapper.version_id_col in mapper._cols_by_table[table]
740 ):
741 params[mapper.version_id_col.key] = update_version_id
742 yield params, connection
745def _emit_update_statements(
746 base_mapper,
747 uowtransaction,
748 mapper,
749 table,
750 update,
751 *,
752 bookkeeping=True,
753 use_orm_update_stmt=None,
754 enable_check_rowcount=True,
755):
756 """Emit UPDATE statements corresponding to value lists collected
757 by _collect_update_commands()."""
759 needs_version_id = (
760 mapper.version_id_col is not None
761 and mapper.version_id_col in mapper._cols_by_table[table]
762 )
764 execution_options = {"compiled_cache": base_mapper._compiled_cache}
766 def update_stmt(existing_stmt=None):
767 clauses = BooleanClauseList._construct_raw(operators.and_)
769 for col in mapper._pks_by_table[table]:
770 clauses._append_inplace(
771 col == sql.bindparam(col._label, type_=col.type)
772 )
774 if needs_version_id:
775 clauses._append_inplace(
776 mapper.version_id_col
777 == sql.bindparam(
778 mapper.version_id_col._label,
779 type_=mapper.version_id_col.type,
780 )
781 )
783 if existing_stmt is not None:
784 stmt = existing_stmt.where(clauses)
785 else:
786 stmt = table.update().where(clauses)
787 return stmt
789 if use_orm_update_stmt is not None:
790 cached_stmt = update_stmt(use_orm_update_stmt)
792 else:
793 cached_stmt = base_mapper._memo(("update", table), update_stmt)
795 for (
796 (connection, paramkeys, hasvalue, has_all_defaults, has_all_pks),
797 records,
798 ) in groupby(
799 update,
800 lambda rec: (
801 rec[4], # connection
802 set(rec[2]), # set of parameter keys
803 bool(rec[5]), # whether or not we have "value" parameters
804 rec[6], # has_all_defaults
805 rec[7], # has all pks
806 ),
807 ):
808 rows = 0
809 records = list(records)
811 statement = cached_stmt
813 if use_orm_update_stmt is not None:
814 statement = statement._annotate(
815 {
816 "_emit_update_table": table,
817 "_emit_update_mapper": mapper,
818 }
819 )
821 return_defaults = False
823 if not has_all_pks:
824 statement = statement.return_defaults(*mapper._pks_by_table[table])
825 return_defaults = True
827 if (
828 bookkeeping
829 and not has_all_defaults
830 and mapper.base_mapper.eager_defaults is True
831 # change as of #8889 - if RETURNING is not going to be used anyway,
832 # (applies to MySQL, MariaDB which lack UPDATE RETURNING) ensure
833 # we can do an executemany UPDATE which is more efficient
834 and table.implicit_returning
835 and connection.dialect.update_returning
836 ):
837 statement = statement.return_defaults(
838 *mapper._server_onupdate_default_cols[table]
839 )
840 return_defaults = True
842 if mapper._version_id_has_server_side_value:
843 statement = statement.return_defaults(mapper.version_id_col)
844 return_defaults = True
846 assert_singlerow = connection.dialect.supports_sane_rowcount
848 assert_multirow = (
849 assert_singlerow
850 and connection.dialect.supports_sane_multi_rowcount
851 )
853 # change as of #8889 - if RETURNING is not going to be used anyway,
854 # (applies to MySQL, MariaDB which lack UPDATE RETURNING) ensure
855 # we can do an executemany UPDATE which is more efficient
856 allow_executemany = not return_defaults and not needs_version_id
858 if hasvalue:
859 for (
860 state,
861 state_dict,
862 params,
863 mapper,
864 connection,
865 value_params,
866 has_all_defaults,
867 has_all_pks,
868 ) in records:
869 c = connection.execute(
870 statement.values(value_params),
871 params,
872 execution_options=execution_options,
873 )
874 if bookkeeping:
875 _postfetch(
876 mapper,
877 uowtransaction,
878 table,
879 state,
880 state_dict,
881 c,
882 c.context.compiled_parameters[0],
883 value_params,
884 True,
885 c.returned_defaults,
886 )
887 rows += c.rowcount
888 check_rowcount = enable_check_rowcount and assert_singlerow
889 else:
890 if not allow_executemany:
891 check_rowcount = enable_check_rowcount and assert_singlerow
892 for (
893 state,
894 state_dict,
895 params,
896 mapper,
897 connection,
898 value_params,
899 has_all_defaults,
900 has_all_pks,
901 ) in records:
902 c = connection.execute(
903 statement, params, execution_options=execution_options
904 )
906 # TODO: why with bookkeeping=False?
907 if bookkeeping:
908 _postfetch(
909 mapper,
910 uowtransaction,
911 table,
912 state,
913 state_dict,
914 c,
915 c.context.compiled_parameters[0],
916 value_params,
917 True,
918 c.returned_defaults,
919 )
920 rows += c.rowcount
921 else:
922 multiparams = [rec[2] for rec in records]
924 check_rowcount = enable_check_rowcount and (
925 assert_multirow
926 or (assert_singlerow and len(multiparams) == 1)
927 )
929 c = connection.execute(
930 statement, multiparams, execution_options=execution_options
931 )
933 rows += c.rowcount
935 for (
936 state,
937 state_dict,
938 params,
939 mapper,
940 connection,
941 value_params,
942 has_all_defaults,
943 has_all_pks,
944 ) in records:
945 if bookkeeping:
946 _postfetch(
947 mapper,
948 uowtransaction,
949 table,
950 state,
951 state_dict,
952 c,
953 c.context.compiled_parameters[0],
954 value_params,
955 True,
956 (
957 c.returned_defaults
958 if not c.context.executemany
959 else None
960 ),
961 )
963 if check_rowcount:
964 if rows != len(records):
965 raise orm_exc.StaleDataError(
966 "UPDATE statement on table '%s' expected to "
967 "update %d row(s); %d were matched."
968 % (table.description, len(records), rows)
969 )
971 elif needs_version_id:
972 util.warn(
973 "Dialect %s does not support updated rowcount "
974 "- versioning cannot be verified."
975 % c.dialect.dialect_description
976 )
979def _emit_insert_statements(
980 base_mapper,
981 uowtransaction,
982 mapper,
983 table,
984 insert,
985 *,
986 bookkeeping=True,
987 use_orm_insert_stmt=None,
988 execution_options=None,
989):
990 """Emit INSERT statements corresponding to value lists collected
991 by _collect_insert_commands()."""
993 if use_orm_insert_stmt is not None:
994 cached_stmt = use_orm_insert_stmt
995 exec_opt = util.EMPTY_DICT
997 # if a user query with RETURNING was passed, we definitely need
998 # to use RETURNING.
999 returning_is_required_anyway = bool(use_orm_insert_stmt._returning)
1000 deterministic_results_reqd = (
1001 returning_is_required_anyway
1002 and use_orm_insert_stmt._sort_by_parameter_order
1003 ) or bookkeeping
1004 else:
1005 returning_is_required_anyway = False
1006 deterministic_results_reqd = bookkeeping
1007 cached_stmt = base_mapper._memo(("insert", table), table.insert)
1008 exec_opt = {"compiled_cache": base_mapper._compiled_cache}
1010 if execution_options:
1011 execution_options = util.EMPTY_DICT.merge_with(
1012 exec_opt, execution_options
1013 )
1014 else:
1015 execution_options = exec_opt
1017 return_result = None
1019 for (
1020 (connection, _, hasvalue, has_all_pks, has_all_defaults),
1021 records,
1022 ) in groupby(
1023 insert,
1024 lambda rec: (
1025 rec[4], # connection
1026 set(rec[2]), # parameter keys
1027 bool(rec[5]), # whether we have "value" parameters
1028 rec[6],
1029 rec[7],
1030 ),
1031 ):
1032 statement = cached_stmt
1034 if use_orm_insert_stmt is not None:
1035 statement = statement._annotate(
1036 {
1037 "_emit_insert_table": table,
1038 "_emit_insert_mapper": mapper,
1039 }
1040 )
1042 if (
1043 (
1044 not bookkeeping
1045 or (
1046 has_all_defaults
1047 or not base_mapper._prefer_eager_defaults(
1048 connection.dialect, table
1049 )
1050 or not table.implicit_returning
1051 or not connection.dialect.insert_returning
1052 )
1053 )
1054 and not returning_is_required_anyway
1055 and has_all_pks
1056 and not hasvalue
1057 ):
1058 # the "we don't need newly generated values back" section.
1059 # here we have all the PKs, all the defaults or we don't want
1060 # to fetch them, or the dialect doesn't support RETURNING at all
1061 # so we have to post-fetch / use lastrowid anyway.
1062 records = list(records)
1063 multiparams = [rec[2] for rec in records]
1065 result = connection.execute(
1066 statement, multiparams, execution_options=execution_options
1067 )
1068 if bookkeeping:
1069 for (
1070 (
1071 state,
1072 state_dict,
1073 params,
1074 mapper_rec,
1075 conn,
1076 value_params,
1077 has_all_pks,
1078 has_all_defaults,
1079 ),
1080 last_inserted_params,
1081 ) in zip(records, result.context.compiled_parameters):
1082 if state:
1083 _postfetch(
1084 mapper_rec,
1085 uowtransaction,
1086 table,
1087 state,
1088 state_dict,
1089 result,
1090 last_inserted_params,
1091 value_params,
1092 False,
1093 (
1094 result.returned_defaults
1095 if not result.context.executemany
1096 else None
1097 ),
1098 )
1099 else:
1100 _postfetch_bulk_save(mapper_rec, state_dict, table)
1102 else:
1103 # here, we need defaults and/or pk values back or we otherwise
1104 # know that we are using RETURNING in any case
1106 records = list(records)
1108 if returning_is_required_anyway or (
1109 table.implicit_returning and not hasvalue and len(records) > 1
1110 ):
1111 if (
1112 deterministic_results_reqd
1113 and connection.dialect.insert_executemany_returning_sort_by_parameter_order # noqa: E501
1114 ) or (
1115 not deterministic_results_reqd
1116 and connection.dialect.insert_executemany_returning
1117 ):
1118 do_executemany = True
1119 elif returning_is_required_anyway:
1120 if deterministic_results_reqd:
1121 dt = " with RETURNING and sort by parameter order"
1122 else:
1123 dt = " with RETURNING"
1124 raise sa_exc.InvalidRequestError(
1125 f"Can't use explicit RETURNING for bulk INSERT "
1126 f"operation with "
1127 f"{connection.dialect.dialect_description} backend; "
1128 f"executemany{dt} is not enabled for this dialect."
1129 )
1130 else:
1131 do_executemany = False
1132 else:
1133 do_executemany = False
1135 if use_orm_insert_stmt is None:
1136 if (
1137 not has_all_defaults
1138 and base_mapper._prefer_eager_defaults(
1139 connection.dialect, table
1140 )
1141 ):
1142 statement = statement.return_defaults(
1143 *mapper._server_default_cols[table],
1144 sort_by_parameter_order=bookkeeping,
1145 )
1147 if mapper.version_id_col is not None:
1148 statement = statement.return_defaults(
1149 mapper.version_id_col,
1150 sort_by_parameter_order=bookkeeping,
1151 )
1152 elif do_executemany:
1153 statement = statement.return_defaults(
1154 *table.primary_key, sort_by_parameter_order=bookkeeping
1155 )
1157 if do_executemany:
1158 multiparams = [rec[2] for rec in records]
1160 result = connection.execute(
1161 statement, multiparams, execution_options=execution_options
1162 )
1164 if use_orm_insert_stmt is not None:
1165 if return_result is None:
1166 return_result = result
1167 else:
1168 return_result = return_result.splice_vertically(result)
1170 if bookkeeping:
1171 for (
1172 (
1173 state,
1174 state_dict,
1175 params,
1176 mapper_rec,
1177 conn,
1178 value_params,
1179 has_all_pks,
1180 has_all_defaults,
1181 ),
1182 last_inserted_params,
1183 inserted_primary_key,
1184 returned_defaults,
1185 ) in zip_longest(
1186 records,
1187 result.context.compiled_parameters,
1188 result.inserted_primary_key_rows,
1189 result.returned_defaults_rows or (),
1190 ):
1191 if inserted_primary_key is None:
1192 # this is a real problem and means that we didn't
1193 # get back as many PK rows. we can't continue
1194 # since this indicates PK rows were missing, which
1195 # means we likely mis-populated records starting
1196 # at that point with incorrectly matched PK
1197 # values.
1198 raise orm_exc.FlushError(
1199 "Multi-row INSERT statement for %s did not "
1200 "produce "
1201 "the correct number of INSERTed rows for "
1202 "RETURNING. Ensure there are no triggers or "
1203 "special driver issues preventing INSERT from "
1204 "functioning properly." % mapper_rec
1205 )
1207 for pk, col in zip(
1208 inserted_primary_key,
1209 mapper._pks_by_table[table],
1210 ):
1211 prop = mapper_rec._columntoproperty[col]
1212 if state_dict.get(prop.key) is None:
1213 state_dict[prop.key] = pk
1215 if state:
1216 _postfetch(
1217 mapper_rec,
1218 uowtransaction,
1219 table,
1220 state,
1221 state_dict,
1222 result,
1223 last_inserted_params,
1224 value_params,
1225 False,
1226 returned_defaults,
1227 )
1228 else:
1229 _postfetch_bulk_save(mapper_rec, state_dict, table)
1230 else:
1231 assert not returning_is_required_anyway
1233 for (
1234 state,
1235 state_dict,
1236 params,
1237 mapper_rec,
1238 connection,
1239 value_params,
1240 has_all_pks,
1241 has_all_defaults,
1242 ) in records:
1243 if value_params:
1244 result = connection.execute(
1245 statement.values(value_params),
1246 params,
1247 execution_options=execution_options,
1248 )
1249 else:
1250 result = connection.execute(
1251 statement,
1252 params,
1253 execution_options=execution_options,
1254 )
1256 primary_key = result.inserted_primary_key
1257 if primary_key is None:
1258 raise orm_exc.FlushError(
1259 "Single-row INSERT statement for %s "
1260 "did not produce a "
1261 "new primary key result "
1262 "being invoked. Ensure there are no triggers or "
1263 "special driver issues preventing INSERT from "
1264 "functioning properly." % (mapper_rec,)
1265 )
1266 for pk, col in zip(
1267 primary_key, mapper._pks_by_table[table]
1268 ):
1269 prop = mapper_rec._columntoproperty[col]
1270 if (
1271 col in value_params
1272 or state_dict.get(prop.key) is None
1273 ):
1274 state_dict[prop.key] = pk
1275 if bookkeeping:
1276 if state:
1277 _postfetch(
1278 mapper_rec,
1279 uowtransaction,
1280 table,
1281 state,
1282 state_dict,
1283 result,
1284 result.context.compiled_parameters[0],
1285 value_params,
1286 False,
1287 (
1288 result.returned_defaults
1289 if not result.context.executemany
1290 else None
1291 ),
1292 )
1293 else:
1294 _postfetch_bulk_save(mapper_rec, state_dict, table)
1296 if use_orm_insert_stmt is not None:
1297 if return_result is None:
1298 return _cursor.null_dml_result()
1299 else:
1300 return return_result
1303def _emit_post_update_statements(
1304 base_mapper, uowtransaction, mapper, table, update
1305):
1306 """Emit UPDATE statements corresponding to value lists collected
1307 by _collect_post_update_commands()."""
1309 execution_options = {"compiled_cache": base_mapper._compiled_cache}
1311 needs_version_id = (
1312 mapper.version_id_col is not None
1313 and mapper.version_id_col in mapper._cols_by_table[table]
1314 )
1316 def update_stmt():
1317 clauses = BooleanClauseList._construct_raw(operators.and_)
1319 for col in mapper._pks_by_table[table]:
1320 clauses._append_inplace(
1321 col == sql.bindparam(col._label, type_=col.type)
1322 )
1324 if needs_version_id:
1325 clauses._append_inplace(
1326 mapper.version_id_col
1327 == sql.bindparam(
1328 mapper.version_id_col._label,
1329 type_=mapper.version_id_col.type,
1330 )
1331 )
1333 stmt = table.update().where(clauses)
1335 return stmt
1337 statement = base_mapper._memo(("post_update", table), update_stmt)
1339 if mapper._version_id_has_server_side_value:
1340 statement = statement.return_defaults(mapper.version_id_col)
1342 # execute each UPDATE in the order according to the original
1343 # list of states to guarantee row access order, but
1344 # also group them into common (connection, cols) sets
1345 # to support executemany().
1346 for key, records in groupby(
1347 update,
1348 lambda rec: (rec[3], set(rec[4])), # connection # parameter keys
1349 ):
1350 rows = 0
1352 records = list(records)
1353 connection = key[0]
1355 assert_singlerow = connection.dialect.supports_sane_rowcount
1356 assert_multirow = (
1357 assert_singlerow
1358 and connection.dialect.supports_sane_multi_rowcount
1359 )
1360 allow_executemany = not needs_version_id or assert_multirow
1362 if not allow_executemany:
1363 check_rowcount = assert_singlerow
1364 for state, state_dict, mapper_rec, connection, params in records:
1365 c = connection.execute(
1366 statement, params, execution_options=execution_options
1367 )
1369 _postfetch_post_update(
1370 mapper_rec,
1371 uowtransaction,
1372 table,
1373 state,
1374 state_dict,
1375 c,
1376 c.context.compiled_parameters[0],
1377 )
1378 rows += c.rowcount
1379 else:
1380 multiparams = [
1381 params
1382 for state, state_dict, mapper_rec, conn, params in records
1383 ]
1385 check_rowcount = assert_multirow or (
1386 assert_singlerow and len(multiparams) == 1
1387 )
1389 c = connection.execute(
1390 statement, multiparams, execution_options=execution_options
1391 )
1393 rows += c.rowcount
1394 for i, (
1395 state,
1396 state_dict,
1397 mapper_rec,
1398 connection,
1399 params,
1400 ) in enumerate(records):
1401 _postfetch_post_update(
1402 mapper_rec,
1403 uowtransaction,
1404 table,
1405 state,
1406 state_dict,
1407 c,
1408 c.context.compiled_parameters[i],
1409 )
1411 if check_rowcount:
1412 if rows != len(records):
1413 raise orm_exc.StaleDataError(
1414 "UPDATE statement on table '%s' expected to "
1415 "update %d row(s); %d were matched."
1416 % (table.description, len(records), rows)
1417 )
1419 elif needs_version_id:
1420 util.warn(
1421 "Dialect %s does not support updated rowcount "
1422 "- versioning cannot be verified."
1423 % c.dialect.dialect_description
1424 )
1427def _emit_delete_statements(
1428 base_mapper, uowtransaction, mapper, table, delete
1429):
1430 """Emit DELETE statements corresponding to value lists collected
1431 by _collect_delete_commands()."""
1433 need_version_id = (
1434 mapper.version_id_col is not None
1435 and mapper.version_id_col in mapper._cols_by_table[table]
1436 )
1438 def delete_stmt():
1439 clauses = BooleanClauseList._construct_raw(operators.and_)
1441 for col in mapper._pks_by_table[table]:
1442 clauses._append_inplace(
1443 col == sql.bindparam(col.key, type_=col.type)
1444 )
1446 if need_version_id:
1447 clauses._append_inplace(
1448 mapper.version_id_col
1449 == sql.bindparam(
1450 mapper.version_id_col.key, type_=mapper.version_id_col.type
1451 )
1452 )
1454 return table.delete().where(clauses)
1456 statement = base_mapper._memo(("delete", table), delete_stmt)
1457 for connection, recs in groupby(delete, lambda rec: rec[1]): # connection
1458 del_objects = [params for params, connection in recs]
1460 execution_options = {"compiled_cache": base_mapper._compiled_cache}
1461 expected = len(del_objects)
1462 rows_matched = -1
1463 only_warn = False
1465 if (
1466 need_version_id
1467 and not connection.dialect.supports_sane_multi_rowcount
1468 ):
1469 if connection.dialect.supports_sane_rowcount:
1470 rows_matched = 0
1471 # execute deletes individually so that versioned
1472 # rows can be verified
1473 for params in del_objects:
1474 c = connection.execute(
1475 statement, params, execution_options=execution_options
1476 )
1477 rows_matched += c.rowcount
1478 else:
1479 util.warn(
1480 "Dialect %s does not support deleted rowcount "
1481 "- versioning cannot be verified."
1482 % connection.dialect.dialect_description
1483 )
1484 connection.execute(
1485 statement, del_objects, execution_options=execution_options
1486 )
1487 else:
1488 c = connection.execute(
1489 statement, del_objects, execution_options=execution_options
1490 )
1492 if not need_version_id:
1493 only_warn = True
1495 rows_matched = c.rowcount
1497 if (
1498 base_mapper.confirm_deleted_rows
1499 and rows_matched > -1
1500 and expected != rows_matched
1501 and (
1502 connection.dialect.supports_sane_multi_rowcount
1503 or len(del_objects) == 1
1504 )
1505 ):
1506 # TODO: why does this "only warn" if versioning is turned off,
1507 # whereas the UPDATE raises?
1508 if only_warn:
1509 util.warn(
1510 "DELETE statement on table '%s' expected to "
1511 "delete %d row(s); %d were matched. Please set "
1512 "confirm_deleted_rows=False within the mapper "
1513 "configuration to prevent this warning."
1514 % (table.description, expected, rows_matched)
1515 )
1516 else:
1517 raise orm_exc.StaleDataError(
1518 "DELETE statement on table '%s' expected to "
1519 "delete %d row(s); %d were matched. Please set "
1520 "confirm_deleted_rows=False within the mapper "
1521 "configuration to prevent this warning."
1522 % (table.description, expected, rows_matched)
1523 )
1526def _finalize_insert_update_commands(base_mapper, uowtransaction, states):
1527 """finalize state on states that have been inserted or updated,
1528 including calling after_insert/after_update events.
1530 """
1531 for state, state_dict, mapper, connection, has_identity in states:
1532 if mapper._readonly_props:
1533 readonly = state.unmodified_intersection(
1534 [
1535 p.key
1536 for p in mapper._readonly_props
1537 if (
1538 p.expire_on_flush
1539 and (not p.deferred or p.key in state.dict)
1540 )
1541 or (
1542 not p.expire_on_flush
1543 and not p.deferred
1544 and p.key not in state.dict
1545 )
1546 ]
1547 )
1548 if readonly:
1549 state._expire_attributes(state.dict, readonly)
1551 # if eager_defaults option is enabled, load
1552 # all expired cols. Else if we have a version_id_col, make sure
1553 # it isn't expired.
1554 toload_now = []
1556 # this is specifically to emit a second SELECT for eager_defaults,
1557 # so only if it's set to True, not "auto"
1558 if base_mapper.eager_defaults is True:
1559 toload_now.extend(
1560 state._unloaded_non_object.intersection(
1561 mapper._server_default_plus_onupdate_propkeys
1562 )
1563 )
1565 if (
1566 mapper.version_id_col is not None
1567 and mapper.version_id_generator is False
1568 ):
1569 if mapper._version_id_prop.key in state.unloaded:
1570 toload_now.extend([mapper._version_id_prop.key])
1572 if toload_now:
1573 state.key = base_mapper._identity_key_from_state(state)
1574 stmt = sql.select(mapper)
1575 loading._load_on_ident(
1576 uowtransaction.session,
1577 stmt,
1578 state.key,
1579 refresh_state=state,
1580 only_load_props=toload_now,
1581 )
1583 # call after_XXX extensions
1584 if not has_identity:
1585 mapper.dispatch.after_insert(mapper, connection, state)
1586 else:
1587 mapper.dispatch.after_update(mapper, connection, state)
1589 if (
1590 mapper.version_id_generator is False
1591 and mapper.version_id_col is not None
1592 ):
1593 if state_dict[mapper._version_id_prop.key] is None:
1594 raise orm_exc.FlushError(
1595 "Instance does not contain a non-NULL version value"
1596 )
1599def _postfetch_post_update(
1600 mapper, uowtransaction, table, state, dict_, result, params
1601):
1602 needs_version_id = (
1603 mapper.version_id_col is not None
1604 and mapper.version_id_col in mapper._cols_by_table[table]
1605 )
1607 if not uowtransaction.is_deleted(state):
1608 # post updating after a regular INSERT or UPDATE, do a full postfetch
1609 prefetch_cols = result.context.compiled.prefetch
1610 postfetch_cols = result.context.compiled.postfetch
1611 elif needs_version_id:
1612 # post updating before a DELETE with a version_id_col, need to
1613 # postfetch just version_id_col
1614 prefetch_cols = postfetch_cols = ()
1615 else:
1616 # post updating before a DELETE without a version_id_col,
1617 # don't need to postfetch
1618 return
1620 if needs_version_id:
1621 prefetch_cols = list(prefetch_cols) + [mapper.version_id_col]
1623 refresh_flush = bool(mapper.class_manager.dispatch.refresh_flush)
1624 if refresh_flush:
1625 load_evt_attrs = []
1627 for c in prefetch_cols:
1628 if c.key in params and c in mapper._columntoproperty:
1629 dict_[mapper._columntoproperty[c].key] = params[c.key]
1630 if refresh_flush:
1631 load_evt_attrs.append(mapper._columntoproperty[c].key)
1633 if refresh_flush and load_evt_attrs:
1634 mapper.class_manager.dispatch.refresh_flush(
1635 state, uowtransaction, load_evt_attrs
1636 )
1638 if postfetch_cols:
1639 state._expire_attributes(
1640 state.dict,
1641 [
1642 mapper._columntoproperty[c].key
1643 for c in postfetch_cols
1644 if c in mapper._columntoproperty
1645 ],
1646 )
1649def _postfetch(
1650 mapper,
1651 uowtransaction,
1652 table,
1653 state,
1654 dict_,
1655 result,
1656 params,
1657 value_params,
1658 isupdate,
1659 returned_defaults,
1660):
1661 """Expire attributes in need of newly persisted database state,
1662 after an INSERT or UPDATE statement has proceeded for that
1663 state."""
1665 prefetch_cols = result.context.compiled.prefetch
1666 postfetch_cols = result.context.compiled.postfetch
1667 returning_cols = result.context.compiled.effective_returning
1669 if (
1670 mapper.version_id_col is not None
1671 and mapper.version_id_col in mapper._cols_by_table[table]
1672 ):
1673 prefetch_cols = list(prefetch_cols) + [mapper.version_id_col]
1675 if not isupdate:
1676 # ORM-level defaults that _collect_insert_commands delivered to the
1677 # INSERT for attributes with no value are handled in the same way as
1678 # prefetched column defaults, #13748
1679 orm_default_cols = [
1680 col
1681 for pkey, col, _ in mapper._insert_orm_defaults[table]
1682 if dict_.get(pkey) is None and params.get(col.key) is not None
1683 ]
1684 if orm_default_cols:
1685 prefetch_cols = list(prefetch_cols) + orm_default_cols
1687 refresh_flush = bool(mapper.class_manager.dispatch.refresh_flush)
1688 if refresh_flush:
1689 load_evt_attrs = []
1691 if returning_cols:
1692 row = returned_defaults
1693 if row is not None:
1694 for row_value, col in zip(row, returning_cols):
1695 # pk cols returned from insert are handled
1696 # distinctly, don't step on the values here
1697 if col.primary_key and result.context.isinsert:
1698 continue
1700 # note that columns can be in the "return defaults" that are
1701 # not mapped to this mapper, typically because they are
1702 # "excluded", which can be specified directly or also occurs
1703 # when using declarative w/ single table inheritance
1704 prop = mapper._columntoproperty.get(col)
1705 if prop:
1706 dict_[prop.key] = row_value
1707 if refresh_flush:
1708 load_evt_attrs.append(prop.key)
1710 for c in prefetch_cols:
1711 if c.key in params and c in mapper._columntoproperty:
1712 pkey = mapper._columntoproperty[c].key
1714 # set prefetched value in dict and also pop from committed_state,
1715 # since this is new database state that replaces whatever might
1716 # have previously been fetched (see #10800). this is essentially a
1717 # shorthand version of set_committed_value(), which could also be
1718 # used here directly (with more overhead)
1719 dict_[pkey] = params[c.key]
1720 state.committed_state.pop(pkey, None)
1722 if refresh_flush:
1723 load_evt_attrs.append(pkey)
1725 if refresh_flush and load_evt_attrs:
1726 mapper.class_manager.dispatch.refresh_flush(
1727 state, uowtransaction, load_evt_attrs
1728 )
1730 if isupdate and value_params:
1731 # explicitly suit the use case specified by
1732 # [ticket:3801], PK SQL expressions for UPDATE on non-RETURNING
1733 # database which are set to themselves in order to do a version bump.
1734 postfetch_cols.extend(
1735 [
1736 col
1737 for col in value_params
1738 if col.primary_key and col not in returning_cols
1739 ]
1740 )
1742 if postfetch_cols:
1743 state._expire_attributes(
1744 state.dict,
1745 [
1746 mapper._columntoproperty[c].key
1747 for c in postfetch_cols
1748 if c in mapper._columntoproperty
1749 ],
1750 )
1752 # synchronize newly inserted ids from one table to the next
1753 # TODO: this still goes a little too often. would be nice to
1754 # have definitive list of "columns that changed" here
1755 for m, equated_pairs in mapper._table_to_equated[table]:
1756 sync._populate(
1757 state,
1758 m,
1759 state,
1760 m,
1761 equated_pairs,
1762 uowtransaction,
1763 mapper.passive_updates,
1764 )
1767def _postfetch_bulk_save(mapper, dict_, table):
1768 for m, equated_pairs in mapper._table_to_equated[table]:
1769 sync._bulk_populate_inherit_keys(dict_, m, equated_pairs)
1772def _connections_for_states(base_mapper, uowtransaction, states):
1773 """Return an iterator of (state, state.dict, mapper, connection).
1775 The states are sorted according to _sort_states, then paired
1776 with the connection they should be using for the given
1777 unit of work transaction.
1779 """
1780 # if session has a connection callable,
1781 # organize individual states with the connection
1782 # to use for update
1783 if uowtransaction.session.connection_callable:
1784 connection_callable = uowtransaction.session.connection_callable
1785 else:
1786 connection = uowtransaction.transaction.connection(base_mapper)
1787 connection_callable = None
1789 for state in _sort_states(base_mapper, states):
1790 if connection_callable:
1791 connection = connection_callable(base_mapper, state.obj())
1793 mapper = state.manager.mapper
1795 yield state, state.dict, mapper, connection
1798def _sort_states(mapper, states):
1799 pending = set(states)
1800 persistent = {s for s in pending if s.key is not None}
1801 pending.difference_update(persistent)
1803 try:
1804 persistent_sorted = sorted(
1805 persistent, key=mapper._persistent_sortkey_fn
1806 )
1807 except TypeError as err:
1808 raise sa_exc.InvalidRequestError(
1809 "Could not sort objects by primary key; primary key "
1810 "values must be sortable in Python (was: %s)" % err
1811 ) from err
1812 return (
1813 sorted(pending, key=operator.attrgetter("insert_order"))
1814 + persistent_sorted
1815 )