'await connection.close()' returns once all transaction queue items's results have been forwarded, including the _STOP_RUNNING_SENTINEL result

Author: davidandreoletti

aiosqlite/core.py

@@ -72,10 +72,17 @@ def __init__(
                 DeprecationWarning,
             )
 
-    def _stop_running(self):
+    async def _stop_running(self):
         self._running = False
-        # PEP 661 is not accepted yet, so we cannot type a sentinel
-        self._tx.put_nowait(_STOP_RUNNING_SENTINEL)  # type: ignore[arg-type]
+
+        function = partial(lambda :_STOP_RUNNING_SENTINEL)
+        future = asyncio.get_event_loop().create_future()
+
+        self._tx.put_nowait((future, function))
+
+        return await future
+
+
 
     @property
     def _conn(self) -> sqlite3.Connection:
@@ -105,16 +112,16 @@ def run(self) -> None:
             # futures)
 
             tx_item = self._tx.get()
-            if tx_item is _STOP_RUNNING_SENTINEL:
-                break
-
             future, function = tx_item
 
             try:
                 LOG.debug("executing %s", function)
                 result = function()
                 LOG.debug("operation %s completed", function)
                 future.get_loop().call_soon_threadsafe(set_result, future, result)
+
+                if result is _STOP_RUNNING_SENTINEL:
+                    break
             except BaseException as e:  # noqa B036
                 LOG.debug("returning exception %s", e)
                 future.get_loop().call_soon_threadsafe(set_exception, future, e)
@@ -139,7 +146,7 @@ async def _connect(self) -> "Connection":
                 self._tx.put_nowait((future, self._connector))
                 self._connection = await future
             except Exception:
-                self._stop_running()
+                await self._stop_running()
                 self._connection = None
                 raise
 
@@ -180,7 +187,7 @@ async def close(self) -> None:
             LOG.info("exception occurred while closing connection")
             raise
         finally:
-            self._stop_running()
+            await self._stop_running()
             self._connection = None
 
     @contextmanager