Merge pull request #909 from procrastinate-org/aiopg-broken

procrastinate/contrib/aiopg/aiopg_connector.py

@@ -143,20 +143,20 @@ def __init__(
         self._pool_externally_set: bool = False
         self.json_dumps = json_dumps
         self.json_loads = json_loads
+        self._original_kwargs = kwargs
         self._pool_args = self._adapt_pool_args(kwargs, json_loads)
         self._lock: asyncio.Lock | None = None
         self._sync_connector: connector.BaseConnector | None = None
 
     def get_sync_connector(self) -> connector.BaseConnector:
         if self._pool:
             return self
-        args_copy = dict(self._pool_args)
-        args_copy.pop("on_connect", None)
+
         if self._sync_connector is None:
             self._sync_connector = psycopg2_connector.Psycopg2Connector(
                 json_dumps=self.json_dumps,
                 json_loads=self.json_loads,
-                **args_copy,
+                **dict(self._original_kwargs),
             )
         return self._sync_connector
 

tests/integration/contrib/aiopg/test_aiopg_connector.py

@@ -4,6 +4,7 @@
 import functools
 import json
 
+import asgiref.sync
 import attr
 import pytest
 
@@ -14,15 +15,16 @@
 async def aiopg_connector_factory(connection_params):
     connectors = []
 
-    async def _(**kwargs):
+    async def _(*, open: bool = True, **kwargs):
         json_dumps = kwargs.pop("json_dumps", None)
         json_loads = kwargs.pop("json_loads", None)
         connection_params.update(kwargs)
         connector = aiopg.AiopgConnector(
             json_dumps=json_dumps, json_loads=json_loads, **connection_params
         )
         connectors.append(connector)
-        await connector.open_async()
+        if open:
+            await connector.open_async()
         return connector
 
     yield _
@@ -119,6 +121,24 @@ async def test_execute_query(aiopg_connector):
     assert result == [{"obj_description": "foo"}]
 
 
+async def test_get_sync_connector(aiopg_connector_factory):
+    result = []
+
+    aiopg_connector = await aiopg_connector_factory(open=False)
+
+    @asgiref.sync.sync_to_async
+    def f():
+        sync_conn = aiopg_connector.get_sync_connector()
+        sync_conn.open()
+        try:
+            result.append(sync_conn.execute_query_one("SELECT 1"))
+        finally:
+            sync_conn.close()
+
+    await f()
+    assert list(result[0].values()) == [1]
+
+
 async def test_execute_query_interpolate(aiopg_connector):
     result = await aiopg_connector.execute_query_one_async(
         "SELECT %(foo)s as foo;", foo="bar"