Function _emit_post_update_statements

lib/sqlalchemy/orm/persistence.py:1290–1411

Emit UPDATE statements corresponding to value lists collected by _collect_post_update_commands().

(
    base_mapper, uowtransaction, mapper, table, update
)

def _emit_post_update_statements(
    base_mapper, uowtransaction, mapper, table, update
):
    """Emit UPDATE statements corresponding to value lists collected
    by _collect_post_update_commands()."""

    execution_options = {"compiled_cache": base_mapper._compiled_cache}

    needs_version_id = (
        mapper.version_id_col is not None
        and mapper.version_id_col in mapper._cols_by_table[table]
    )

    def update_stmt():
        clauses = BooleanClauseList._construct_raw(operators.and_)

        for col in mapper._pks_by_table[table]:
            clauses._append_inplace(
                col == sql.bindparam(col._label, type_=col.type)
            )

        if needs_version_id:
            clauses._append_inplace(
                mapper.version_id_col
                == sql.bindparam(
                    mapper.version_id_col._label,
                    type_=mapper.version_id_col.type,
                )
            )

        stmt = table.update().where(clauses)

        return stmt

    statement = base_mapper._memo(("post_update", table), update_stmt)

    if mapper._version_id_has_server_side_value:
        statement = statement.return_defaults(mapper.version_id_col)

    # execute each UPDATE in the order according to the original
    # list of states to guarantee row access order, but
    # also group them into common (connection, cols) sets
    # to support executemany().
    for key, records in groupby(
        update,
        lambda rec: (rec[3], set(rec[4])),  # connection  # parameter keys
    ):
        rows = 0

        records = list(records)
        connection = key[0]

        assert_singlerow = connection.dialect.supports_sane_rowcount
        assert_multirow = (
            assert_singlerow
            and connection.dialect.supports_sane_multi_rowcount
        )
        allow_executemany = not needs_version_id or assert_multirow