Postpone parameters change in ordered updates

Postpone as much as possible the change of update parameters to
OrderedDict from list or tuple of pairs.

This way we won't have problems with query's update method.

5 files changed, 26 insertions, 28 deletions

diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index 373bd93fb..0a6ac98f6 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -1259,16 +1259,12 @@ class BulkUpdate(BulkUD):
 
     def _do_exec(self):
         if isinstance(self.values, (list, tuple)):
-            dict_type = util.OrderedDict
-            values = self.values
+            values = tuple((self._resolve_string_to_expr(k), v)
+                           for k, v in self.values)
         else:
-            dict_type = dict
-            values = self.values.items()
+            values = {self._resolve_string_to_expr(k): v
+                      for k, v in self.values.items()}
 
-        values = dict_type(
-            (self._resolve_string_to_expr(k), v)
-            for k, v in values
-        )
         update_stmt = sql.update(self.primary_table,
                                  self.context.whereclause, values,
                                  **self.update_kwargs)
diff --git a/lib/sqlalchemy/sql/crud.py b/lib/sqlalchemy/sql/crud.py
index 235889ad9..0ab52a13e 100644
--- a/lib/sqlalchemy/sql/crud.py
+++ b/lib/sqlalchemy/sql/crud.py
@@ -68,8 +68,13 @@ def _get_crud_params(compiler, stmt, **kw):
     # without isupdate check, but adding it shortcircuits the boolean operation
     # resulting in false for all inserts.
     keep_order = (compiler.isupdate
-                  and sql_util.is_value_pair_dict(stmt.parameters))
-    dict_type = util.OrderedDict if keep_order else dict
+                  and sql_util.is_value_pair_dict(stmt_parameters))
+    if keep_order:
+        stmt_parameters = util.OrderedDict(stmt_parameters)
+        dict_type = util.OrderedDict
+    else:
+        dict_type = dict
+
     # if we have statement parameters - set defaults in the
     # compiled params
     if compiler.column_keys is None:
diff --git a/lib/sqlalchemy/sql/dml.py b/lib/sqlalchemy/sql/dml.py
index c8407e3fd..dead61d9a 100644
--- a/lib/sqlalchemy/sql/dml.py
+++ b/lib/sqlalchemy/sql/dml.py
@@ -32,13 +32,9 @@ class UpdateBase(DialectKWArgs, HasPrefixes, Executable, ClauseElement):
 
     def _process_colparams(self, parameters):
         def process_single(p):
-            if isinstance(p, (list, tuple)):
-                if sql_util.is_value_pair_dict(p):
-                    return util.OrderedDict(p)
-                return dict(
-                    (c.key, pval)
-                    for c, pval in zip(self.table.c, p)
-                )
+            if (isinstance(p, (list, tuple)) and
+                    not sql_util.is_value_pair_dict(p)):
+                return {c.key: pval for c, pval in zip(self.table.c, p)}
             else:
                 return p
 
diff --git a/lib/sqlalchemy/sql/util.py b/lib/sqlalchemy/sql/util.py
index c73f710af..dedbdfe32 100644
--- a/lib/sqlalchemy/sql/util.py
+++ b/lib/sqlalchemy/sql/util.py
@@ -439,7 +439,7 @@ def criterion_as_pairs(expression, consider_as_foreign_keys=None,
 
 def is_value_pair_dict(params):
     """Check if params is a value list/tuple representing a dictionary."""
-    return (isinstance(params, (list, tuple)) and
+    return (isinstance(params, (list, tuple)) and len(params) > 0 and
             all(isinstance(p, (list, tuple)) and len(p) == 2 and
                 isinstance(p[0], schema.Column) for p in params))
 
diff --git a/test/orm/test_query.py b/test/orm/test_query.py
index 00d085898..eb77759dc 100644
--- a/test/orm/test_query.py
+++ b/test/orm/test_query.py
@@ -3855,8 +3855,9 @@ class SessionBindTest(QueryTest):
         with self._assert_bind_args(session) as mock_args:
             session.query(User).filter(User.id == 15).update(
                 {'name': 'foob', 'id': 123})
-            # Confirm that an unordered dict is being used
-            assert dict == type(mock_args.mock_calls[0][2]['clause'].parameters)
+            # Confirm that parameters are a dict instead of tuple or list
+            params_type = type(mock_args.mock_calls[0][2]['clause'].parameters)
+            assert params_type is dict
 
     def test_bulk_update_ordered_dict(self):
         User = self.classes.User
@@ -3866,29 +3867,29 @@ class SessionBindTest(QueryTest):
         # are unordered
         with self._assert_bind_args(session) as mock_args:
             session.query(User).filter(User.id == 15).update(
-                {'name': 'foob', 'id': 123})
-            assert dict == type(mock_args.mock_calls[0][2]['clause'].parameters)
+                util.OrderedDict((('name', 'foob'), ('id', 123))))
+            params_type = type(mock_args.mock_calls[0][2]['clause'].parameters)
+            assert params_type is dict
 
     def test_bulk_update_with_order(self):
         User = self.classes.User
         session = Session()
 
         # Do update using a tuple and check that order is preserved
-        # preserved from the dictionary but from the columns order
         with self._assert_bind_args(session) as mock_args:
             session.query(User).filter(User.id == 15).update(
                 (('id', 123), ('name', 'foob')))
-            assert isinstance(
-                mock_args.mock_calls[0][2]['clause'].parameters,
-                dict)
+            cols = [c[0].name for c
+                    in mock_args.mock_calls[0][2]['clause'].parameters]
+            assert ['id', 'name'] == cols
 
         # Now invert the order and use a list instead, and check that order is
         # also preserved
         with self._assert_bind_args(session) as mock_args:
             session.query(User).filter(User.id == 15).update(
                 [('id', 123), ('name', 'foob')])
-            cols = [c.name for c
-                    in mock_args.mock_calls[0][2]['clause'].parameters.keys()]
+            cols = [c[0].name for c
+                    in mock_args.mock_calls[0][2]['clause'].parameters]
             assert ['id', 'name'] == cols
 
     def test_bulk_delete_no_sync(self):