Dev

src/pyper/_core/pipeline.py

@@ -11,11 +11,11 @@
 
 
 class Pipeline:
-    """A sequence of at least 1 connected Tasks.
+    """A sequence of at least 1 Tasks.
 
     Two pipelines can be piped into another via:
     ```python
-    new_pipeline = p1 >> p2
+    new_pipeline = p1 | p2
     # OR
     new_pipeline = p1.pipe(p2)
     ```
@@ -43,8 +43,8 @@ def pipe(self, other) -> Pipeline:
             raise TypeError(f"{other} of type {type(other)} cannot be piped into a Pipeline")
         return Pipeline(self.tasks + other.tasks)
 
-    def __rshift__(self, other: Pipeline) -> Pipeline:
-        """Allow the syntax `pipeline1 >> pipeline2`."""
+    def __or__(self, other: Pipeline) -> Pipeline:
+        """Allow the syntax `pipeline1 | pipeline2`."""
         return self.pipe(other)
 
     def consume(self, other: Callable) -> Callable:
@@ -55,8 +55,8 @@ def consumer(*args, **kwargs):
             return consumer
         raise TypeError(f"{other} must be a callable that takes a generator")
 
-    def __and__(self, other: Callable) -> Callable:
-        """Allow the syntax `pipeline & consumer`."""
+    def __gt__(self, other: Callable) -> Callable:
+        """Allow the syntax `pipeline > consumer`."""
         return self.consume(other)
 
     def __repr__(self):

src/pyper/_core/sync_helper/output.py

@@ -38,7 +38,7 @@ def __call__(self, *args, **kwargs):
                 tp.raise_error_if_exists()
                 try:
                     # Use the timeout strategy for unblocking main thread without busy waiting
-                    if (data := q_out.get(timeout=1)) is StopSentinel:
+                    if (data := q_out.get(timeout=0.1)) is StopSentinel:
                         tp.raise_error_if_exists()
                         break
                     yield data

src/pyper/_core/task.py

@@ -7,6 +7,9 @@
 
 class Task:
     """The representation of a function within a Pipeline."""
+
+    __slots__ = "is_gen", "is_async", "func", "join", "concurrency", "throttle", "daemon"
+
     def __init__(
         self,
         func: Callable,

tests/test_async.py

@@ -32,23 +32,23 @@ async def consumer(data):
 
 @pytest.mark.asyncio
 async def test_pipeline():
-    p = task(f1) >> task(f2)
+    p = task(f1) | task(f2)
     assert p(1).__next__() == 1
 
 @pytest.mark.asyncio
 async def test_joined_pipeline():
-    p = task(af1) >> task(af2) >> task(af4, join=True)
+    p = task(af1) | task(af2) | task(af4, join=True)
     assert await p(1).__anext__() == 1
 
 @pytest.mark.asyncio
 async def test_consumer():
-    p = task(af1) >> task(af2) & consumer
+    p = task(af1) | task(af2) > consumer
     assert await p(1) == 1
 
 @pytest.mark.asyncio
 async def test_invalid_first_stage_concurrency():
     try:
-        p = task(af1, concurrency=2) >> task(af2) & consumer
+        p = task(af1, concurrency=2) | task(af2) > consumer
         await p(1)
     except Exception as e:
         assert isinstance(e, RuntimeError)
@@ -58,7 +58,7 @@ async def test_invalid_first_stage_concurrency():
 @pytest.mark.asyncio
 async def test_invalid_first_stage_join():
     try:
-        p = task(af1, join=True) >> task(af2) & consumer
+        p = task(af1, join=True) | task(af2) > consumer
         await p(1)
     except Exception as e:
         assert isinstance(e, RuntimeError)
@@ -68,7 +68,7 @@ async def test_invalid_first_stage_join():
 @pytest.mark.asyncio
 async def test_error_handling():
     try:
-        p = task(af1) >> task(af2) >> task(af3) & consumer
+        p = task(af1) | task(af2) | task(af3) > consumer
         await p(1)
     except Exception as e:
         assert isinstance(e, RuntimeError)
@@ -77,13 +77,13 @@ async def test_error_handling():
 
 @pytest.mark.asyncio
 async def test_unified_pipeline():
-    p = task(af1) >> task(f1) >> task(af2) >> task(f2) & consumer
+    p = task(af1) | task(f1) | task(af2) | task(f2) > consumer
     assert await p(1) == 1
 
 @pytest.mark.asyncio
 async def test_error_handling_in_daemon():
     try:
-        p = task(af1) >> task(af2) >> task(f3, daemon=True) & consumer
+        p = task(af1) | task(af2) | task(f3, daemon=True) > consumer
         await p(1)
     except Exception as e:
         assert isinstance(e, RuntimeError)

tests/test_sync.py

@@ -17,7 +17,7 @@ def f4(a1, a2, a3, data, k1, k2):
 
 def f5(data):
     # Make queue monitor timeout on main thread
-    time.sleep(1.2)
+    time.sleep(0.2)
     raise RuntimeError
 
 def consumer(data):
@@ -27,28 +27,28 @@ def consumer(data):
     return total
 
 def test_pipeline():
-    p = task(f1) >> task(f2)
+    p = task(f1) | task(f2)
     assert p(1).__next__() == 1
 
 def test_joined_pipeline():
-    p = task(f1) >> task(f2) >> task(f3, join=True)
+    p = task(f1) | task(f2) | task(f3, join=True)
     assert p(1).__next__() == 1
 
 def test_bind():
-    p = task(f1) >> task(f4, bind=task.bind(1, 1, 1, k1=1, k2=2))
+    p = task(f1) | task(f4, bind=task.bind(1, 1, 1, k1=1, k2=2))
     assert p(1).__next__() == 1
 
 def test_redundant_bind_ok():
-    p = task(f1) >> task(f2, bind=task.bind())
+    p = task(f1) | task(f2, bind=task.bind())
     assert p(1).__next__() == 1
 
 def test_consumer():
-    p = task(f1) >> task(f2) & consumer
+    p = task(f1) | task(f2) > consumer
     assert p(1) == 1
 
 def test_invalid_first_stage_concurrency():
     try:
-        p = task(f1, concurrency=2) >> task(f2) & consumer
+        p = task(f1, concurrency=2) | task(f2) > consumer
         p(1)
     except Exception as e:
         assert isinstance(e, RuntimeError)
@@ -57,7 +57,7 @@ def test_invalid_first_stage_concurrency():
 
 def test_invalid_first_stage_join():
     try:
-        p = task(f1, join=True) >> task(f2) & consumer
+        p = task(f1, join=True) | task(f2) > consumer
         p(1)
     except Exception as e:
         assert isinstance(e, RuntimeError)
@@ -66,7 +66,7 @@ def test_invalid_first_stage_join():
 
 def test_error_handling():
     try:
-        p = task(f1) >> task(f2) >> task(f5) & consumer
+        p = task(f1) | task(f2) | task(f5) > consumer
         p(1)
     except Exception as e:
         print(e)
@@ -76,7 +76,7 @@ def test_error_handling():
 
 def test_error_handling_in_daemon():
     try:
-        p = task(f5, daemon=True) >> task(f2) & consumer
+        p = task(f5, daemon=True) | task(f2) > consumer
         p(1)
     except Exception as e:
         print(e)

tests/test_task.py

@@ -87,36 +87,36 @@ def test_async_task():
     assert isinstance(p, AsyncPipeline)
 
 def test_piped_async_task():
-    p = task(afunc) >> task(func)
+    p = task(afunc) | task(func)
     assert isinstance(p, AsyncPipeline)
 
 def test_invalid_pipe():
     try:
-        task(func) >> 1
+        task(func) | 1
     except Exception as e:
         assert isinstance(e, TypeError)
     else:
         raise AssertionError
 
 def test_invalid_async_pipe():
     try:
-        task(afunc) >> 1
+        task(afunc) | 1
     except Exception as e:
         assert isinstance(e, TypeError)
     else:
         raise AssertionError
 
 def test_invalid_consumer():
     try:
-        task(func) & 1
+        task(func) > 1
     except Exception as e:
         assert isinstance(e, TypeError)
     else:
         raise AssertionError
 
 def test_invalid_async_consumer():
     try:
-        task(afunc) & func
+        task(afunc) > func
     except Exception as e:
         assert isinstance(e, TypeError)
     else: