Function _emit_update_statements

lib/sqlalchemy/orm/persistence.py:732–963

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

(
    base_mapper,
    uowtransaction,
    mapper,
    table,
    update,
    *,
    bookkeeping=True,
    use_orm_update_stmt=None,
    enable_check_rowcount=True,
)

def _emit_update_statements(
    base_mapper,
    uowtransaction,
    mapper,
    table,
    update,
    *,
    bookkeeping=True,
    use_orm_update_stmt=None,
    enable_check_rowcount=True,
):
    """Emit UPDATE statements corresponding to value lists collected
    by _collect_update_commands()."""

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

    execution_options = {"compiled_cache": base_mapper._compiled_cache}

    def update_stmt(existing_stmt=None):
        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,
                )
            )

        if existing_stmt is not None:
            stmt = existing_stmt.where(clauses)
        else:
            stmt = table.update().where(clauses)
        return stmt

    if use_orm_update_stmt is not None:
        cached_stmt = update_stmt(use_orm_update_stmt)

    else:
        cached_stmt = base_mapper._memo(("update", table), update_stmt)

    for (
        (connection, paramkeys, hasvalue, has_all_defaults, has_all_pks),
        records,
    ) in groupby(
        update,
        lambda rec: (
            rec[4],  # connection
            set(rec[2]),  # set of parameter keys