Merge pull request #14 from pyper-dev/dev

Fix type hints. Update docs/readme

Author: pyper-dev

README.md

@@ -22,11 +22,11 @@
 
 ---
 
-Pyper is a comprehensive framework for concurrent and parallel data-processing, based on functional programming patterns. Used for 🌐 **Data Collection**, 🔀 **ETL Systems**, and general-purpose 🛠️ **Python Scripting**
+Pyper is a flexible framework for concurrent and parallel data-processing, based on functional programming patterns. Used for 🔀 **ETL Systems**, ⚙️ **Data Microservices**, and 🌐 **Data Collection**
 
 See the [Documentation](https://pyper-dev.github.io/pyper/)
 
-Key features:
+**Key features:**
 
 * 💡**Intuitive API**: Easy to learn, easy to think about. Implements clean abstractions to seamlessly unify threaded, multiprocessed, and asynchronous work.
 * 🚀 **Functional Paradigm**: Python functions are the building blocks of data pipelines. Let's you write clean, reusable code naturally.
@@ -99,7 +99,7 @@ if __name__ == "__main__":
     asyncio.run(main())
 ```
 
-Pyper provides an elegant abstraction of the execution of each function via `pyper.task`, allowing you to focus on building out the **logical** functions of your program. In the `main` function:
+Pyper provides an elegant abstraction of the execution of each task, allowing you to focus on building out the **logical** functions of your program. In the `main` function:
 
 * `pipeline` defines a function; this takes the parameters of its first task (`get_data`) and yields each output from its last task (`step3`)
 * Tasks are piped together using the `|` operator (motivated by Unix's pipe operator) as a syntactic representation of passing inputs/outputs between tasks.
@@ -114,7 +114,7 @@ In the pipeline, we are executing three different types of work:
 
 `task` acts as one intuitive API for unifying the execution of each different type of function.
 
-Each task submits their outputs to the next task within the pipeline via queue-based data structures, which is the mechanism underpinning how concurrency and parallelism are achieved. See the [docs](https://pyper-dev.github.io/pyper/docs/UserGuide/BasicConcepts) for a breakdown of what a pipeline looks like under the hood.
+Each task has workers that submit outputs to the next task within the pipeline via queue-based data structures; this is the mechanism underpinning how concurrency and parallelism are achieved. See the [docs](https://pyper-dev.github.io/pyper/docs/UserGuide/BasicConcepts) for a breakdown of what a pipeline looks like under the hood.
 
 ---
 

docs/src/assets/img/hint1.png

@@ -75,14 +75,14 @@ The correct way to optimize the performance of CPU-bound tasks is through parall
 # Okay
 @task(workers=10, multiprocess=True)
 def long_computation(data: int):
-    for i in range(10_000_000):
+    for i in range(1, 1_000_000):
         data *= i
     return data
 
 # Bad -- cannot benefit from concurrency
 @task(workers=10)
 def long_computation(data: int):
-    for i in range(10_000_000):
+    for i in range(1, 1_000_000):
         data *= i
     return data
 ```
@@ -150,7 +150,7 @@ The correct pattern is generally to define functions which take these resources
 from aiohttp import ClientSession
 from pyper import task
 
-async def list_user_ids(session: ClientSession) -> list:
+async def list_user_ids(session: ClientSession) -> list[int]:
     async with session.get("/users") as r:
         return await r.json()
 
@@ -163,7 +163,7 @@ When defining a pipeline, these additional arguments are plugged into tasks usin
 
 ```python
 async def main():
-    async with ClientSession("http://localhost:8000") as session:
+    async with ClientSession("http://localhost:8000/api") as session:
         user_data_pipeline = (
             task(list_user_ids, branch=True, bind=task.bind(session=session))
             | task(fetch_user_data, workers=10, bind=task.bind(session=session))
@@ -183,7 +183,7 @@ from pyper import task, AsyncPipeline
 
 def user_data_pipeline(session: ClientSession) -> AsyncPipeline:
 
-    async def list_user_ids() -> list:
+    async def list_user_ids() -> list[int]:
         async with session.get("/users") as r:
             return await r.json()
 
@@ -197,11 +197,11 @@ def user_data_pipeline(session: ClientSession) -> AsyncPipeline:
     )
 ```
 
-Now `user_data_pipeline` constructs a self-contained sata-flow, which can be reused without having to define its internal pipeline everytime.
+Now `user_data_pipeline` constructs a self-contained data-flow, which can be reused without having to define its internal pipeline everytime.
 
 ```python
 async def main():
-    async with ClientSession("http://localhost:8000") as session:
+    async with ClientSession("http://localhost:8000/api") as session:
         run = (
             user_data_pipeline(session)
             | task(write_to_file, join=True)

docs/src/assets/img/hint2.png

@@ -89,7 +89,13 @@ if __name__ == "__main__":
 ```
 
 {: .info}
-Pyper comes with fantastic IDE intellisense support which understands these operators, and will always show you which variables are `Pipeline` or `AsyncPipeline` objects; this also preserves type hints from your own functions, showing you the parameter and return type specs for each pipeline or consumer
+Pyper comes with fantastic intellisense support which understands these operators and preserves parameter/return type hints from user-defined functions
+
+<img src="../../assets/img/hint1.png" alt="Type Hint" style="width: 500; height: auto;">
+
+<img src="../../assets/img/hint2.png" alt="Type Hint" style="width: 500; height: auto;">
+
+<img src="../../assets/img/hint3.png" alt="Type Hint" style="width: 500; height: auto;">
 
 ## Nested Pipelines
 

docs/src/assets/img/hint3.png

@@ -41,10 +41,13 @@ The Python package space has great support for achieving
 
 However, the solutions for _general-purpose_ data processing are less established.
 
-Pyper aims to offer a comprehensive framework for concurrent and parallel data-processing in Python, designed with the following goals in mind:
+{: .text-green-200}
+**Pyper aims to offer a flexible framework for concurrent and parallel data-processing in Python**
+
+It is designed with the following goals in mind:
 
 * **Unified API**: Combine threads, processes and async code using one intuitive pattern
-* **Functional Paradigm**: Data pipelines compose together flexibly as functions
+* **Functional Paradigm**: Data pipelines compose together straightforwardly as functions
 * **Lazy Execution**: Built from the ground up to support generators, and provides mechanisms for fine-grained memory control
 * **Error Handling**: Data flows fail fast, even in long-running threads, and propagate their errors cleanly
 * **Complex Data Flows**: Data pipelines support branching/joining data flows, as well as sharing contexts/resources between tasks
@@ -61,6 +64,6 @@ $ pip install python-pyper
 
 ## Where Next?
 
-* Check out the 📖 **[User Guide](./docs/UserGuide/)** to get started with Pyper
+* Check out the 📖 **[User Guide](./docs/UserGuide/BasicConcepts)** to get started with Pyper
 
 * See some 🎯 **[Examples](./docs/Examples/)** of possible use cases

docs/src/docs/UserGuide/AdvancedConcepts.md

@@ -15,7 +15,8 @@
 
 _P = ParamSpec('P')
 _R = t.TypeVar('R')
-_ArgsKwargs: t.TypeAlias = t.Optional[t.Tuple[t.Tuple[t.Any], t.Dict[str, t.Any]]]
+_Default = t.TypeVar('T', bound=t.NoReturn)  # Matches to no type hints
+ArgsKwargs: t.TypeAlias = t.Optional[t.Tuple[t.Tuple[t.Any], t.Dict[str, t.Any]]]
 
 
 class task:
@@ -28,19 +29,57 @@ class task:
         workers (int): Defines the number of workers to run the task
         throttle (int): Limits the number of results the task is able to produce when all consumers are busy
         multiprocess (bool): Allows the task to be multiprocessed (cannot be `True` for async tasks)
-        bind (tuple[args, kwargs]): Additional args and kwargs to bind to the task when defining a pipeline
+        bind (tuple[tuple, dict]): Additional args and kwargs to bind to the task when defining a pipeline
 
     Returns:
-        Pipeline: A `Pipeline` instance consisting of one task.
+        A `Pipeline` instance consisting of one task.
 
-    Example:
-    ```python
-    def f(x: int):
-        return x + 1
+    Examples:
+        ```python
+        def spam(x: int):
+            return x + 1
+
+        p = task(spam)
+
+        def ham(x: int):
+            return [x, x + 1, x + 2]
 
-    p = task(f, workers=10, multiprocess=True)
-    ```
+        p = task(ham, branch=True, workers=10)
+
+        async def eggs(x: int):
+            yield x
+            yield x + 1
+            yield x + 2
+
+        p = task(eggs, branch=True, throttle=1)
+        ```
     """
+    @t.overload
+    def __new__(
+            cls,
+            func: t.Callable[_P, _Default],
+            /,
+            *,
+            branch: bool = False,
+            join: bool = False,
+            workers: int = 1,
+            throttle: int = 0,
+            multiprocess: bool = False,
+            bind: ArgsKwargs = None) -> Pipeline[_P, _Default]: ...
+    
+    @t.overload
+    def __new__(
+            cls,
+            func: None = None,
+            /,
+            *,
+            branch: t.Literal[True],
+            join: bool = False,
+            workers: int = 1,
+            throttle: int = 0,
+            multiprocess: bool = False,
+            bind: ArgsKwargs = None) -> t.Type[_branched_partial_task]: ...
+    
     @t.overload
     def __new__(
             cls,
@@ -52,20 +91,20 @@ def __new__(
             workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
-            bind: _ArgsKwargs = None) -> t.Type[task]: ...
+            bind: ArgsKwargs = None) -> t.Type[task]: ...
 
     @t.overload
     def __new__(
             cls,
             func: t.Callable[_P, t.Union[t.Awaitable[t.Iterable[_R]], t.AsyncGenerator[_R]]],
             /,
             *,
-            branch: True,
+            branch: t.Literal[True],
             join: bool = False,
             workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
-            bind: _ArgsKwargs = None) -> AsyncPipeline[_P, _R]: ...
+            bind: ArgsKwargs = None) -> AsyncPipeline[_P, _R]: ...
 
     @t.overload
     def __new__(
@@ -78,20 +117,20 @@ def __new__(
             workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
-            bind: _ArgsKwargs = None) -> AsyncPipeline[_P, _R]: ...
+            bind: ArgsKwargs = None) -> AsyncPipeline[_P, _R]: ...
 
     @t.overload
     def __new__(
             cls,
             func: t.Callable[_P, t.Iterable[_R]],
             /,
             *,
-            branch: True,
+            branch: t.Literal[True],
             join: bool = False,
             workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
-            bind: _ArgsKwargs = None) -> Pipeline[_P, _R]: ...
+            bind: ArgsKwargs = None) -> Pipeline[_P, _R]: ...
 
     @t.overload
     def __new__(
@@ -104,7 +143,7 @@ def __new__(
             workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
-            bind: _ArgsKwargs = None) -> Pipeline[_P, _R]: ...
+            bind: ArgsKwargs = None) -> Pipeline[_P, _R]: ...
 
     def __new__(
             cls,
@@ -116,25 +155,44 @@ def __new__(
             workers: int = 1,
             throttle: int = 0,
             multiprocess: bool = False,
-            bind: _ArgsKwargs = None):
+            bind: ArgsKwargs = None):
         # Classic decorator trick: @task() means func is None, @task without parentheses means func is passed. 
         if func is None:
             return functools.partial(cls, branch=branch, join=join, workers=workers, throttle=throttle, multiprocess=multiprocess, bind=bind)
         return Pipeline([Task(func=func, branch=branch, join=join, workers=workers, throttle=throttle, multiprocess=multiprocess, bind=bind)])
 
     @staticmethod
-    def bind(*args, **kwargs) -> _ArgsKwargs:
+    def bind(*args, **kwargs) -> ArgsKwargs:
         """Bind additional `args` and `kwargs` to a task.
 
         Example:
-        ```python
-        def f(x: int, y: int):
-            return x + y
+            ```python
+            def f(x: int, y: int):
+                return x + y
 
-        p = task(f, bind=task.bind(y=1))
-        p(x=1)
-        ```
+            p = task(f, bind=task.bind(y=1))
+            p(x=1)
+            ```
         """
         if not args and not kwargs:
             return None
         return args, kwargs
+
+
+class _branched_partial_task:
+    @t.overload
+    def __new__(cls, func: t.Callable[_P, _Default]) -> Pipeline[_P, _Default]: ...
+
+    @t.overload
+    def __new__(
+            cls,
+            func: t.Callable[_P, t.Union[t.Awaitable[t.Iterable[_R]], t.AsyncGenerator[_R]]]) -> AsyncPipeline[_P, _R]: ...
+        
+    @t.overload
+    def __new__(cls, func: t.Callable[_P, t.Iterable[_R]]) -> Pipeline[_P, _R]: ...
+
+    @t.overload
+    def __new__(cls, func: t.Callable[_P, _R]) -> Pipeline[_P, t.Any]: ...
+
+    def __new__(cls):
+        raise NotImplementedError