Merge pull request #8 from pyper-dev/dev

Dev

Author: pyper-dev

.coveragerc

@@ -3,4 +3,7 @@ exclude_also =
     # pragma: no cover
     if TYPE_CHECKING:
     if t.TYPE_CHECKING:
-    raise NotImplementedError
+    raise NotImplementedError
+
+[run]
+    omit = **/pyper/_core/util/task_group.py

.github/workflows/test.yml

@@ -15,6 +15,9 @@ jobs:
     strategy:
       matrix:
         python-version: [
+            3.8,
+            3.9,
+            3.10,
             3.11,
             3.12,
             3.13

pyproject.toml

@@ -9,7 +9,7 @@ name = "python-pyper"
 dynamic = ["version"]
 description = "Concurrent Python made simple"
 readme = "README.md"
-requires-python = ">=3.11"
+requires-python = ">=3.8"
 authors = [
     { name = "Richard Zhu", email = "[email protected]" },
 ]
@@ -23,10 +23,17 @@ classifiers = [
     "Framework :: AsyncIO",
     "License :: OSI Approved :: MIT License",
     "Programming Language :: Python :: 3 :: Only",
+    "Programming Language :: Python :: 3.8",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
     "Programming Language :: Python :: 3.11",
     "Programming Language :: Python :: 3.12",
     "Programming Language :: Python :: 3.13"
 ]
+dependencies = [
+    # Typing support dependency only for Python versions < 3.10
+    "typing_extensions; python_version<'3.10'"
+]
 
 [project.urls]
 Documentation = "https://pyper-dev.github.io/pyper/"

src/pyper/_core/async_helper/output.py

@@ -1,12 +1,18 @@
 from __future__ import annotations
 
 import asyncio
+import sys
 from typing import TYPE_CHECKING
 
 from .stage import AsyncProducer, AsyncProducerConsumer
 from ..util.sentinel import StopSentinel
 from ..util.thread_pool import ThreadPool
 
+if sys.version_info < (3, 11):  # pragma: no cover
+    from ..util.task_group import TaskGroup, ExceptionGroup
+else:
+    from asyncio import TaskGroup
+
 if TYPE_CHECKING:
     from ..pipeline import AsyncPipeline
 
@@ -15,7 +21,7 @@ class AsyncPipelineOutput:
     def __init__(self, pipeline: AsyncPipeline):
         self.pipeline = pipeline
 
-    def _get_q_out(self, tg: asyncio.TaskGroup, tp: ThreadPool ,*args, **kwargs) -> asyncio.Queue:
+    def _get_q_out(self, tg: TaskGroup, tp: ThreadPool ,*args, **kwargs) -> asyncio.Queue:
         """Feed forward each stage to the next, returning the output queue of the final stage."""
         q_out = None
         for task, next_task in zip(self.pipeline.tasks, self.pipeline.tasks[1:] + [None]):
@@ -36,7 +42,7 @@ async def __call__(self, *args, **kwargs):
             # We unify async and thread-based concurrency by
             # 1. using TaskGroup to spin up asynchronous tasks
             # 2. using ThreadPool to spin up synchronous tasks
-            async with asyncio.TaskGroup() as tg, ThreadPool() as tp:
+            async with TaskGroup() as tg, ThreadPool() as tp:
                 q_out = self._get_q_out(tg, tp, *args, **kwargs)
                 while (data := await q_out.get()) is not StopSentinel:
                     yield data

src/pyper/_core/async_helper/stage.py

@@ -1,19 +1,25 @@
 from __future__ import annotations
 
 import asyncio
+import sys
 from typing import TYPE_CHECKING
 
 from .queue_io import AsyncDequeue, AsyncEnqueue
 from ..util.asynchronize import ascynchronize
 from ..util.sentinel import StopSentinel
 
+if sys.version_info < (3, 11):  # pragma: no cover
+    from ..util.task_group import TaskGroup
+else:
+    from asyncio import TaskGroup
+
 if TYPE_CHECKING:
     from ..util.thread_pool import ThreadPool
     from ..task import Task
 
 
 class AsyncProducer:
-    def __init__(self, task: Task, tg: asyncio.TaskGroup, tp: ThreadPool, n_consumers: int):
+    def __init__(self, task: Task, tg: TaskGroup, tp: ThreadPool, n_consumers: int):
         self.task = ascynchronize(task, tp)
         if task.concurrency > 1:
             raise RuntimeError(f"The first task in a pipeline ({task.func.__qualname__}) cannot have concurrency greater than 1")
@@ -36,7 +42,7 @@ def start(self, *args, **kwargs):
 
 
 class AsyncProducerConsumer:
-    def __init__(self, q_in: asyncio.Queue, task: Task, tg: asyncio.TaskGroup, tp: ThreadPool, n_consumers: int):
+    def __init__(self, q_in: asyncio.Queue, task: Task, tg: TaskGroup, tp: ThreadPool, n_consumers: int):
         self.q_in = q_in
         self.task = ascynchronize(task, tp)
         self.tg = tg

src/pyper/_core/decorators.py

@@ -1,13 +1,19 @@
 from __future__ import annotations
 
 import functools
+import sys
 import typing as t
 
 from .pipeline import AsyncPipeline, Pipeline
 from .task import Task
 
+if sys.version_info < (3, 10):  # pragma: no cover
+    from typing_extensions import ParamSpec
+else:
+    from typing import ParamSpec
 
-_P = t.ParamSpec('P')
+
+_P = ParamSpec('P')
 _R = t.TypeVar('R')
 _ArgsKwargs: t.TypeAlias = t.Optional[t.Tuple[t.Tuple[t.Any], t.Dict[str, t.Any]]]
 

src/pyper/_core/pipeline.py

@@ -1,18 +1,24 @@
 from __future__ import annotations
 
 import inspect
+import sys
 import typing as t
 
 from .async_helper.output import AsyncPipelineOutput
 from .sync_helper.output import PipelineOutput
 
+if sys.version_info < (3, 10):  # pragma: no cover
+    from typing_extensions import ParamSpec
+else:
+    from typing import ParamSpec
+
 if t.TYPE_CHECKING:
     from .task import Task
 
 
-_P = t.ParamSpec('P')
+_P = ParamSpec('P')
 _R = t.TypeVar('R')
-_P_Other = t.ParamSpec("P_Other")
+_P_Other = ParamSpec("P_Other")
 _R_Other = t.TypeVar("R_Other")
 
 
@@ -90,7 +96,7 @@ def __gt__(self, other: t.Callable[..., _R_Other]) -> t.Callable[_P, _R_Other]:
         return self.consume(other)
 
     def __repr__(self):
-        return f"{self.__class__.__name__} {[task.func for task in self.tasks]}"
+        return f"<{self.__class__.__name__} {[task.func for task in self.tasks]}>"
 
 
 class AsyncPipeline(Pipeline[_P, _R]):