Random tokens cause memory leaks, perhaps consider using futures?

[whoward]

Hello,
I was looking for a library that allowed for a threadsafe single connection to sqlite for a side project I've been working on and came across your project here. I like the functionality available but was curious about the internals when I noticed the execute / fetch_results pattern which generates a token and uses that token to fetch the result set when it eventually executes.

On a hunch I asked AI to perform a code review on the file and from the result it does indeed look like generated tokens are retained forever and never get expunged from _select_events which means that memory will continue to grow and will never shrink. The prompt I used was:

do a code review of https://raw.githubusercontent.com/roshanlam/sqlite-worker/refs/heads/main/sqlite_worker/main.py
 
i'd like you to analyze in particular for memory leaks and thread safety issues

You should be able to run this and get similar results

I'm curious if you've ever considered an approach based on the Future object? As a result of using futures you sidestep any memory leaking due to the tokens approach. Here's an example implementation that even implements full async functionality:

import asyncio
import queue
import sqlite3
import threading
from typing import Any, Optional

class AsyncSqliteWorker:
    """Minimal async reimplementation of the original SqliteWorker.
    
    Only execute() + execution thread. No tokens, no leaks.
    """

    def __init__(self, db_path: str, max_queue_size: int = 100):
        self.db_path = db_path
        self._queue: queue.Queue = queue.Queue(maxsize=max_queue_size)
        self._loop: Optional[asyncio.AbstractEventLoop] = None
        self._close_event = threading.Event()

        # Start the execution thread (exactly like the original)
        self._thread = threading.Thread(target=self._worker, daemon=True)
        self._thread.start()

    def _is_select_query(self, query: str) -> bool:
        """Same detection logic as the original library."""
        return query.lower().lstrip().startswith("select")

    def _worker(self):
        """The original execution thread — now talks to asyncio futures."""
        conn = sqlite3.connect(
            self.db_path,
            check_same_thread=False,
            detect_types=sqlite3.PARSE_DECLTYPES,
            timeout=30.0,
        )
        try:
            while not self._close_event.is_set():
                try:
                    task = self._queue.get(timeout=1.0)
                except queue.Empty:
                    continue

                if task is None:        # shutdown sentinel
                    break

                future: asyncio.Future
                query: str
                values: list

                future, query, values = task

                try:
                    cursor = conn.cursor()
                    cursor.execute(query, values or [])

                    # Return sensible result (unlike original that always did fetchall)
                    if self._is_select_query(query):
                        result: Any = cursor.fetchall()
                    else:
                        result = cursor.rowcount
                        conn.commit()               # simple auto-commit for DML

                    # Safe cross-thread result delivery
                    if self._loop is not None:
                        self._loop.call_soon_threadsafe(
                            lambda f, r: f.set_result(r) if not f.done() else None,
                            future, result
                        )
                except Exception as exc:
                    if self._loop is not None:
                        self._loop.call_soon_threadsafe(
                            lambda f, e: f.set_exception(e) if not f.done() else None,
                            future, exc
                        )
        finally:
            conn.close()

    async def execute(self, query: str, values: Optional[list] = None) -> Any:
        """Fully async version of the original execute().
        
        - No token
        - No fetch_results()
        - Just await it → get rows or rowcount (or exception)
        """
        if self._close_event.is_set():
            raise RuntimeError("Worker is closed")

        # Capture the running loop on first call (safe pattern)
        if self._loop is None:
            self._loop = asyncio.get_running_loop()

        future: asyncio.Future = self._loop.create_future()

        # Same queue pattern as the original library
        self._queue.put((future, query, values or []), timeout=5)

        return await future

    def close(self):
        """Simple shutdown (for completeness in a POC)."""
        self._close_event.set()
        self._queue.put(None, timeout=5)   # sentinel
        self._thread.join(timeout=5)


# ====================== Tiny usage example ======================
async def demo():
    db = AsyncSqliteWorker("test.db")

    # CREATE
    await db.execute("CREATE TABLE IF NOT EXISTS test (id INTEGER PRIMARY KEY, name TEXT)")

    # INSERT
    await db.execute("INSERT INTO test (name) VALUES (?)", ["hello async"])

    # SELECT
    rows = await db.execute("SELECT * FROM test")
    print("Rows:", rows)

    db.close()

if __name__ == "__main__":
    asyncio.run(demo())