@@ -113,7 +113,7 @@ jobs:
       - name: Check formatting (black)
         working-directory: ladybug/tools/python_api
         run: |
-          uv pip install black==26.3.0
+          uv pip install black==26.10.0
           .venv/bin/black --check src_py test
 
       - name: Run ruff check
@@ -184,7 +184,7 @@ jobs:
       - name: Check formatting (black)
         working-directory: ladybug/tools/python_api
         run: |
-          uv pip install black==26.3.0
+          uv pip install black==26.10.0
           .venv/bin/black --check src_py test
 
       - name: Run ruff check
 
@@ -24,8 +24,8 @@ dev = [
     "pytest",
     "pytest-asyncio~=1.0",
     "pytest-xdist~=3.0",
-    "setuptools~=80.9",
-    "black==26.3.0",
+    "setuptools>=80.9",
+    "black>=26.3.1",
     "ruff==0.15.19",
     "mypy==1.16.0",
     "torch>=2.5.0",
 
@@ -11,6 +11,13 @@ PythonCachedImport::~PythonCachedImport() {
     allObjects.clear();
 }
 
+std::unique_lock<std::mutex> PythonCachedImport::lockForLoading() {
+    // Wait with the GIL released (detached on free-threaded builds): the holder needs the GIL to
+    // finish its import, and a detached waiter does not stall stop-the-world pauses.
+    py::gil_scoped_release release;
+    return std::unique_lock<std::mutex>(loadMutex);
+}
+
 py::handle PythonCachedImport::addToCache(py::object obj) {
     auto ptr = obj.ptr();
     allObjects.push_back(obj);
 
@@ -7,16 +7,18 @@ namespace lbug {
 
 py::handle PythonCachedItem::operator()() {
     assert((bool)PyGILState_Check());
-    // load if unloaded, return cached object if already loaded
-    if (loaded) {
+    if (isLoaded()) {
         return object;
     }
-    if (parent == nullptr) {
-        object = importCache->addToCache(std::move(py::module::import(name.c_str())));
-    } else {
-        object = importCache->addToCache(std::move((*parent)().attr(name.c_str())));
+    // The parent takes the same lock, so load it before locking.
+    py::handle parentObject = parent == nullptr ? py::handle() : (*parent)();
+    auto lock = importCache->lockForLoading();
+    if (!isLoaded()) {
+        auto obj = parent == nullptr ? py::object(py::module::import(name.c_str())) :
+                                       py::object(parentObject.attr(name.c_str()));
+        object = importCache->addToCache(std::move(obj));
+        loaded.store(true, std::memory_order_release);
     }
-    loaded = true;
     return object;
 }
 
 
@@ -1,5 +1,6 @@
 #pragma once
 
+#include <mutex>
 #include <string>
 #include <vector>
 
@@ -15,7 +16,9 @@ class PythonCachedImport {
     PythonCachedImport() = default;
     ~PythonCachedImport();
 
+    // Call with the lock from lockForLoading() held.
     py::handle addToCache(py::object obj);
+    std::unique_lock<std::mutex> lockForLoading();
 
     DateTimeCachedItem datetime;
     DecimalCachedItem decimal;
@@ -28,6 +31,7 @@ class PythonCachedImport {
     UUIDCachedItem uuid;
 
 private:
+    std::mutex loadMutex;
     std::vector<py::object> allObjects;
 };
 
 
@@ -1,5 +1,6 @@
 #pragma once
 
+#include <atomic>
 #include <memory>
 #include <string>
 
@@ -10,16 +11,16 @@ namespace lbug {
 class PythonCachedItem {
 public:
     explicit PythonCachedItem(std::string name, PythonCachedItem* parent = nullptr)
-        : name(std::move(name)), parent(parent), loaded(false) {}
+        : name(std::move(name)), parent(parent) {}
     virtual ~PythonCachedItem() = default;
 
-    bool isLoaded() const { return loaded; }
+    bool isLoaded() const { return loaded.load(std::memory_order_acquire); }
     py::handle operator()();
 
 protected:
     std::string name;
     PythonCachedItem* parent;
-    bool loaded;
+    std::atomic<bool> loaded{false};
     py::handle object;
 };
 
 
@@ -193,7 +193,7 @@ static py::object findFirstNonNull(const py::handle& row, uint64_t numRows) {
 
 common::LogicalType PandasAnalyzer::innerAnalyze(py::object column, bool& canConvert) {
     auto numRows = py::len(column);
-    auto pandasModule = importCache->pandas;
+    auto& pandasModule = importCache->pandas;
     auto pandasSeries = pandasModule.core.series.Series();
 
     if (py::isinstance(column, pandasSeries)) {
 
@@ -1,6 +1,7 @@
 #include "include/py_database.h"
 
 #include <memory>
+#include <mutex>
 
 #include "common/exception/runtime.h"
 #include "extension/extension.h"
@@ -65,6 +66,9 @@ PyDatabase::PyDatabase(const std::string& databasePath, uint64_t bufferPoolSize,
     lbug::extension::ExtensionUtils::addTableFunc<lbug::PandasScanFunction>(*state->database);
     state->storageDriver = std::make_unique<StorageDriver>(state->database.get());
     py::gil_scoped_acquire acquire;
+    // Taken with the GIL held: nothing inside may call into Python.
+    static std::mutex importCacheMutex;
+    std::lock_guard<std::mutex> lock(importCacheMutex);
     if (lbug::importCache.get() == nullptr) {
         lbug::importCache = std::make_shared<lbug::PythonCachedImport>();
     }
 
@@ -3,6 +3,7 @@
 import asyncio
 import threading
 import warnings
+import weakref
 from concurrent.futures import ThreadPoolExecutor
 from typing import TYPE_CHECKING, Any
 
@@ -51,6 +52,14 @@ def __init__(
         self.connections = [Connection(database) for _ in range(max_concurrent_queries)]
         self.connections_counter = [0 for _ in range(max_concurrent_queries)]
         self.lock = threading.Lock()
+        # Calls on one connection run one at a time (the engine serializes them
+        # anyway), so the call holding a connection's lock is the one an
+        # interrupt would hit. Queries run directly on a connection obtained
+        # from acquire_connection() are not tracked.
+        self._connection_locks: weakref.WeakKeyDictionary[
+            Connection, threading.Lock
+        ] = weakref.WeakKeyDictionary()
+        self._running: dict[Connection, _Call] = {}
 
         for conn in self.connections:
             conn.init_connection()
@@ -114,10 +123,9 @@ def release_connection(self, conn: Connection) -> None:
 
 
         """
-        for i, existing_conn in enumerate(self.connections):
-            if existing_conn == conn:
-                self.__decrement_connection_counter(i)
-                break
+        conn_index = self.__index_of(conn)
+        if conn_index is not None:
+            self.__decrement_connection_counter(conn_index)
 
     def set_query_timeout(self, timeout_in_ms: int) -> None:
         """
@@ -153,28 +161,22 @@ async def execute(
         QueryResult
             Query result.
 
+        Raises
+        ------
+        asyncio.CancelledError
+            If the awaiting task is cancelled. A running query is interrupted; a
+            query still waiting for a free connection is not started.
+
         """
-        loop = asyncio.get_running_loop()
-        # If the query is a prepared statement, use the connection associated with it
         if isinstance(query, PreparedStatement):
             conn = query._connection
-            for i, existing_conn in enumerate(self.connections):
-                if existing_conn == conn:
-                    conn_index = i
-                    with self.lock:
-                        self.connections_counter[conn_index] += 1
-                    break
+            conn_index = self.__index_of(conn)
+            if conn_index is not None:
+                with self.lock:
+                    self.connections_counter[conn_index] += 1
         else:
             conn, conn_index = self.__get_connection_with_least_queries()
-
-        try:
-            return await loop.run_in_executor(
-                self.executor, conn.execute, query, parameters
-            )
-        except asyncio.CancelledError:
-            conn.interrupt()
-        finally:
-            self.__decrement_connection_counter(conn_index)
+        return await self.__submit(conn, conn_index, conn.execute, query, parameters)
 
     async def _prepare(
         self, query: str, parameters: dict[str, Any] | None = None
@@ -183,16 +185,64 @@ async def _prepare(
         The only parameters supported during prepare are dataframes.
         Any remaining parameters will be ignored and should be passed to execute().
         """  # noqa: D401
-        loop = asyncio.get_running_loop()
         conn, conn_index = self.__get_connection_with_least_queries()
+        return await self.__submit(conn, conn_index, conn._prepare, query, parameters)
 
+    def __index_of(self, conn: Connection) -> int | None:
+        for i, existing_conn in enumerate(self.connections):
+            if existing_conn is conn:
+                return i
+        return None
+
+    async def __submit(
+        self, conn: Connection, conn_index: int | None, fn: Any, *args: Any
+    ) -> Any:
+        call = _Call()
         try:
-            preparedStatement = await loop.run_in_executor(
-                self.executor, conn.prepare, query, parameters
+            future = self.executor.submit(self.__run, conn, call, fn, *args)
+        except RuntimeError:
+            if conn_index is not None:
+                self.__decrement_connection_counter(conn_index)
+            raise
+        if conn_index is not None:
+            # Release the slot when the work has actually finished (or was
+            # cancelled before it started), not when the awaiting task exits.
+            future.add_done_callback(
+                lambda _: self.__decrement_connection_counter(conn_index)
             )
-            return preparedStatement
-        finally:
-            self.__decrement_connection_counter(conn_index)
+        try:
+            return await asyncio.wrap_future(future)
+        except asyncio.CancelledError:
+            self.__cancel(conn, call)
+            raise
+
+    def __run(self, conn: Connection, call: _Call, fn: Any, *args: Any) -> Any:
+        with self.lock:
+            connection_lock = self._connection_locks.setdefault(conn, threading.Lock())
+        with connection_lock:
+            with self.lock:
+                if call.cancelled:
+                    return None
+                self._running[conn] = call
+            try:
+                return fn(*args)
+            finally:
+                with self.lock:
+                    del self._running[conn]
+
+    def __cancel(self, conn: Connection, call: _Call) -> None:
+        # Interrupting acts on whatever the connection is running, so only
+        # interrupt while this call is the one running. A call still waiting
+        # for its connection is skipped when its turn comes.
+        with self.lock:
+            call.cancelled = True
+            running = self._running.get(conn) is call
+            if running:
+                conn.interrupt()
+        if running:
+            # An interrupt that arrives while the query is still compiling is
+            # cleared when execution starts, so repeat it until the call returns.
+            asyncio.get_running_loop().call_later(0.05, self.__cancel, conn, call)
 
     async def prepare(
         self, query: str, parameters: dict[str, Any] | None = None
@@ -225,10 +275,17 @@ def close(self) -> None:
         """
         Close all connections and shutdown the thread pool.
 
+        Queries still running are interrupted and fail with "Interrupted.".
+
         Note: Call to this method is optional. The connections and thread pool
         will be closed automatically when the instance is garbage collected.
         """
         for conn in self.connections:
             conn.close()
 
         self.executor.shutdown(wait=True)
+
+
+class _Call:
+    def __init__(self) -> None:
+        self.cancelled = False