ensure sa_pool_key isn't passed to connect

2 files changed, 16 insertions, 0 deletions

diff --git a/lib/sqlalchemy/pool.py b/lib/sqlalchemy/pool.py
index 0bca5f599..0645c45b0 100644
--- a/lib/sqlalchemy/pool.py
+++ b/lib/sqlalchemy/pool.py
@@ -918,6 +918,7 @@ class _DBProxy(object):
             self._create_pool_mutex.acquire()
             try:
                 if key not in self.pools:
+                    kw.pop('sa_pool_key', None)
                     pool = self.poolclass(lambda: 
                                 self.module.connect(*args, **kw), **self.kw)
                     self.pools[key] = pool
diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index b515de0f8..789324445 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -68,6 +68,21 @@ class PoolTest(PoolTestBase):
         assert c4 is c6
         assert c4 is not c5
 
+    def test_manager_with_key(self):
+        class NoKws(object):
+            def connect(self, arg):
+                return MockConnection()
+
+        manager = pool.manage(NoKws(), use_threadlocal=True)
+
+        c1 = manager.connect('foo.db', sa_pool_key="a")
+        c2 = manager.connect('foo.db', sa_pool_key="b")
+        c3 = manager.connect('bar.db', sa_pool_key="a")
+
+        assert c1.cursor() is not None
+        assert c1 is not c2
+        assert c1 is  c3
+
     def test_bad_args(self):
         manager = pool.manage(MockDBAPI())
         connection = manager.connect(None)