Merge branch 'dev' into ft-multiprocessing

Author: RichardZhu2

.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

@@ -7,6 +7,9 @@ on:
     paths:
       - "src/pyper/**"
       - "tests/**"
+  
+  workflow_dispatch:
+
 
 jobs:
   test:
@@ -15,9 +18,12 @@ jobs:
     strategy:
       matrix:
         python-version: [
-            3.11,
-            3.12,
-            3.13
+            "3.8",
+            "3.9",
+            "3.10",
+            "3.11",
+            "3.12",
+            "3.13"
         ] 
 
     steps:

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, pp: ProcessPool, *args, **kwargs) -> asyncio.Queue:
+    def _get_q_out(self, tg: TaskGroup, tp: ThreadPool, pp: ProcessPool, *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]):
@@ -33,12 +39,20 @@ def _get_q_out(self, tg: asyncio.TaskGroup, tp: ThreadPool, pp: ProcessPool, *ar
     async def __call__(self, *args, **kwargs):
         """Call the pipeline, taking the inputs to the first task, and returning the output from the last task."""
         try:
+<<<<<<< HEAD
             # Unify async, threaded, and multiprocessed work by:
             # 1. using TaskGroup to execute asynchronous tasks
             # 2. using ThreadPool to execute threaded synchronous tasks
             # 3. using ProcessPool to execute multiprocessed synchronous tasks
             async with asyncio.TaskGroup() as tg, ThreadPool() as tp, ProcessPool as pp:
                 q_out = self._get_q_out(tg, tp, pp, *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 TaskGroup() as tg, ThreadPool() as tp:
+                q_out = self._get_q_out(tg, tp, *args, **kwargs)
+>>>>>>> dev
                 while (data := await q_out.get()) is not StopSentinel:
                     yield data
         except ExceptionGroup as eg:

src/pyper/_core/async_helper/stage.py

@@ -1,12 +1,18 @@
 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
@@ -16,7 +22,7 @@ class AsyncProducer:
     def __init__(
             self,
             task: Task,
-            tg: asyncio.TaskGroup,
+            tg: TaskGroup,
             tp: ThreadPool,
             pp,
             n_consumers: int):
@@ -46,7 +52,7 @@ def __init__(
             self,
             q_in: asyncio.Queue,
             task: Task,
-            tg: asyncio.TaskGroup,
+            tg: TaskGroup,
             tp: ThreadPool,
             pp,
             n_consumers: int):

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]):