py3.10 compat

Author: nsmith-

.pre-commit-config.yaml

@@ -48,7 +48,7 @@ repos:
   - id: setup-cfg-fmt
 
 - repo: https://github.com/astral-sh/ruff-pre-commit
-  rev: v0.14.3
+  rev: v0.14.4
   hooks:
   - id: ruff
     args: [--fix, --show-fixes]
@@ -64,5 +64,5 @@ repos:
   rev: v1.18.2
   hooks:
   - id: mypy
-    files: src/coffea/compute
+    pass_filenames: false  # to allow mypy to respect pyproject.toml config
     additional_dependencies: []

pyproject.toml

@@ -65,6 +65,7 @@ dependencies = [
   "fsspec",
   "rich",
   "ipywidgets",
+  "typing-extensions;python_version<'3.11'",
 ]
 dynamic = ["version"]
 
@@ -131,3 +132,11 @@ line-length = 160
 
 [tool.ruff.lint]
 ignore = ["F403", "F405", "E402"]
+
+[tool.mypy]
+python_version = "3.10"
+files = "src/coffea/compute"
+
+[[tool.mypy.overrides]]
+module = ["uproot.*"]
+follow_untyped_imports = true

src/coffea/__init__.py

@@ -27,8 +27,6 @@
 # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-from . import _version
+from ._version import __version__
 
-__version__ = _version.__version__
-
-__all__ = ["_version", "__version__"]
+__all__ = ["__version__"]

src/coffea/compute/backends/threaded.py

@@ -1,7 +1,7 @@
 from collections.abc import Iterator
 from dataclasses import dataclass, field
 from itertools import starmap
-from queue import Queue, ShutDown
+from queue import Queue
 from threading import Condition, Thread
 from types import TracebackType
 from typing import TYPE_CHECKING, Generic
@@ -199,11 +199,20 @@ def _run(self) -> None:
             self._cv.notify_all()
 
 
+class _Shutdown:
+    """A sentinel to signal worker threads to exit.
+
+    In python 3.13+, we can use queue.ShutDown directly.
+    """
+
+    pass
+
+
 def _work(task_queue: Queue[ThreadedTask]) -> None:  # type: ignore[type-arg]
     while True:
-        try:
-            task = task_queue.get()
-        except ShutDown:
+        task = task_queue.get()
+        if isinstance(task, _Shutdown):
+            task_queue.task_done()
             break
         try:
             task._run()
@@ -215,7 +224,7 @@ def _work(task_queue: Queue[ThreadedTask]) -> None:  # type: ignore[type-arg]
 
 
 class SingleThreadedBackend:
-    _task_queue: Queue[ThreadedTask] | None  # type: ignore[type-arg]
+    _task_queue: Queue[ThreadedTask | _Shutdown] | None  # type: ignore[type-arg]
     _thread: Thread | None
 
     def __init__(self) -> None:
@@ -230,7 +239,7 @@ def __enter__(self) -> "RunningSingleThreadedBackend":
             args=(self._task_queue,),
         )
         self._thread.start()
-        return RunningSingleThreadedBackend(self._task_queue)
+        return RunningSingleThreadedBackend(self)
 
     def __exit__(
         self,
@@ -239,32 +248,32 @@ def __exit__(
         traceback: TracebackType | None,
     ) -> None:
         assert self._thread and self._task_queue
-        self._task_queue.shutdown()
+        self._task_queue.put(_Shutdown())
         self._thread.join()
         self._task_queue = None
         self._thread = None
 
 
 class RunningSingleThreadedBackend:
-    queue: Queue[ThreadedTask]  # type: ignore[type-arg]
+    _backend: SingleThreadedBackend
 
-    def __init__(self, queue: Queue[ThreadedTask]):  # type: ignore[type-arg]
-        self.queue = queue
+    def __init__(self, backend: SingleThreadedBackend):  # type: ignore[type-arg]
+        self._backend = backend
 
     def compute(
         self,
         item: Computable[InputT, ResultT],
         /,
         error_policy: ErrorPolicy = ErrorPolicy(),
     ) -> ThreadedTask[InputT, ResultT]:
-        if self.queue.is_shutdown:
+        if self._backend._task_queue is None:
             raise RuntimeError(
                 "Cannot compute on a backend that has been exited from its context manager"
             )
         if hasattr(item, "__next__"):
             raise TypeError("Computable items must be iterables, not iterators")
         task = ThreadedTask(item, error_policy)
-        self.queue.put(task)
+        self._backend._task_queue.put(task)
         return task
 
     def wait_all(self, progress: bool = False) -> None:
@@ -278,7 +287,8 @@ def wait_all(self, progress: bool = False) -> None:
         if progress:
             raise NotImplementedError("Progress bars are not yet implemented")
         else:
-            self.queue.join()
+            if self._backend._task_queue:
+                self._backend._task_queue.join()
 
 
 if TYPE_CHECKING:

src/coffea/compute/group.py

@@ -1,5 +1,7 @@
 from dataclasses import dataclass
-from typing import Callable, Generic, Self
+from typing import Callable, Generic
+
+from typing_extensions import Self  # Python < 3.11
 
 from coffea.compute.context import ContextInput, Ctx_co
 from coffea.compute.protocol import InputT, ResultT