From ee2c3b8a31fc0dc25fb6a434268de846d4999fd0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mat=C4=9Bj=20Cepl?= Date: Sun, 27 Sep 2026 20:25:00 +0200 Subject: [PATCH 1/3] autoimport: track and close SQLite connections across worker threads AutoImport dynamically allocates per-thread SQLite connections using threading.local(). However, AutoImport.close() only closed the connection belonging to the calling thread, orphaning any connections created by worker threads and triggering ResourceWarning on Python 3.12+. Keep a thread-safe registry of all active connections on the AutoImport instance, create connections with check_same_thread=False, and ensure AutoImport.close() commits and closes all registered connections. Also expose close_thread_connection() for worker threads and implement context manager protocol support on AutoImport. Fixes: https://github.com/python-rope/rope/issues/886 --- CHANGELOG.md | 1 + rope/contrib/autoimport/sqlite.py | 92 ++++++++++++++++-- ropetest/contrib/autoimport/autoimporttest.py | 93 ++++++++++++++++--- ropetest/contrib/autoimport/modeltest.py | 4 +- ropetest/contrib/autoimporttest.py | 14 +-- 5 files changed, 174 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ba328bd9a..58fdb09d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ # **Upcoming release** +- #886 Track and close SQLite connections created in worker threads in AutoImport (@mcepl) - #895 patchedast Starred and keyword now consumes their syntactically expected and ** (@lieryan) - #896 patchedast cleanup and refactoring (@lieryan) - #905 Ignore `node_modules` directories by default when scanning project files and source folders. (@marcoeangeli) diff --git a/rope/contrib/autoimport/sqlite.py b/rope/contrib/autoimport/sqlite.py index 7ae8fb718..524f29f45 100644 --- a/rope/contrib/autoimport/sqlite.py +++ b/rope/contrib/autoimport/sqlite.py @@ -15,7 +15,7 @@ from hashlib import sha256 from itertools import chain from pathlib import Path -from threading import local +from threading import Lock, local from typing import ( TYPE_CHECKING, Generator, @@ -142,6 +142,10 @@ def __init__( "`AutoImport(memory=True)` explicitly.", DeprecationWarning, ) + self._closed = False + self._connections: Set[sqlite3.Connection] = set() + self._connections_lock = Lock() + self._observer: Optional[resourceobserver.ResourceObserver] = None self.thread_local = local() self.connection = self.create_database_connection( project=project, @@ -149,10 +153,10 @@ def __init__( ) self._setup_db() if observe: - observer = resourceobserver.ResourceObserver( + self._observer = resourceobserver.ResourceObserver( changed=self._changed, moved=self._moved, removed=self._removed ) - project.add_observer(observer) + project.add_observer(self._observer) @classmethod def create_database_connection( @@ -186,10 +190,25 @@ def calculate_project_hash(data: str) -> str: else: project_hash = calculate_project_hash(project.ropefolder.real_path) return sqlite3.connect( - f"file:rope-{project_hash}:?mode=memory&cache=shared", uri=True + f"file:rope-{project_hash}:?mode=memory&cache=shared", + uri=True, + check_same_thread=False, ) else: - return sqlite3.connect(project.ropefolder.pathlib / "autoimport.db") + return sqlite3.connect( + project.ropefolder.pathlib / "autoimport.db", + check_same_thread=False, + ) + + def _register_connection(self, conn: sqlite3.Connection) -> None: + with self._connections_lock: + if self._closed: + with contextlib.suppress( + sqlite3.ProgrammingError, sqlite3.OperationalError + ): + conn.close() + raise exceptions.RopeError("AutoImport instance has been closed") + self._connections.add(conn) @property def connection(self) -> sqlite3.Connection: @@ -198,15 +217,26 @@ def connection(self) -> sqlite3.Connection: This makes sure AutoImport can be shared across threads. """ + if self._closed: + raise exceptions.RopeError("AutoImport instance has been closed") if not hasattr(self.thread_local, "connection"): - self.thread_local.connection = self.create_database_connection( + conn = self.create_database_connection( project=self.project, memory=self.memory, ) + self._register_connection(conn) + self.thread_local.connection = conn return self.thread_local.connection @connection.setter def connection(self, value: sqlite3.Connection): + if self._closed: + raise exceptions.RopeError("AutoImport instance has been closed") + old_conn = getattr(self.thread_local, "connection", None) + if old_conn is not None and old_conn is not value: + with self._connections_lock: + self._connections.discard(old_conn) + self._register_connection(value) self.thread_local.connection = value def _setup_db(self): @@ -457,10 +487,56 @@ def update_module(self, module: str): self._del_if_exist(module) self.generate_modules_cache([module]) + def close_thread_connection(self): + """Close the SQLite connection for the current thread.""" + conn = getattr(self.thread_local, "connection", None) + if conn is not None: + with self._connections_lock: + self._connections.discard(conn) + with contextlib.suppress(AttributeError): + del self.thread_local.connection + with contextlib.suppress( + sqlite3.ProgrammingError, sqlite3.OperationalError + ): + conn.commit() + with contextlib.suppress( + sqlite3.ProgrammingError, sqlite3.OperationalError + ): + conn.close() + def close(self): """Close the autoimport database.""" - self.connection.commit() - self.connection.close() + with self._connections_lock: + if self._closed: + return + self._closed = True + connections = list(self._connections) + self._connections.clear() + + if self._observer is not None: + with contextlib.suppress(Exception): + self.project.remove_observer(self._observer) + self._observer = None + + for conn in connections: + with contextlib.suppress( + sqlite3.ProgrammingError, sqlite3.OperationalError + ): + conn.commit() + with contextlib.suppress( + sqlite3.ProgrammingError, sqlite3.OperationalError + ): + conn.close() + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + self.close() + + def __del__(self): + with contextlib.suppress(Exception): + self.close() def get_name_locations(self, name): """Return a list of ``(resource, lineno)`` tuples.""" diff --git a/ropetest/contrib/autoimport/autoimporttest.py b/ropetest/contrib/autoimport/autoimporttest.py index 072ab1ed8..64376e83f 100644 --- a/ropetest/contrib/autoimport/autoimporttest.py +++ b/ropetest/contrib/autoimport/autoimporttest.py @@ -6,6 +6,7 @@ import pytest +from rope.base import exceptions from rope.base.project import Project from rope.base.resources import File, Folder from rope.contrib.autoimport import models @@ -46,16 +47,16 @@ def test_autoimport_connection_parameter_with_in_memory( project: Project, autoimport: AutoImport, ): - connection = AutoImport.create_database_connection(memory=True) - assert is_in_memory_database(connection) + with closing(AutoImport.create_database_connection(memory=True)) as connection: + assert is_in_memory_database(connection) def test_autoimport_connection_parameter_with_project( project: Project, autoimport: AutoImport, ): - connection = AutoImport.create_database_connection(project=project) - assert not is_in_memory_database(connection) + with closing(AutoImport.create_database_connection(project=project)) as connection: + assert not is_in_memory_database(connection) def test_autoimport_create_database_connection_conflicting_parameter( @@ -102,25 +103,87 @@ def foo(): def test_multithreading( - autoimport: AutoImport, project: Project, pkg1: Folder, mod1: File, ): mod1_init = pkg1.get_child("__init__.py") - mod1_init.write(dedent("""\ + mod1_init.write( + dedent("""\ def foo(): pass - """)) - mod1.write(dedent("""\ + """) + ) + mod1.write( + dedent("""\ foo - """)) - autoimport = AutoImport(project, memory=False) - autoimport.generate_cache([mod1_init]) - - tp = ThreadPoolExecutor(1) - results = tp.submit(autoimport.search, "foo", True).result() - assert [("from pkg1 import foo", "foo")] == results + """) + ) + with closing(AutoImport(project, memory=False)) as autoimport: + autoimport.generate_cache([mod1_init]) + + with ThreadPoolExecutor(1) as tp: + results = tp.submit(autoimport.search, "foo", True).result() + assert [("from pkg1 import foo", "foo")] == results + + +def test_multithread_connections_closed_on_close(project: Project): + with AutoImport(project, memory=True) as ai: + main_conn = ai.connection + worker_conns = [] + + def worker(): + conn = ai.connection + worker_conns.append(conn) + return list(ai.search("foo")) + + with ThreadPoolExecutor(3) as tp: + futures = [tp.submit(worker) for _ in range(3)] + for f in futures: + f.result() + + all_conns = {main_conn} | set(worker_conns) + assert len(all_conns) > 1 + assert all_conns.issubset(ai._connections) + + assert ai._closed + assert len(ai._connections) == 0 + for conn in all_conns: + with pytest.raises(sqlite3.ProgrammingError, match="Cannot operate on a closed database"): + conn.execute("SELECT 1") + + with pytest.raises(exceptions.RopeError, match="AutoImport instance has been closed"): + _ = ai.connection + + +def test_close_thread_connection(project: Project): + with AutoImport(project, memory=True) as ai: + worker_conn = None + + def worker(): + nonlocal worker_conn + worker_conn = ai.connection + assert worker_conn in ai._connections + ai.close_thread_connection() + assert worker_conn not in ai._connections + with pytest.raises(sqlite3.ProgrammingError, match="Cannot operate on a closed database"): + worker_conn.execute("SELECT 1") + new_conn = ai.connection + assert new_conn is not worker_conn + assert new_conn in ai._connections + + with ThreadPoolExecutor(1) as tp: + tp.submit(worker).result() + + +def test_close_idempotent(project: Project): + ai = AutoImport(project, memory=True) + conn = ai.connection + ai.close() + assert ai._closed + ai.close() + with pytest.raises(sqlite3.ProgrammingError, match="Cannot operate on a closed database"): + conn.execute("SELECT 1") def test_connection(project: Project, project2: Project): diff --git a/ropetest/contrib/autoimport/modeltest.py b/ropetest/contrib/autoimport/modeltest.py index 0044434cc..8eef06d35 100644 --- a/ropetest/contrib/autoimport/modeltest.py +++ b/ropetest/contrib/autoimport/modeltest.py @@ -8,7 +8,9 @@ @pytest.fixture def empty_db(): - return sqlite3.connect(":memory:") + conn = sqlite3.connect(":memory:") + yield conn + conn.close() class TestQuery: diff --git a/ropetest/contrib/autoimporttest.py b/ropetest/contrib/autoimporttest.py index d42ec14f7..ebf21af56 100644 --- a/ropetest/contrib/autoimporttest.py +++ b/ropetest/contrib/autoimporttest.py @@ -15,6 +15,7 @@ def setUp(self): self.importer = autoimport.AutoImport(self.project, observe=False) def tearDown(self): + self.importer.close() testutils.remove_project(self.project) super().tearDown() @@ -178,12 +179,12 @@ def test_skipping_directories_not_accessible_because_of_permission_error(self): def test_search_submodule(project, external_fixturepkg): - importer = autoimport.AutoImport(project, observe=False) - importer.update_module("external_fixturepkg") - import_statement = ("from external_fixturepkg import mod1", "mod1") - assert import_statement in importer.search("mod1", exact_match=True) - assert import_statement in importer.search("mo") - assert import_statement in importer.search("mod1") + with autoimport.AutoImport(project, observe=False) as importer: + importer.update_module("external_fixturepkg") + import_statement = ("from external_fixturepkg import mod1", "mod1") + assert import_statement in importer.search("mod1", exact_match=True) + assert import_statement in importer.search("mo") + assert import_statement in importer.search("mod1") class AutoImportObservingTest(unittest.TestCase): @@ -196,6 +197,7 @@ def setUp(self): self.importer = autoimport.AutoImport(self.project, observe=True) def tearDown(self): + self.importer.close() testutils.remove_project(self.project) super().tearDown() From 3867ebb1beeb05ea33c92a8549564722e024d09e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mat=C4=9Bj=20Cepl?= Date: Mon, 28 Sep 2026 18:25:46 +0200 Subject: [PATCH 2/3] autoimport: cover the remaining connection-tracking branches Codecov reported 90% patch coverage for PR #889 with 6 uncovered new lines in rope/contrib/autoimport/sqlite.py: - the guard branch in _register_connection() that closes the new connection and raises when the instance is already closed - the closed-instance guard in the `connection` setter - the setter path discarding the previously tracked thread-local connection when replaced by a different one Add tests hitting each of these paths so the patch coverage is complete. --- ropetest/contrib/autoimport/autoimporttest.py | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/ropetest/contrib/autoimport/autoimporttest.py b/ropetest/contrib/autoimport/autoimporttest.py index 64376e83f..b813af3a0 100644 --- a/ropetest/contrib/autoimport/autoimporttest.py +++ b/ropetest/contrib/autoimport/autoimporttest.py @@ -186,6 +186,36 @@ def test_close_idempotent(project: Project): conn.execute("SELECT 1") +def test_register_connection_after_close(project: Project): + ai = AutoImport(project, memory=True) + ai.close() + conn = AutoImport.create_database_connection(memory=True) + with pytest.raises(exceptions.RopeError, match="AutoImport instance has been closed"): + ai._register_connection(conn) + assert conn not in ai._connections + with pytest.raises(sqlite3.ProgrammingError, match="Cannot operate on a closed database"): + conn.execute("SELECT 1") + + +def test_connection_setter_after_close(project: Project): + ai = AutoImport(project, memory=True) + ai.close() + with closing(AutoImport.create_database_connection(memory=True)) as conn: + with pytest.raises(exceptions.RopeError, match="AutoImport instance has been closed"): + ai.connection = conn + + +def test_connection_setter_replaces_existing(project: Project): + with AutoImport(project, memory=True) as ai: + old_conn = ai.connection + assert old_conn in ai._connections + with closing(AutoImport.create_database_connection(memory=True)) as conn: + ai.connection = conn + assert ai.connection is conn + assert conn in ai._connections + assert old_conn not in ai._connections + + def test_connection(project: Project, project2: Project): ai1 = AutoImport(project) ai2 = AutoImport(project) From 130619278f369c830e92e4d75e7aa7359492d98c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mat=C4=9Bj=20Cepl?= Date: Mon, 9 Nov 2026 11:27:32 +0100 Subject: [PATCH 3/3] autoimport: reap connections from finished threads Associate each connection with its owner thread and close it only after that thread exits. Reject AutoImport.close() while another live thread owns a connection, and document the shutdown requirement. Add tests for worker connection reaping and refused concurrent close. --- docs/library.rst | 7 +- rope/contrib/autoimport/sqlite.py | 98 +++++++++++-------- ropetest/contrib/autoimport/autoimporttest.py | 67 +++++++++++-- 3 files changed, 124 insertions(+), 48 deletions(-) diff --git a/docs/library.rst b/docs/library.rst index 35d497337..b8a1f9d64 100644 --- a/docs/library.rst +++ b/docs/library.rst @@ -853,7 +853,12 @@ by passing memory=False to AutoImport constructor. This default will change in the future, if you want to always store the autoimport cache in-memory, then you need to explicitly pass memory=True. -It must be closed when done with the ``AutoImport.close()`` method. +It must be closed when done with the ``AutoImport.close()`` method. When +sharing an instance across threads, wait for worker threads to finish before +calling ``close()``. It raises ``RopeError`` rather than closing a connection +owned by a live worker thread. A worker can close its own connection sooner +with ``close_thread_connection()``. Connections left by finished workers are +closed when another thread creates a connection or when ``close()`` runs. AutoImport can search for a name from both modules and statements you can import from them. diff --git a/rope/contrib/autoimport/sqlite.py b/rope/contrib/autoimport/sqlite.py index 524f29f45..d8ef55a73 100644 --- a/rope/contrib/autoimport/sqlite.py +++ b/rope/contrib/autoimport/sqlite.py @@ -15,9 +15,10 @@ from hashlib import sha256 from itertools import chain from pathlib import Path -from threading import Lock, local +from threading import Lock, Thread, current_thread, local from typing import ( TYPE_CHECKING, + Dict, Generator, Iterable, Iterator, @@ -143,7 +144,7 @@ def __init__( DeprecationWarning, ) self._closed = False - self._connections: Set[sqlite3.Connection] = set() + self._connections: Dict[Thread, sqlite3.Connection] = {} self._connections_lock = Lock() self._observer: Optional[resourceobserver.ResourceObserver] = None self.thread_local = local() @@ -200,15 +201,31 @@ def calculate_project_hash(data: str) -> str: check_same_thread=False, ) + @staticmethod + def _commit_and_close(conn: sqlite3.Connection) -> None: + with contextlib.suppress(sqlite3.ProgrammingError, sqlite3.OperationalError): + conn.commit() + with contextlib.suppress(sqlite3.ProgrammingError, sqlite3.OperationalError): + conn.close() + + def _close_finished_thread_connections(self) -> None: + """Close connections whose owning threads have exited. + + The caller must hold ``_connections_lock``. + """ + for thread, conn in list(self._connections.items()): + if not thread.is_alive(): + del self._connections[thread] + self._commit_and_close(conn) + def _register_connection(self, conn: sqlite3.Connection) -> None: + thread = current_thread() with self._connections_lock: if self._closed: - with contextlib.suppress( - sqlite3.ProgrammingError, sqlite3.OperationalError - ): - conn.close() + self._commit_and_close(conn) raise exceptions.RopeError("AutoImport instance has been closed") - self._connections.add(conn) + self._close_finished_thread_connections() + self._connections[thread] = conn @property def connection(self) -> sqlite3.Connection: @@ -230,14 +247,18 @@ def connection(self) -> sqlite3.Connection: @connection.setter def connection(self, value: sqlite3.Connection): - if self._closed: - raise exceptions.RopeError("AutoImport instance has been closed") - old_conn = getattr(self.thread_local, "connection", None) - if old_conn is not None and old_conn is not value: - with self._connections_lock: - self._connections.discard(old_conn) - self._register_connection(value) - self.thread_local.connection = value + thread = current_thread() + with self._connections_lock: + if self._closed: + self._commit_and_close(value) + raise exceptions.RopeError("AutoImport instance has been closed") + self._close_finished_thread_connections() + old_conn = getattr(self.thread_local, "connection", None) + if old_conn is not None and old_conn is not value: + if self._connections.get(thread) is old_conn: + del self._connections[thread] + self._connections[thread] = value + self.thread_local.connection = value def _setup_db(self): models.Metadata.create_table(self.connection) @@ -488,29 +509,39 @@ def update_module(self, module: str): self.generate_modules_cache([module]) def close_thread_connection(self): - """Close the SQLite connection for the current thread.""" + """Close the SQLite connection owned by the current thread.""" conn = getattr(self.thread_local, "connection", None) if conn is not None: + thread = current_thread() with self._connections_lock: - self._connections.discard(conn) + if self._connections.get(thread) is conn: + del self._connections[thread] with contextlib.suppress(AttributeError): del self.thread_local.connection - with contextlib.suppress( - sqlite3.ProgrammingError, sqlite3.OperationalError - ): - conn.commit() - with contextlib.suppress( - sqlite3.ProgrammingError, sqlite3.OperationalError - ): - conn.close() + self._commit_and_close(conn) def close(self): - """Close the autoimport database.""" + """Close the database after all worker threads have stopped. + + Raises ``RopeError`` if another live thread still owns a connection. + """ + thread = current_thread() with self._connections_lock: if self._closed: return + self._close_finished_thread_connections() + active_threads = [ + owner + for owner in self._connections + if owner is not thread and owner.is_alive() + ] + if active_threads: + raise exceptions.RopeError( + "Cannot close AutoImport while other threads own connections; " + "wait for them to finish first" + ) self._closed = True - connections = list(self._connections) + connections = list(self._connections.values()) self._connections.clear() if self._observer is not None: @@ -519,14 +550,7 @@ def close(self): self._observer = None for conn in connections: - with contextlib.suppress( - sqlite3.ProgrammingError, sqlite3.OperationalError - ): - conn.commit() - with contextlib.suppress( - sqlite3.ProgrammingError, sqlite3.OperationalError - ): - conn.close() + self._commit_and_close(conn) def __enter__(self): return self @@ -534,10 +558,6 @@ def __enter__(self): def __exit__(self, exc_type, exc_val, exc_tb): self.close() - def __del__(self): - with contextlib.suppress(Exception): - self.close() - def get_name_locations(self, name): """Return a list of ``(resource, lineno)`` tuples.""" result = [] diff --git a/ropetest/contrib/autoimport/autoimporttest.py b/ropetest/contrib/autoimport/autoimporttest.py index b813af3a0..77bfde41f 100644 --- a/ropetest/contrib/autoimport/autoimporttest.py +++ b/ropetest/contrib/autoimport/autoimporttest.py @@ -1,6 +1,7 @@ import sqlite3 from concurrent.futures import ThreadPoolExecutor from contextlib import closing, contextmanager +from threading import Event from textwrap import dedent from unittest.mock import ANY, patch @@ -144,7 +145,7 @@ def worker(): all_conns = {main_conn} | set(worker_conns) assert len(all_conns) > 1 - assert all_conns.issubset(ai._connections) + assert all_conns.issubset(ai._connections.values()) assert ai._closed assert len(ai._connections) == 0 @@ -156,6 +157,56 @@ def worker(): _ = ai.connection +def test_finished_thread_connections_are_reaped(project: Project): + ai = AutoImport(project, memory=True) + worker_connections = [] + + for _ in range(3): + with ThreadPoolExecutor(1) as executor: + worker_connections.append(executor.submit(lambda: ai.connection).result()) + assert len(ai._connections) <= 2 + + for conn in worker_connections[:-1]: + with pytest.raises( + sqlite3.ProgrammingError, match="Cannot operate on a closed database" + ): + conn.execute("SELECT 1") + + ai.close() + for conn in worker_connections: + with pytest.raises( + sqlite3.ProgrammingError, match="Cannot operate on a closed database" + ): + conn.execute("SELECT 1") + + +def test_close_rejects_live_worker_connections(project: Project): + ai = AutoImport(project, memory=True) + started = Event() + finish = Event() + worker_connections = [] + + def worker(): + worker_connections.append(ai.connection) + started.set() + finish.wait() + + with ThreadPoolExecutor(1) as executor: + future = executor.submit(worker) + try: + assert started.wait(timeout=5) + with pytest.raises(exceptions.RopeError, match="wait for them to finish"): + ai.close() + assert not ai._closed + assert worker_connections[0].execute("SELECT 1").fetchone() == (1,) + finally: + finish.set() + future.result(timeout=5) + + ai.close() + assert ai._closed + + def test_close_thread_connection(project: Project): with AutoImport(project, memory=True) as ai: worker_conn = None @@ -163,14 +214,14 @@ def test_close_thread_connection(project: Project): def worker(): nonlocal worker_conn worker_conn = ai.connection - assert worker_conn in ai._connections + assert worker_conn in ai._connections.values() ai.close_thread_connection() - assert worker_conn not in ai._connections + assert worker_conn not in ai._connections.values() with pytest.raises(sqlite3.ProgrammingError, match="Cannot operate on a closed database"): worker_conn.execute("SELECT 1") new_conn = ai.connection assert new_conn is not worker_conn - assert new_conn in ai._connections + assert new_conn in ai._connections.values() with ThreadPoolExecutor(1) as tp: tp.submit(worker).result() @@ -192,7 +243,7 @@ def test_register_connection_after_close(project: Project): conn = AutoImport.create_database_connection(memory=True) with pytest.raises(exceptions.RopeError, match="AutoImport instance has been closed"): ai._register_connection(conn) - assert conn not in ai._connections + assert conn not in ai._connections.values() with pytest.raises(sqlite3.ProgrammingError, match="Cannot operate on a closed database"): conn.execute("SELECT 1") @@ -208,12 +259,12 @@ def test_connection_setter_after_close(project: Project): def test_connection_setter_replaces_existing(project: Project): with AutoImport(project, memory=True) as ai: old_conn = ai.connection - assert old_conn in ai._connections + assert old_conn in ai._connections.values() with closing(AutoImport.create_database_connection(memory=True)) as conn: ai.connection = conn assert ai.connection is conn - assert conn in ai._connections - assert old_conn not in ai._connections + assert conn in ai._connections.values() + assert old_conn not in ai._connections.values() def test_connection(project: Project, project2: Project):