Pipeline signature: move source to first argument

Author: jsbueno

extraasync/pipeline.py

@@ -26,6 +26,14 @@
 EOD = object()
 EXC_MARKER = object()
 
+try:
+    from functools import Placeholder  # new in Python 3.14
+except ImportError:
+    # Pickleable sentinel:
+
+    class Placeholder:
+        pass
+
 
 class Heap:
     def __init__(self):
@@ -48,11 +56,12 @@ def __bool__(self):
     def __repr__(self):
         return f"<Heap {self.data!r}>"
 
+
 class AutoSet(MutableSet):
     """Set with an associated asyncio.Queue
 
-    WHenever an item is removed/discarded, an item is fetched from the queue as
-    a task factory - no argument callable: it is called and added to the set.
+    Whenever an item is removed/discarded, an item is fetched from the queue as
+    a task factory-no argument callable: it is called and added to the set.
 
 
     """
@@ -169,6 +178,10 @@ def __call__(self, value):
         "just run the stage"
         return self.code(value)
 
+    def __repr__(self):
+        return f"{self.__class__.__name__}{self.code}"
+
+
 
 class Pipeline:
     """
@@ -185,32 +198,32 @@ class Pipeline:
 
     def __init__(
         self,
+        source: t.Optional[t.AsyncIterable[T] | t.Iterable[T]],
         *stages: t.Sequence[t.Callable | Stage],
-        data: t.Optional[t.AsyncIterable[T] | t.Iterable[T]],
         max_concurrency: t.Optional[int] = None,
         on_error: PipelineErrors = "strict",
         preserve_order: bool = False,
         max_simultaneous_records: t.Optional[int] = None,
     ):
-    """
-    Args:
-        - stages: One async or sync callable which will process one data item at a time
-            - TBD? accept generators as stages? (input data would be ".send"ed into it)
-        - data: async or sync generator representing the data source
-        - max_concurrency: Maximum number of concurrent tasks _for_ _each_ stage
-            (i.e. if there are 2 stages, and max_concurrency is set to 4, we may have
-            up to 8 concurrent tasks running at once in the pipeline, but each stage is
-            limited to 4)
-        - on_error: WHat to do if any stage raises an exeception - defaults to re-raise the
-                exception and stop the whole pipeline
-        - preserve_order: whether to yield the final results in the same order they were acquired from data.
-        - max_simultaneous_records: limit on amount of records to hold across all stages and input in internal
-            data structures: the idea is throtle data consumption in order to limit the
-            amount of memory used by the Pipeline
+        """
+        Args:
+            - stages: One async or sync callable which will process one data item at a time
+                - TBD? accept generators as stages? (input data would be ".send"ed into it)
+            - data: async or sync generator representing the data source
+            - max_concurrency: Maximum number of concurrent tasks _for_ _each_ stage
+                (i.e. if there are 2 stages, and max_concurrency is set to 4, we may have
+                up to 8 concurrent tasks running at once in the pipeline, but each stage is
+                limited to 4)
+            - on_error: WHat to do if any stage raises an exeception - defaults to re-raise the
+                    exception and stop the whole pipeline
+            - preserve_order: whether to yield the final results in the same order they were acquired from data.
+            - max_simultaneous_records: limit on amount of records to hold across all stages and input in internal
+                data structures: the idea is throtle data consumption in order to limit the
+                amount of memory used by the Pipeline
 
-    """
+        """
         self.max_concurrency = max_concurrency
-        self.data = _as_async_iterable(data)
+        self.data = _as_async_iterable(source) if source not in (None, Placeholder) else None
         self.preserve_order = preserve_order
         # TBD: maybe allow limitting total memory usage instead of elements in the pipeline?
         self.max_simultaneous_records = max_simultaneous_records

tests/test_pipeline.py

@@ -21,7 +21,7 @@ async def map_function(n, map_interval=0):
         return n * 2
 
     results = []
-    async for result in Pipeline(map_function, data=producer(10)):
+    async for result in Pipeline(producer(10), map_function):
         results.append(result)
 
     assert results == list(range(0, 20, 2))
@@ -38,7 +38,7 @@ def map_function(n, map_interval=0):
         return n * 2
 
     results = []
-    async for result in Pipeline(map_function, data=producer(10)):
+    async for result in Pipeline(producer(10), map_function):
         results.append(result)
 
     assert results == list(range(0, 20, 2))
@@ -59,7 +59,7 @@ async def f2(v):
         return v + 1
 
     results = []
-    async for result in Pipeline(map_function, f2, data=producer(10)):
+    async for result in Pipeline(producer(10), map_function, f2):
         results.append(result)
 
     assert results == [n * 2 + 1 for n in range(10)]
@@ -88,7 +88,7 @@ async def map_function(n):
 
     results = []
     async for result in Pipeline(
-        map_function, data=producer(task_amount), max_concurrency=None
+        producer(task_amount), map_function, max_concurrency=None
     ):
         results.append(result)
 
@@ -130,7 +130,7 @@ async def map_function(n):
 
     results = []
     async for result in Pipeline(
-        map_function, data=producer(task_amount), max_concurrency=max_concurrency
+        producer(task_amount), map_function, max_concurrency=max_concurrency
     ):
         results.append(result)
 
@@ -216,7 +216,9 @@ async def stage(value):
         await asyncio.sleep(int(f"{value:02d}"[::-1]) / 200)
         return value
 
-    assert await Pipeline(stage, data=generator(100), preserve_order=True).results() == list(range(100))
+    assert await Pipeline(generator(100), stage, preserve_order=True).results() == list(
+        range(100)
+    )
 
 
 @pytest.mark.asyncio
@@ -234,7 +236,9 @@ async def map_function(n):
     try:
         async with asyncio.timeout(0.1):
             try:
-                async for result in Pipeline(map_function, data=producer(1), on_error="strict"):
+                async for result in Pipeline(
+                    producer(1), map_function, on_error="strict"
+                ):
                     results.append(result)
             except* Exception:
                 # for this specific test, any other outcome is good.
@@ -276,51 +280,34 @@ async def map_function(n):
 
 @pytest.mark.skip
 @pytest.mark.asyncio
-async def test_pipeline_reorder_results():
-    ...
+async def test_pipeline_reorder_results(): ...
+
 
 @pytest.mark.skip
 @pytest.mark.asyncio
-async def test_pipeline_add_stage_pipe_operator():
-
-    ...
-
+async def test_pipeline_add_stage_pipe_operator(): ...
 
 
 @pytest.mark.skip
 @pytest.mark.asyncio
-async def test_pipeline_add_data_and_execute_l_rhift_operator():
-
-    ...
+async def test_pipeline_add_data_and_execute_l_rhift_operator(): ...
 
 
 @pytest.mark.skip
 @pytest.mark.asyncio
-async def test_pipeline_store_result_r_rshift_operator():
-    ...
+async def test_pipeline_store_result_r_rshift_operator(): ...
 
 
 @pytest.mark.skip
 @pytest.mark.asyncio
-async def test_pipeline_fine_tune_stages():
-    ...
+async def test_pipeline_fine_tune_stages(): ...
 
 
 @pytest.mark.skip
 @pytest.mark.asyncio
-async def test_pipeline_concurrency_rate_limit():
-    ...
+async def test_pipeline_concurrency_rate_limit(): ...
 
 
 @pytest.mark.skip
 @pytest.mark.asyncio
-async def test_pipeline_max_simultaneous_record_limit():
-    ...
-
-
-
-
-
-
-
-
+async def test_pipeline_max_simultaneous_record_limit(): ...