# orm/bulk_persistence.py

# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors

# <see AUTHORS file>

#

# This module is part of SQLAlchemy and is released under

# the MIT License: https://www.opensource.org/licenses/mit-license.php

# mypy: ignore-errors

"""additional ORM persistence classes related to "bulk" operations,

specifically outside of the flush() process.

"""

from __future__ import annotations

from typing import Any

from typing import cast

from typing import Dict

from typing import Iterable

from typing import Literal

from typing import Optional

from typing import overload

from typing import TYPE_CHECKING

from typing import TypeVar

from typing import Union

from . import attributes

from . import context

from . import evaluator

from . import exc as orm_exc

from . import loading

from . import persistence

from .base import NO_VALUE

from .context import _AbstractORMCompileState

from .context import _ORMFromStatementCompileState

from .context import FromStatement

from .context import QueryContext

from .interfaces import PropComparator

from .. import exc as sa_exc

from .. import util

from ..engine import Dialect

from ..engine import result as _result

from ..sql import coercions

from ..sql import dml

from ..sql import expression

from ..sql import roles

from ..sql import select

from ..sql import sqltypes

from ..sql.base import _entity_namespace_key

from ..sql.base import CompileState

from ..sql.base import Options

from ..sql.dml import DeleteDMLState

from ..sql.dml import InsertDMLState

from ..sql.dml import UpdateDMLState

from ..util import EMPTY_DICT

from ..util.typing import TupleAny

from ..util.typing import Unpack

if TYPE_CHECKING:

from ._typing import DMLStrategyArgument

from ._typing import OrmExecuteOptionsParameter

from ._typing import SynchronizeSessionArgument

from .mapper import Mapper

from .session import _BindArguments

from .session import ORMExecuteState

from .session import Session

from .session import SessionTransaction

from .state import InstanceState

from ..engine import Connection

from ..engine import cursor

from ..engine.interfaces import _CoreAnyExecuteParams

_O = TypeVar("_O", bound=object)

@overload

def _bulk_insert(

mapper: Mapper[_O],

mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],

session_transaction: SessionTransaction,

*,

isstates: bool,

return_defaults: bool,

render_nulls: bool,

use_orm_insert_stmt: Literal[None] = ...,

execution_options: Optional[OrmExecuteOptionsParameter] = ...,

) -> None: ...

@overload

def _bulk_insert(

mapper: Mapper[_O],

mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],

session_transaction: SessionTransaction,

*,

isstates: bool,

return_defaults: bool,

render_nulls: bool,

use_orm_insert_stmt: Optional[dml.Insert] = ...,

execution_options: Optional[OrmExecuteOptionsParameter] = ...,

) -> cursor.CursorResult[Any]: ...

def _bulk_insert(

mapper: Mapper[_O],

mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],

session_transaction: SessionTransaction,

*,

isstates: bool,

return_defaults: bool,

render_nulls: bool,

use_orm_insert_stmt: Optional[dml.Insert] = None,

execution_options: Optional[OrmExecuteOptionsParameter] = None,

) -> Optional[cursor.CursorResult[Any]]:

base_mapper = mapper.base_mapper

if session_transaction.session.connection_callable:

raise NotImplementedError(

"connection_callable / per-instance sharding "

"not supported in bulk_insert()"

)

if isstates:

if TYPE_CHECKING:

mappings = cast(Iterable[InstanceState[_O]], mappings)

if return_defaults:

# list of states allows us to attach .key for return_defaults case

states = [(state, state.dict) for state in mappings]

mappings = [dict_ for (state, dict_) in states]

else:

mappings = [state.dict for state in mappings]

else:

if TYPE_CHECKING:

mappings = cast(Iterable[Dict[str, Any]], mappings)

if return_defaults:

# use dictionaries given, so that newly populated defaults

# can be delivered back to the caller (see #11661). This is **not**

# compatible with other use cases such as a session-executed

# insert() construct, as this will confuse the case of

# insert-per-subclass for joined inheritance cases (see

# test_bulk_statements.py::BulkDMLReturningJoinedInhTest).

#

# So in this conditional, we have **only** called

# session.bulk_insert_mappings() which does not have this

# requirement

mappings = list(mappings)

else:

# for all other cases we need to establish a local dictionary

# so that the incoming dictionaries aren't mutated

mappings = [dict(m) for m in mappings]

_expand_other_attrs(mapper, mappings)

connection = session_transaction.connection(base_mapper)

return_result: Optional[cursor.CursorResult[Any]] = None

mappers_to_run = [

(table, mp)

for table, mp in base_mapper._sorted_tables.items()

if table in mapper._pks_by_table

]

if return_defaults:

# not used by new-style bulk inserts, only used for legacy

bookkeeping = True

elif len(mappers_to_run) > 1:

# if we have more than one table, mapper to run where we will be

# either horizontally splicing, or copying values between tables,

# we need the "bookkeeping" / deterministic returning order

bookkeeping = True

else:

bookkeeping = False

for table, super_mapper in mappers_to_run:

# find bindparams in the statement. For bulk, we don't really know if

# a key in the params applies to a different table since we are

# potentially inserting for multiple tables here; looking at the

# bindparam() is a lot more direct. in most cases this will

# use _generate_cache_key() which is memoized, although in practice

# the ultimate statement that's executed is probably not the same

# object so that memoization might not matter much.

extra_bp_names = (

[

b.key

for b in use_orm_insert_stmt._get_embedded_bindparams()

if b.key in mappings[0]

]

if use_orm_insert_stmt is not None

else ()

)

records = (

(

None,

state_dict,

params,

mapper,

connection,

value_params,

has_all_pks,

has_all_defaults,

)

for (

state,

state_dict,

params,

mp,

conn,

value_params,

has_all_pks,

has_all_defaults,

) in persistence._collect_insert_commands(

table,

((None, mapping, mapper, connection) for mapping in mappings),

bulk=True,

return_defaults=bookkeeping,

render_nulls=render_nulls,

include_bulk_keys=extra_bp_names,

)

)

result = persistence._emit_insert_statements(

base_mapper,

None,

super_mapper,

table,

records,

bookkeeping=bookkeeping,

use_orm_insert_stmt=use_orm_insert_stmt,

execution_options=execution_options,

)

if use_orm_insert_stmt is not None:

if not use_orm_insert_stmt._returning or return_result is None:

return_result = result

elif result.returns_rows:

assert bookkeeping

return_result = return_result.splice_horizontally(result)

if return_defaults and isstates:

identity_cls = mapper._identity_class

identity_props = [p.key for p in mapper._identity_key_props]

for state, dict_ in states:

state.key = (

identity_cls,

tuple([dict_[key] for key in identity_props]),

None,

)

if use_orm_insert_stmt is not None:

assert return_result is not None

return return_result

@overload

def _bulk_update(

mapper: Mapper[Any],

mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],

session_transaction: SessionTransaction,

*,

isstates: bool,

update_changed_only: bool,

use_orm_update_stmt: Literal[None] = ...,

enable_check_rowcount: bool = True,

) -> None: ...

@overload

def _bulk_update(

mapper: Mapper[Any],

mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],

session_transaction: SessionTransaction,

*,

isstates: bool,

update_changed_only: bool,

use_orm_update_stmt: Optional[dml.Update] = ...,

enable_check_rowcount: bool = True,

) -> _result.Result[Unpack[TupleAny]]: ...

def _bulk_update(

mapper: Mapper[Any],

mappings: Union[Iterable[InstanceState[_O]], Iterable[Dict[str, Any]]],

session_transaction: SessionTransaction,

*,

isstates: bool,

update_changed_only: bool,

use_orm_update_stmt: Optional[dml.Update] = None,

enable_check_rowcount: bool = True,

) -> Optional[_result.Result[Unpack[TupleAny]]]:

base_mapper = mapper.base_mapper

search_keys = mapper._primary_key_propkeys

if mapper._version_id_prop:

search_keys = {mapper._version_id_prop.key}.union(search_keys)

def _changed_dict(mapper, state):

return {

k: v

for k, v in state.dict.items()

if k in state.committed_state or k in search_keys

}

if isstates:

if update_changed_only:

mappings = [_changed_dict(mapper, state) for state in mappings]

else:

mappings = [state.dict for state in mappings]

else:

mappings = [dict(m) for m in mappings]

_expand_other_attrs(mapper, mappings)

if session_transaction.session.connection_callable:

raise NotImplementedError(

"connection_callable / per-instance sharding "

"not supported in bulk_update()"

)

connection = session_transaction.connection(base_mapper)

# find bindparams in the statement. see _bulk_insert for similar

# notes for the insert case

extra_bp_names = (

[

b.key

for b in use_orm_update_stmt._get_embedded_bindparams()

if b.key in mappings[0]

]

if use_orm_update_stmt is not None

else ()

)

for table, super_mapper in base_mapper._sorted_tables.items():

if not mapper.isa(super_mapper) or table not in mapper._pks_by_table:

continue

records = persistence._collect_update_commands(

None,

table,

(

(

None,

mapping,

mapper,

connection,

(

mapping[mapper._version_id_prop.key]

if mapper._version_id_prop

else None

),

)

for mapping in mappings

),

bulk=True,

use_orm_update_stmt=use_orm_update_stmt,

include_bulk_keys=extra_bp_names,

)

persistence._emit_update_statements(

base_mapper,

None,

super_mapper,

table,

records,

bookkeeping=False,

use_orm_update_stmt=use_orm_update_stmt,

enable_check_rowcount=enable_check_rowcount,

)

if use_orm_update_stmt is not None:

return _result.null_result()

def _expand_other_attrs(

mapper: Mapper[Any], mappings: Iterable[Dict[str, Any]]

) -> None:

all_attrs = mapper.all_orm_descriptors

attr_keys = set(all_attrs.keys())

bulk_dml_setters = {

key: setter

for key, setter in (

(key, attr._bulk_dml_setter(key))

for key, attr in (

(key, _entity_namespace_key(mapper, key, default=NO_VALUE))

for key in attr_keys

)

if attr is not NO_VALUE and isinstance(attr, PropComparator)

)

if setter is not None

}

setters_todo = set(bulk_dml_setters)

if not setters_todo:

return

for mapping in mappings:

for key in setters_todo.intersection(mapping):

bulk_dml_setters[key](mapping)

class _ORMDMLState(_AbstractORMCompileState):

is_dml_returning = True

from_statement_ctx: Optional[_ORMFromStatementCompileState] = None

@classmethod

def _get_orm_crud_kv_pairs(

cls, mapper, statement, kv_iterator, needs_to_be_cacheable

):

core_get_crud_kv_pairs = UpdateDMLState._get_crud_kv_pairs

for k, v in kv_iterator:

k = coercions.expect(roles.DMLColumnRole, k)

if isinstance(k, str):

desc = _entity_namespace_key(mapper, k, default=NO_VALUE)

if not isinstance(desc, PropComparator):

yield (

coercions.expect(roles.DMLColumnRole, k),

(

coercions.expect(

roles.ExpressionElementRole,

v,

type_=sqltypes.NullType(),

is_crud=True,

)

if needs_to_be_cacheable

else v

),

)

else:

yield from core_get_crud_kv_pairs(

statement,

desc._bulk_update_tuples(v),

needs_to_be_cacheable,

)

elif "entity_namespace" in k._annotations:

k_anno = k._annotations

attr = _entity_namespace_key(

k_anno["entity_namespace"], k_anno["proxy_key"]

)

assert isinstance(attr, PropComparator)

yield from core_get_crud_kv_pairs(

statement,

attr._bulk_update_tuples(v),

needs_to_be_cacheable,

)

else:

yield (

k,

(

v

if not needs_to_be_cacheable

else coercions.expect(

roles.ExpressionElementRole,

v,

type_=sqltypes.NullType(),

is_crud=True,

)

),

)

@classmethod

def _get_dml_plugin_subject(cls, statement):

plugin_subject = statement.table._propagate_attrs.get("plugin_subject")

if (

not plugin_subject

or not plugin_subject.mapper

or plugin_subject

is not statement._propagate_attrs["plugin_subject"]

):

return None

return plugin_subject

@classmethod

def _get_multi_crud_kv_pairs(cls, statement, kv_iterator):

plugin_subject = cls._get_dml_plugin_subject(statement)

if not plugin_subject:

return UpdateDMLState._get_multi_crud_kv_pairs(

statement, kv_iterator

)

return [

dict(

cls._get_orm_crud_kv_pairs(

plugin_subject.mapper, statement, value_dict.items(), False

)

)

for value_dict in kv_iterator

]

@classmethod

def _get_crud_kv_pairs(cls, statement, kv_iterator, needs_to_be_cacheable):

assert (

needs_to_be_cacheable

), "no test coverage for needs_to_be_cacheable=False"

plugin_subject = cls._get_dml_plugin_subject(statement)

if not plugin_subject:

return UpdateDMLState._get_crud_kv_pairs(

statement, kv_iterator, needs_to_be_cacheable

)

return list(

cls._get_orm_crud_kv_pairs(

plugin_subject.mapper,

statement,

kv_iterator,

needs_to_be_cacheable,

)

)

@classmethod

def get_entity_description(cls, statement):

ext_info = statement.table._annotations["parententity"]

mapper = ext_info.mapper

if ext_info.is_aliased_class:

_label_name = ext_info.name

else:

_label_name = mapper.class_.__name__

return {

"name": _label_name,

"type": mapper.class_,

"expr": ext_info.entity,

"entity": ext_info.entity,

"table": mapper.local_table,

}

@classmethod

def get_returning_column_descriptions(cls, statement):

def _ent_for_col(c):

return c._annotations.get("parententity", None)

def _attr_for_col(c, ent):

if ent is None:

return c

proxy_key = c._annotations.get("proxy_key", None)

if not proxy_key:

return c

else:

return getattr(ent.entity, proxy_key, c)

return [

{

"name": c.key,

"type": c.type,

"expr": _attr_for_col(c, ent),

"aliased": ent.is_aliased_class,

"entity": ent.entity,

}

for c, ent in [

(c, _ent_for_col(c)) for c in statement._all_selected_columns

]

]

def _setup_orm_returning(

self,

compiler,

orm_level_statement,

dml_level_statement,

dml_mapper,

*,

use_supplemental_cols=True,

):

"""establish ORM column handlers for an INSERT, UPDATE, or DELETE

which uses explicit returning().

called within compilation level create_for_statement.

The _return_orm_returning() method then receives the Result

after the statement was executed, and applies ORM loading to the

state that we first established here.

"""

if orm_level_statement._returning:

fs = FromStatement(

orm_level_statement._returning,

dml_level_statement,

_adapt_on_names=False,

)

fs = fs.execution_options(**orm_level_statement._execution_options)

fs = fs.options(*orm_level_statement._with_options)

self.select_statement = fs

self.from_statement_ctx = fsc = (

_ORMFromStatementCompileState.create_for_statement(

fs, compiler

)

)

fsc.setup_dml_returning_compile_state(dml_mapper)

dml_level_statement = dml_level_statement._generate()

dml_level_statement._returning = ()

cols_to_return = [c for c in fsc.primary_columns if c is not None]

# since we are splicing result sets together, make sure there

# are columns of some kind returned in each result set

if not cols_to_return:

cols_to_return.extend(dml_mapper.primary_key)

if use_supplemental_cols:

dml_level_statement = dml_level_statement.return_defaults(

# this is a little weird looking, but by passing

# primary key as the main list of cols, this tells

# return_defaults to omit server-default cols (and

# actually all cols, due to some weird thing we should

# clean up in crud.py).

# Since we have cols_to_return, just return what we asked

# for (plus primary key, which ORM persistence needs since

# we likely set bookkeeping=True here, which is another

# whole thing...). We dont want to clutter the

# statement up with lots of other cols the user didn't

# ask for. see #9685

*dml_mapper.primary_key,

supplemental_cols=cols_to_return,

)

else:

dml_level_statement = dml_level_statement.returning(

*cols_to_return

)

return dml_level_statement

@classmethod

def _return_orm_returning(

cls,

session,

statement,

params,

execution_options,

bind_arguments,

result,

):

execution_context = result.context

compile_state = execution_context.compiled.compile_state

if (

compile_state.from_statement_ctx

and not compile_state.from_statement_ctx.compile_options._is_star

):

load_options = execution_options.get(

"_sa_orm_load_options", QueryContext.default_load_options

)

querycontext = QueryContext(

compile_state.from_statement_ctx,

compile_state.select_statement,

statement,

params,

session,

load_options,

execution_options,

bind_arguments,

)

return loading.instances(result, querycontext)

else:

return result

class _BulkUDCompileState(_ORMDMLState):

class default_update_options(Options):

_dml_strategy: DMLStrategyArgument = "auto"

_synchronize_session: SynchronizeSessionArgument = "auto"

_can_use_returning: bool = False

_is_delete_using: bool = False

_is_update_from: bool = False

_autoflush: bool = True

_subject_mapper: Optional[Mapper[Any]] = None

_resolved_values = EMPTY_DICT

_eval_condition = None

_matched_rows = None

_identity_token = None

_populate_existing: bool = False

@classmethod

def can_use_returning(

cls,

dialect: Dialect,

mapper: Mapper[Any],

*,

is_multitable: bool = False,

is_update_from: bool = False,

is_delete_using: bool = False,

is_executemany: bool = False,

) -> bool:

raise NotImplementedError()

@classmethod

def orm_pre_session_exec(

cls,

session,

statement,

params,

execution_options,

bind_arguments,

is_pre_event,

):

(

update_options,

execution_options,

) = _BulkUDCompileState.default_update_options.from_execution_options(

"_sa_orm_update_options",

{

"synchronize_session",

"autoflush",

"populate_existing",

"identity_token",

"is_delete_using",

"is_update_from",

"dml_strategy",

},

execution_options,

statement._execution_options,

)

bind_arguments["clause"] = statement

try:

plugin_subject = statement._propagate_attrs["plugin_subject"]

except KeyError:

assert False, "statement had 'orm' plugin but no plugin_subject"

else:

if plugin_subject:

bind_arguments["mapper"] = plugin_subject.mapper

update_options += {"_subject_mapper": plugin_subject.mapper}

if "parententity" not in statement.table._annotations:

update_options += {"_dml_strategy": "core_only"}

elif not isinstance(params, list):

if update_options._dml_strategy == "auto":

update_options += {"_dml_strategy": "orm"}

elif update_options._dml_strategy == "bulk":

raise sa_exc.InvalidRequestError(

'Can\'t use "bulk" ORM insert strategy without '

"passing separate parameters"

)

else:

if update_options._dml_strategy == "auto":

update_options += {"_dml_strategy": "bulk"}

sync = update_options._synchronize_session

if sync is not None:

if sync not in ("auto", "evaluate", "fetch", False):

raise sa_exc.ArgumentError(

"Valid strategies for session synchronization "

"are 'auto', 'evaluate', 'fetch', False"

)

if update_options._dml_strategy == "bulk" and sync == "fetch":

raise sa_exc.InvalidRequestError(

"The 'fetch' synchronization strategy is not available "

"for 'bulk' ORM updates (i.e. multiple parameter sets)"

)

if not is_pre_event:

if update_options._autoflush:

session._autoflush()

if update_options._dml_strategy == "orm":

if update_options._synchronize_session == "auto":

update_options = cls._do_pre_synchronize_auto(

session,

statement,

params,

execution_options,

bind_arguments,

update_options,

)

elif update_options._synchronize_session == "evaluate":

update_options = cls._do_pre_synchronize_evaluate(

session,

statement,

params,

execution_options,

bind_arguments,

update_options,

)

elif update_options._synchronize_session == "fetch":

update_options = cls._do_pre_synchronize_fetch(

session,

statement,

params,

execution_options,

bind_arguments,

update_options,

)

elif update_options._dml_strategy == "bulk":

if update_options._synchronize_session == "auto":

update_options += {"_synchronize_session": "evaluate"}

# indicators from the "pre exec" step that are then

# added to the DML statement, which will also be part of the cache

# key. The compile level create_for_statement() method will then

# consume these at compiler time.

statement = statement._annotate(

{

"synchronize_session": update_options._synchronize_session,

"is_delete_using": update_options._is_delete_using,

"is_update_from": update_options._is_update_from,

"dml_strategy": update_options._dml_strategy,

"can_use_returning": update_options._can_use_returning,

}

)

return (

statement,

util.immutabledict(execution_options).union(

{"_sa_orm_update_options": update_options}

),

params,

)

@classmethod

def orm_setup_cursor_result(

cls,

session,

statement,

params,

execution_options,

bind_arguments,

result,

):

# this stage of the execution is called after the

# do_orm_execute event hook. meaning for an extension like

# horizontal sharding, this step happens *within* the horizontal

# sharding event handler which calls session.execute() re-entrantly

# and will occur for each backend individually.

# the sharding extension then returns its own merged result from the

# individual ones we return here.

update_options = execution_options["_sa_orm_update_options"]

if update_options._dml_strategy == "orm":

if update_options._synchronize_session == "evaluate":

cls._do_post_synchronize_evaluate(

session, statement, result, update_options

)

elif update_options._synchronize_session == "fetch":

cls._do_post_synchronize_fetch(

session, statement, result, update_options

)

elif update_options._dml_strategy == "bulk":

if update_options._synchronize_session == "evaluate":

cls._do_post_synchronize_bulk_evaluate(

session, params, result, update_options

)

return result

return cls._return_orm_returning(

session,

statement,

params,

execution_options,

bind_arguments,

result,

)

@classmethod

def _adjust_for_extra_criteria(cls, global_attributes, ext_info):

"""Apply extra criteria filtering.

For all distinct single-table-inheritance mappers represented in the

table being updated or deleted, produce additional WHERE criteria such

that only the appropriate subtypes are selected from the total results.

Additionally, add WHERE criteria originating from LoaderCriteriaOptions

collected from the statement.

"""

return_crit = ()

adapter = ext_info._adapter if ext_info.is_aliased_class else None

if (

"additional_entity_criteria",

ext_info.mapper,

) in global_attributes:

return_crit += tuple(

ae._resolve_where_criteria(ext_info)

for ae in global_attributes[

("additional_entity_criteria", ext_info.mapper)

]

if ae.include_aliases or ae.entity is ext_info

)

if ext_info.mapper._single_table_criterion is not None:

return_crit += (ext_info.mapper._single_table_criterion,)

if adapter:

return_crit = tuple(adapter.traverse(crit) for crit in return_crit)

return return_crit

@classmethod

def _interpret_returning_rows(cls, result, mapper, rows):

"""return rows that indicate PK cols in mapper.primary_key position

for RETURNING rows.

Prior to 2.0.36, this method seemed to be written for some kind of

inheritance scenario but the scenario was unused for actual joined

inheritance, and the function instead seemed to perform some kind of

partial translation that would remove non-PK cols if the PK cols

happened to be first in the row, but not otherwise. The joined

inheritance walk feature here seems to have never been used as it was

always skipped by the "local_table" check.

As of 2.0.36 the function strips away non-PK cols and provides the

PK cols for the table in mapper PK order.

"""

try:

if mapper.local_table is not mapper.base_mapper.local_table:

# TODO: dive more into how a local table PK is used for fetch

# sync, not clear if this is correct as it depends on the

# downstream routine to fetch rows using

# local_table.primary_key order

pk_keys = result._tuple_getter(mapper.local_table.primary_key)

else:

pk_keys = result._tuple_getter(mapper.primary_key)

except KeyError:

# can't use these rows, they don't have PK cols in them

# this is an unusual case where the user would have used

# .return_defaults()

return []

return [pk_keys(row) for row in rows]

@classmethod

def _get_matched_objects_on_criteria(cls, update_options, states):

mapper = update_options._subject_mapper

eval_condition = update_options._eval_condition

raw_data = [

(state.obj(), state, state.dict)

for state in states

if state.mapper.isa(mapper) and not state.expired

]

identity_token = update_options._identity_token

if identity_token is not None:

raw_data = [

(obj, state, dict_)

for obj, state, dict_ in raw_data

if state.identity_token == identity_token

]

result = []

for obj, state, dict_ in raw_data:

evaled_condition = eval_condition(obj)

# caution: don't use "in ()" or == here, _EXPIRE_OBJECT

# evaluates as True for all comparisons

if (

evaled_condition is True

or evaled_condition is evaluator._EXPIRED_OBJECT

):

result.append(

(

obj,

state,

dict_,

evaled_condition is evaluator._EXPIRED_OBJECT,

)

)

return result

@classmethod

def _eval_condition_from_statement(cls, update_options, statement):

mapper = update_options._subject_mapper

target_cls = mapper.class_

evaluator_compiler = evaluator._EvaluatorCompiler(target_cls)

crit = ()

if statement._where_criteria:

crit += statement._where_criteria

global_attributes = {}

for opt in statement._with_options:

if opt._is_criteria_option:

opt.get_global_criteria(global_attributes)

if global_attributes:

crit += cls._adjust_for_extra_criteria(global_attributes, mapper)

if crit:

eval_condition = evaluator_compiler.process(*crit)

else:

# workaround for mypy https://github.com/python/mypy/issues/14027

def _eval_condition(obj):

return True

eval_condition = _eval_condition

return eval_condition

@classmethod

def _do_pre_synchronize_auto(