add template methods for ORMInsert

Fixed regression caused by :ticket:`7861` where invoking an
:class:`.Insert` construct which contained ORM entities via
:meth:`_orm.Session.execute` would fail.

Fixes: #7878
Change-Id: Icc4d8028249cc417f504fdd3e31e206b5bbc89f8

3 files changed, 62 insertions, 1 deletions

diff --git a/doc/build/changelog/unreleased_14/7878.rst b/doc/build/changelog/unreleased_14/7878.rst
new file mode 100644
index 000000000..6c9e92929
--- /dev/null
+++ b/doc/build/changelog/unreleased_14/7878.rst
@@ -0,0 +1,7 @@
+.. change::
+    :tags: bug, orm, regression
+    :tickets: 7878
+
+    Fixed regression caused by :ticket:`7861` where invoking an
+    :class:`.Insert` construct which contained ORM entities directly via
+    :meth:`_orm.Session.execute` would fail.
diff --git a/lib/sqlalchemy/orm/persistence.py b/lib/sqlalchemy/orm/persistence.py
index f2cddad53..355ddc922 100644
--- a/lib/sqlalchemy/orm/persistence.py
+++ b/lib/sqlalchemy/orm/persistence.py
@@ -2182,7 +2182,32 @@ class ORMDMLState:
 
 @CompileState.plugin_for("orm", "insert")
 class ORMInsert(ORMDMLState, InsertDMLState):
-    pass
+    @classmethod
+    def orm_pre_session_exec(
+        cls,
+        session,
+        statement,
+        params,
+        execution_options,
+        bind_arguments,
+        is_reentrant_invoke,
+    ):
+        return (
+            statement,
+            util.immutabledict(execution_options),
+        )
+
+    @classmethod
+    def orm_setup_cursor_result(
+        cls,
+        session,
+        statement,
+        params,
+        execution_options,
+        bind_arguments,
+        result,
+    ):
+        return result
 
 
 @CompileState.plugin_for("orm", "update")
diff --git a/test/orm/test_session.py b/test/orm/test_session.py
index 56371bc68..814758936 100644
--- a/test/orm/test_session.py
+++ b/test/orm/test_session.py
@@ -2,8 +2,10 @@ import inspect as _py_inspect
 import pickle
 
 import sqlalchemy as sa
+from sqlalchemy import delete
 from sqlalchemy import event
 from sqlalchemy import ForeignKey
+from sqlalchemy import insert
 from sqlalchemy import inspect
 from sqlalchemy import Integer
 from sqlalchemy import select
@@ -11,6 +13,7 @@ from sqlalchemy import Sequence
 from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy import text
+from sqlalchemy import update
 from sqlalchemy.orm import attributes
 from sqlalchemy.orm import backref
 from sqlalchemy.orm import close_all_sessions
@@ -2215,6 +2218,32 @@ class NewStyleExecutionTest(_fixtures.FixtureTest):
         ):
             result.all()
 
+    @testing.combinations("insert", "update", "delete", argnames="dml_expr")
+    @testing.combinations("core", "orm", argnames="coreorm")
+    def test_dml_execute(self, dml_expr, coreorm):
+        User = self.classes.User
+        users = self.tables.users
+
+        sess = fixture_session()
+
+        if coreorm == "orm":
+            if dml_expr == "insert":
+                stmt = insert(User).values(id=12, name="some user")
+            elif dml_expr == "update":
+                stmt = update(User).values(name="sone name").filter_by(id=15)
+            else:
+                stmt = delete(User).filter_by(id=15)
+        else:
+            if dml_expr == "insert":
+                stmt = insert(users).values(id=12, name="some user")
+            elif dml_expr == "update":
+                stmt = update(users).values(name="sone name").filter_by(id=15)
+            else:
+                stmt = delete(users).filter_by(id=15)
+
+        result = sess.execute(stmt)
+        result.close()
+
     @testing.combinations((True,), (False,), argnames="prebuffered")
     @testing.combinations(("close",), ("expunge_all",), argnames="meth")
     def test_unbuffered_result_before_session_is_closed(