Added loader options to session.merge, asyncsession.merge

Added loader options to :meth:`_orm.Session.merge` and
:meth:`_asyncio.AsyncSession.merge`, which will apply the given loader
options to the ``get()`` used internally by merge, allowing eager loading
of relationships etc. to be applied when the merge process loads a new
object. Pull request courtesy Daniel Stone.

Fixes: #6955
Closes: #6957
Pull-request: https://github.com/sqlalchemy/sqlalchemy/pull/6957
Pull-request-sha: ab4d96cd5da9a5dd01112b8dcd6514db64aa8d9f

Change-Id: I5b94dfda1088a8bc6396e9fd9a072827df1f8680

1 files changed, 33 insertions, 0 deletions

diff --git a/test/ext/asyncio/test_session_py3k.py b/test/ext/asyncio/test_session_py3k.py
index 4165991d4..cd9054740 100644
--- a/test/ext/asyncio/test_session_py3k.py
+++ b/test/ext/asyncio/test_session_py3k.py
@@ -107,6 +107,17 @@ class AsyncSessionQueryTest(AsyncFixture):
         is_(u3, None)
 
     @async_test
+    async def test_get_loader_options(self, async_session):
+        User = self.classes.User
+
+        u = await async_session.get(
+            User, 7, options=[selectinload(User.addresses)]
+        )
+
+        eq_(u.name, "jack")
+        eq_(len(u.__dict__["addresses"]), 1)
+
+    @async_test
     @testing.requires.independent_cursors
     @testing.combinations(
         {}, dict(execution_options={"logging_token": "test"}), argnames="kw"
@@ -336,6 +347,28 @@ class AsyncSessionTransactionTest(AsyncFixture):
             eq_(u1.name, "new u1")
 
     @async_test
+    async def test_merge_loader_options(self, async_session):
+        User = self.classes.User
+        Address = self.classes.Address
+
+        async with async_session.begin():
+            u1 = User(id=1, name="u1", addresses=[Address(email_address="e1")])
+
+            async_session.add(u1)
+
+        await async_session.close()
+
+        async with async_session.begin():
+            new_u1 = User(id=1, name="new u1")
+
+            new_u_merged = await async_session.merge(
+                new_u1, options=[selectinload(User.addresses)]
+            )
+
+            eq_(new_u_merged.name, "new u1")
+            eq_(len(new_u_merged.__dict__["addresses"]), 1)
+
+    @async_test
     async def test_join_to_external_transaction(self, async_engine):
         User = self.classes.User