support parameters in all ORM insert modes

Fixed 2.0 regression where use of :func:`_sql.bindparam()` inside of
:meth:`_dml.Insert.values` would fail to be interpreted correctly when
executing the :class:`_dml.Insert` statement using the ORM
:class:`_orm.Session`, due to the new ORM-enabled insert feature not
implementing this use case.

In addition, the bulk INSERT and UPDATE features now add these
capabilities:

* The requirement that extra parameters aren't passed when using ORM
INSERT using the "orm" dml_strategy setting is lifted.
* The requirement that additional WHERE criteria is not passed when using
ORM UPDATE using the "bulk" dml_strategy setting is lifted.  Note that
in this case, the check for expected row count is turned off.

Fixes: #9583
Change-Id: I539c18893b697caeab5a5f0195a27d4f0487e728

1 files changed, 23 insertions, 8 deletions

diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index 1af55df00..6fa338ced 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -326,9 +326,11 @@ def _organize_states_for_delete(base_mapper, states, uowtransaction):
 def _collect_insert_commands(
     table,
     states_to_insert,
+    *,
     bulk=False,
     return_defaults=False,
     render_nulls=False,
+    include_bulk_keys=(),
 ):
     """Identify sets of values to use in INSERT statements for a
     list of states.
@@ -401,10 +403,14 @@ def _collect_insert_commands(
                 None
             )
 
-        if bulk and mapper._set_polymorphic_identity:
-            params.setdefault(
-                mapper._polymorphic_attr_key, mapper.polymorphic_identity
-            )
+        if bulk:
+            if mapper._set_polymorphic_identity:
+                params.setdefault(
+                    mapper._polymorphic_attr_key, mapper.polymorphic_identity
+                )
+
+            if include_bulk_keys:
+                params.update((k, state_dict[k]) for k in include_bulk_keys)
 
         yield (
             state,
@@ -422,8 +428,10 @@ def _collect_update_commands(
     uowtransaction,
     table,
     states_to_update,
+    *,
     bulk=False,
     use_orm_update_stmt=None,
+    include_bulk_keys=(),
 ):
     """Identify sets of values to use in UPDATE statements for a
     list of states.
@@ -581,6 +589,9 @@ def _collect_update_commands(
                         "key value on column %s" % (table, col)
                     )
 
+        if include_bulk_keys:
+            params.update((k, state_dict[k]) for k in include_bulk_keys)
+
         if params or value_params:
             params.update(pk_params)
             yield (
@@ -712,8 +723,10 @@ def _emit_update_statements(
     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()."""
@@ -847,10 +860,10 @@ def _emit_update_statements(
                         c.returned_defaults,
                     )
                 rows += c.rowcount
-                check_rowcount = assert_singlerow
+                check_rowcount = enable_check_rowcount and assert_singlerow
         else:
             if not allow_executemany:
-                check_rowcount = assert_singlerow
+                check_rowcount = enable_check_rowcount and assert_singlerow
                 for (
                     state,
                     state_dict,
@@ -883,8 +896,9 @@ def _emit_update_statements(
             else:
                 multiparams = [rec[2] for rec in records]
 
-                check_rowcount = assert_multirow or (
-                    assert_singlerow and len(multiparams) == 1
+                check_rowcount = enable_check_rowcount and (
+                    assert_multirow
+                    or (assert_singlerow and len(multiparams) == 1)
                 )
 
                 c = connection.execute(
@@ -941,6 +955,7 @@ def _emit_insert_statements(
     mapper,
     table,
     insert,
+    *,
     bookkeeping=True,
     use_orm_insert_stmt=None,
     execution_options=None,