def _get_resolved_values(cls, mapper, statement):
if statement._multi_values:
return []
- elif statement._ordered_values:
- return list(statement._ordered_values)
elif statement._values:
return list(statement._values.items())
else:
# are passed through to the new statement, which will then raise
# InvalidRequestError because UPDATE doesn't support multi_values
# right now.
- if statement._ordered_values:
- new_stmt._ordered_values = self._resolved_values
- elif statement._values:
+ if statement._values:
new_stmt._values = self._resolved_values
new_crit = self._adjust_for_extra_criteria(
UpdateDMLState.__init__(self, statement, compiler, **kw)
- if self._ordered_values:
+ if self._maintain_values_ordering:
raise sa_exc.InvalidRequestError(
"bulk ORM UPDATE does not support ordered_values() for "
"custom UPDATE statements with bulk parameter sets. Use a "
pks = mapper._pks_by_table[table]
- if use_orm_update_stmt is not None:
+ if (
+ use_orm_update_stmt is not None
+ and not use_orm_update_stmt._maintain_values_ordering
+ ):
# TODO: ordered values, etc
+ # ORM bulk_persistence will raise for the maintain_values_ordering
+ # case right now
value_params = use_orm_update_stmt._values
else:
value_params = {}
spd = mp[0]
stmt_parameter_tuples = list(spd.items())
spd_str_key = {_column_as_key(key) for key in spd}
- elif compile_state._ordered_values:
- spd = compile_state._dict_parameters
- stmt_parameter_tuples = compile_state._ordered_values
- assert spd is not None
- spd_str_key = {_column_as_key(key) for key in spd}
elif compile_state._dict_parameters:
spd = compile_state._dict_parameters
stmt_parameter_tuples = list(spd.items())
assert compile_state.isupdate or compile_state.isinsert
- if compile_state._parameter_ordering:
+ if compile_state._maintain_values_ordering:
parameter_ordering = [
- _column_as_key(key) for key in compile_state._parameter_ordering
+ _column_as_key(key) for key in compile_state._dict_parameters
]
ordered_keys = set(parameter_ordering)
cols = [
_multi_parameters: Optional[
List[MutableMapping[_DMLColumnElement, Any]]
] = None
- _ordered_values: Optional[List[Tuple[_DMLColumnElement, Any]]] = None
- _parameter_ordering: Optional[List[_DMLColumnElement]] = None
+ _maintain_values_ordering: bool = False
_primary_table: FromClause
_supports_implicit_returning = True
self.statement = statement
self.isupdate = True
- if statement._ordered_values is not None:
+ if statement._maintain_values_ordering:
self._process_ordered_values(statement)
elif statement._values is not None:
self._process_values(statement)
)
def _process_ordered_values(self, statement: ValuesBase) -> None:
- parameters = statement._ordered_values
-
+ parameters = statement._values
if self._no_parameters:
self._no_parameters = False
assert parameters is not None
self._dict_parameters = dict(parameters)
- self._ordered_values = parameters
- self._parameter_ordering = [key for key, value in parameters]
+ self._maintain_values_ordering = True
else:
raise exc.InvalidRequestError(
"Can only invoke ordered_values() once, and not mixed "
...,
] = ()
- _ordered_values: Optional[List[Tuple[_DMLColumnElement, Any]]] = None
+ _maintain_values_ordering: bool = False
_select_names: Optional[List[str]] = None
_inline: bool = False
@_generative
@_exclusive_against(
"_select_names",
- "_ordered_values",
+ "_maintain_values_ordering",
msgs={
"_select_names": "This construct already inserts from a SELECT",
- "_ordered_values": "This statement already has ordered "
+ "_maintain_values_ordering": "This statement already has ordered "
"values present",
},
+ defaults={"_maintain_values_ordering": False},
)
def values(
self,
("table", InternalTraversal.dp_clauseelement),
("_where_criteria", InternalTraversal.dp_clauseelement_tuple),
("_inline", InternalTraversal.dp_boolean),
- ("_ordered_values", InternalTraversal.dp_dml_ordered_values),
+ ("_maintain_values_ordering", InternalTraversal.dp_boolean),
("_values", InternalTraversal.dp_dml_values),
("_returning", InternalTraversal.dp_clauseelement_tuple),
("_hints", InternalTraversal.dp_table_hint_list),
def __init__(self, table: _DMLTableArgument):
super().__init__(table)
- @_generative
def ordered_values(self, *args: Tuple[_DMLColumnArgument, Any]) -> Self:
"""Specify the VALUES clause of this UPDATE statement with an explicit
parameter ordering that will be maintained in the SET clause of the
""" # noqa: E501
if self._values:
raise exc.ArgumentError(
- "This statement already has values present"
- )
- elif self._ordered_values:
- raise exc.ArgumentError(
- "This statement already has ordered values present"
+ "This statement already has "
+ f"{'ordered ' if self._maintain_values_ordering else ''}"
+ "values present"
)
- kv_generator = DMLState.get_plugin_class(self)._get_crud_kv_pairs
- self._ordered_values = kv_generator(self, args, True)
+ self = self.values(dict(args))
+ self._maintain_values_ordering = True
return self
@_generative
def do_orm_execute(bulk_ud):
cols = [
c.key
- for c, v in (
+ for c in (
(
bulk_ud.result.context
- ).compiled.compile_state.statement._ordered_values
+ ).compiled.compile_state.statement._values
)
]
m1(cols)
result = session.execute(stmt)
cols = [
c.key
- for c, v in (
- (
- result.context
- ).compiled.compile_state.statement._ordered_values
+ for c in (
+ (result.context).compiled.compile_state.statement._values
)
]
eq_(["age_int", "name"], cols)
result = session.execute(stmt)
cols = [
c.key
- for c, v in (
- result.context
- ).compiled.compile_state.statement._ordered_values
+ for c in (result.context).compiled.compile_state.statement._values
]
eq_(["name", "age_int"], cols)
from sqlalchemy.testing import eq_
from sqlalchemy.testing import expect_raises_message
from sqlalchemy.testing import fixtures
-from sqlalchemy.testing import mock
from sqlalchemy.testing.schema import Column
from sqlalchemy.testing.schema import Table
"UPDATE mytable SET foo(myid)=:param_1",
)
- @testing.fixture
- def randomized_param_order_update(self):
- from sqlalchemy.sql.dml import UpdateDMLState
-
- super_process_ordered_values = UpdateDMLState._process_ordered_values
-
- # this fixture is needed for Python 3.6 and above to work around
- # dictionaries being insert-ordered. in python 2.7 the previous
- # logic fails pretty easily without this fixture.
- def _process_ordered_values(self, statement):
- super_process_ordered_values(self, statement)
-
- tuples = list(self._dict_parameters.items())
- random.shuffle(tuples)
- self._dict_parameters = dict(tuples)
-
- dialect = default.StrCompileDialect()
- dialect.paramstyle = "qmark"
- dialect.positional = True
-
- with mock.patch.object(
- UpdateDMLState, "_process_ordered_values", _process_ordered_values
- ):
- yield
-
def random_update_order_parameters():
from sqlalchemy import ARRAY
)
@random_update_order_parameters()
- def test_update_to_expression_two(
- self, randomized_param_order_update, t, idx_to_value
- ):
+ def test_update_to_expression_two(self, t, idx_to_value):
"""test update from an expression.
this logic is triggered currently by a left side that doesn't
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
