### Installing Documentation Dependencies with Bundler

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/Resources/Contributing.md

These commands navigate to the `docs` directory and then use `bundle install` to install all Ruby gem dependencies required for serving the documentation locally. This prepares the environment for Jekyll.

```console
cd docs
bundle install
```

--------------------------------

### Installing Test Dependencies with pip

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/Resources/Contributing.md

This command installs the necessary Python packages for running tests, specified in the `tests/requirements.txt` file. It should be executed within a virtual environment to isolate dependencies.

```console
pip install -r tests/requirements.txt
```

--------------------------------

### Serving Documentation Locally with Jekyll

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/Resources/Contributing.md

This command uses `bundle exec` to run `jekyll serve`, which starts a local web server for the documentation site. The site will be accessible at `http://localhost:4000`, allowing contributors to preview their documentation changes.

```console
bundle exec jekyll serve
```

--------------------------------

### Installing Pyper Python Package

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/index.md

This command installs the Pyper library from PyPI using pip, Python's package installer. It ensures that the latest stable version of `python-pyper` is available in your environment for use in concurrent data processing applications.

```Console
$ pip install python-pyper
```

--------------------------------

### Full AsyncPipeline Example with Mixed Tasks and Async Consumer - Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/ComposingPipelines.md

This comprehensive example demonstrates building and executing an `AsyncPipeline` that integrates both asynchronous (`step1`) and synchronous (`step2`) tasks. It includes an `AsyncJsonFileWriter` class acting as an asynchronous consumer to process and write the pipeline's output to a JSON file, showcasing how to handle asynchronous output and run the pipeline using `asyncio.run`.

```Python
import asyncio
import json
from typing import AsyncIterable, Dict

from pyper import task

async def step1(limit: int):
    for i in range(limit):
        yield {"data": i}

def step2(data: Dict):
    return data | {"hello": "world"}

class AsyncJsonFileWriter:
    def __init__(self, filepath):
        self.filepath = filepath
    
    async def __call__(self, data: AsyncIterable[Dict]):
        data_list = [row async for row in data]
        with open(self.filepath, 'w', encoding='utf-8') as f:
            json.dump(data_list, f, indent=4)

async def main():
    run = (
        task(step1, branch=True)
        | task(step2)
        > AsyncJsonFileWriter("data.json")
    )
    await run(limit=10)

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Example Output: Player Performance by Opening (Console)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/Examples/ChessDataAnalysis.md

This snippet shows an example of the console output generated by the `main` function. It's a pandas DataFrame displaying the top 10 chess openings played by the specified player (Hikaru in this case) when playing as White, along with the total number of games and the average score achieved for each opening, sorted by total games.

```Console
opening               total_games  average_score

Nimzowitsch Larsen            244       0.879098
Closed Sicilian               205       0.924390
Caro Kann                     157       0.882166
Bishops Opening               156       0.900641
French Defense                140       0.846429
Sicilian Defense              127       0.877953
Reti Opening                   97       0.819588
Vienna Game                    71       0.929577
English Opening                61       0.868852
Scandinavian Defense           51       0.862745
```

--------------------------------

### Installing Pyper Python Package

Source: https://github.com/pyper-dev/pyper/blob/main/README.md

This command installs the Pyper library from PyPI using the pip package manager. It makes the `python-pyper` package available for use in Python projects, enabling access to its concurrent and parallel processing functionalities.

```console
$ pip install python-pyper
```

--------------------------------

### Composing Multi-Task Pipelines with Pyper

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/BasicConcepts.md

This example illustrates Pyper's composability, showing how multiple `task` objects can be chained together using the `|` operator to form a complex pipeline. It defines three functions (`len_strings`, `sleep`, `calculate`) and combines them, demonstrating how `workers` can be specified for concurrency, with the final pipeline taking two strings and producing a boolean output.

```Python
import time
from pyper import task

def len_strings(x: str, y: str) -> int:
    return len(x) + len(y)

def sleep(data: int) -> int:
    time.sleep(data)
    return data

def calculate(data: int) -> bool:
    time.sleep(data)
    return data % 2 == 0

pipeline = (
    task(len_strings)
    | task(sleep, workers=3)
    | task(calculate, workers=2)
)
```

--------------------------------

### Creating a Branching Pipeline for Multiple Outputs (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/CreatingPipelines.md

This example shows how to create a 'branching' pipeline using the `branch=True` parameter in the `task` decorator. This configuration allows a task to yield multiple outputs, which are then individually processed by subsequent pipeline steps, enabling parallel processing of results.

```python
from pyper import task

def func(x: int):
    yield x + 1
    yield x + 2
    yield x + 3

if __name__ == "__main__":
    pipeline = task(func, branch=True)
    for output in pipeline(x=0):
        print(output)
        # > 1
        # > 2
        # > 3
```

--------------------------------

### Defining a Pyper Pipeline with a Custom Consumer

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/ComposingPipelines.md

This example defines a multi-step Pyper pipeline and a custom 'consumer' class, `JsonFileWriter`, which processes the pipeline's final output. It demonstrates how to manually run the pipeline and pass its results to the consumer for processing, such as writing data to a JSON file.

```Python
import json
from typing import Dict, Iterable

from pyper import task

def step1(limit: int):
    for i in range(limit):
        yield {"data": i}

def step2(data: Dict):
    return data | {"hello": "world"}

class JsonFileWriter:
    def __init__(self, filepath):
        self.filepath = filepath
    
    def __call__(self, data: Iterable[Dict]):
        data_list = list(data)
        with open(self.filepath, 'w', encoding='utf-8') as f:
            json.dump(data_list, f, indent=4)

if __name__ == "__main__":
    pipeline = task(step1, branch=True) | task(step2)  # The pipeline
    writer = JsonFileWriter("data.json")  # A consumer
    writer(pipeline(limit=10))  # Run
```

--------------------------------

### Wrapping Various Callables with Pyper `task` Decorator (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/CreatingPipelines.md

This example illustrates the flexibility of Pyper's `task` decorator, showing that it can wrap any Python callable. It demonstrates wrapping a class instance with a `__call__` method, a lambda function, and a built-in function like `range`, highlighting the decorator's broad applicability.

```python
from pyper import task

class Doubler:
    def __call__(self, x: int):
        return 2 * x

pipeline1 = task(Doubler())
pipeline2 = task(lambda x: x - 1)
pipeline3 = task(range)
```

--------------------------------

### Defining Task Logic with `func` Parameter in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/task.md

This example demonstrates how to use the `func` parameter to define the core logic of a task. The `func` parameter accepts a callable object or function, which will be executed as part of the pipeline.

```python
from pyper import task

def add_one(x: int):
    return x + 1

pipeline = task(add_one)
```

--------------------------------

### Building a Synchronous Pyper Pipeline in Python

Source: https://github.com/pyper-dev/pyper/blob/main/README.md

This example illustrates how to create a purely synchronous data pipeline with `pyper.task`. It defines synchronous functions for data generation (`get_data`), I/O-bound work (`step1`), and CPU-bound work (`step2`). The pipeline is constructed and executed synchronously, demonstrating Pyper's default non-async behavior when all tasks are synchronous.

```Python
import time

from pyper import task


def get_data(limit: int):
    for i in range(limit):
        yield i

def step1(data: int):
    time.sleep(1)
    print("Finished sync wait", data)
    return data

def step2(data: int):
    for i in range(10_000_000):
        _ = i*i
    print("Finished heavy computation", data)
    return data


def main():
    pipeline = task(get_data, branch=True) \
        | task(step1, workers=20) \
        | task(step2, workers=20, multiprocess=True)
    total = 0
    for output in pipeline(limit=20):
        total += output
    print("Total:", total)


if __name__ == "__main__":
    main()
```

--------------------------------

### Generating Chess.com API URLs for Monthly PGN Downloads (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/Examples/ChessDataAnalysis.md

This function generates a series of URLs for fetching monthly PGN game data from the chess.com API for a specified player. It calculates URLs for a given number of recent months, starting from the current date and going backward. It yields each URL, making it suitable for iterative processing.

```Python
def generate_urls_by_month(player: str, num_months: int):
    """Define a series of pgn game resource urls for a player, for num_months recent months."""
    today = datetime.date.today()
    for i in range(num_months):
        d = today - relativedelta(months=i)
        yield f"https://api.chess.com/pub/player/{player}/games/{d.year}/{d.month:02}/pgn"
```

--------------------------------

### Binding Additional Arguments to a Pyper Task (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/task.md

This example illustrates how to use the `bind` parameter to supply additional keyword arguments to a `pyper` task. The `apply_multiplier` function receives a `multiplier` value of 10, which is bound to the task using `task.bind(multiplier=10)`, allowing flexible function definitions and access to external states within the pipeline.

```python
from pyper import task

def apply_multiplier(data: int, multiplier: int):
    return data * multiplier

if __name__ == "__main__":
    pipeline = (
        task(range, branch=True)
        | task(apply_multiplier, bind=task.bind(multiplier=10))
    )
    for output in pipeline(1, 4):
        print(output)
        #> 10
        #> 20
        #> 30
```

--------------------------------

### Using `branch` with Generator Functions for Lazy Output in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/task.md

This example shows how the `branch=True` parameter can be effectively used with generator functions. This allows tasks to submit outputs lazily, which can be beneficial for memory management when dealing with large datasets.

```python
from pyper import task

def create_data(x: int):
    yield 1
    yield 2
    yield 3

if __name__ == "__main__":
    pipeline = task(create_data, branch=True)
    for output in pipeline(0):
        print(output)
        #> 1
        #> 2
        #> 3
```

--------------------------------

### Using Pyper Tasks with Branched and Non-Branched Generators in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/AdvancedConcepts.md

This Python example illustrates the use of Pyper's `task` decorator with generator functions, demonstrating the effect of the `branch=True` parameter. When `branch=True`, individual yielded values are processed immediately by subsequent tasks. Without `branch=True`, the task outputs the generator object itself, requiring further iteration to access values.

```python
from pyper import task

def get_data():
    yield 1
    yield 2
    yield 3

if __name__ == "__main__":
    branched_pipeline = task(get_data, branch=True)
    for output in branched_pipeline():
        print(output)
        #> 1
        #> 2
        #> 3

    non_branched_pipeline = task(get_data)
    for output in non_branched_pipeline():
        print(output)
        #> <generator object get_data at ...>
```

--------------------------------

### Optimizing CPU-bound Tasks with Multiprocessing in Pyper

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/AdvancedConcepts.md

This snippet illustrates how to correctly handle CPU-bound tasks in Pyper. It defines a `long_computation` function that performs intensive calculations. The example shows that using `multiprocess=True` with `task()` is the correct way to leverage parallel execution for CPU-bound work, whereas omitting it (relying on threads) will not improve performance due to the GIL.

```python
def long_computation(data: int):
    for i in range(1, 1_000_000):
        data *= i
    return data

# Okay
pipeline = task(long_computation, workers=10, multiprocess=True)

# Bad -- cannot benefit from concurrency
pipeline = task(long_computation, workers=10)
```

--------------------------------

### Illustrating Concurrency Benefits for IO-bound Tasks in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/AdvancedConcepts.md

This snippet demonstrates correct and incorrect ways to implement IO-bound functions for concurrency. It shows that synchronous `time.sleep()` in an `async` function prevents other tasks from progressing, while `await asyncio.sleep()` allows for concurrent execution. A synchronous function using `time.sleep()` is also shown as an 'Okay' example, implying it's fine if not intended for async concurrency.

```python
# Okay
def slow_func():
    time.sleep(5)

# Okay
async def slow_func():
    await asyncio.sleep(5)

# Bad -- cannot benefit from concurrency
async def slow_func():
    time.sleep(5)
```

--------------------------------

### Composing a Concurrent Data Processing Pipeline with Pyper (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/Examples/ChessDataAnalysis.md

The `main` function orchestrates the entire data collection and analysis process using the Pyper framework. It defines a concurrent pipeline where URLs are generated, fetched in parallel (3 workers), PGN data is parsed into individual game details, and finally, a pandas DataFrame is built and analyzed. It demonstrates how to bind session and player parameters to tasks and use branching for multiple outputs.

```Python
def main():
    player = "hikaru"
    num_months = 6  # Keep this number low, or add sleeps for etiquette

    with requests.Session() as session:
        run = (
            task(generate_urls_by_month, branch=True)
            | task(
                fetch_text_data,
                workers=3,
                bind=task.bind(session=session))
            | task(
                read_game_data,
                branch=True,
                bind=task.bind(player=player))
            > build_df
        )
        df = run(player, num_months)
        print(df.head(10))
```

--------------------------------

### Running Tests and Coverage with tox for Python 3.12

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/Resources/Contributing.md

This command executes all tests for the project within a Python 3.12 environment using `tox` and generates a code coverage report. It ensures that the code meets the 100% coverage target for the specified Python version.

```console
tox -e 3.12
```

--------------------------------

### Running All Tests Across Supported Versions with Docker Compose

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/Resources/Contributing.md

These commands navigate to the `tests` directory and then use `docker-compose` to build and run all test containers in detached mode. This allows verifying test success across all supported Python versions without local environment configuration.

```console
cd tests
docker-compose up --build --detach
```

--------------------------------

### Executing a Reusable Pyper Pipeline with Shared Resources

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/AdvancedConcepts.md

This snippet demonstrates how to execute a reusable Pyper pipeline created by a factory function. The `ClientSession` is managed within the `main` function's `async with` block, and then passed to `user_data_pipeline` to construct the pipeline. This allows the pipeline to be integrated into larger data flows, such as chaining with `write_to_file` and `copy_to_db` tasks, while ensuring proper resource handling.

```python
async def main():
    async with ClientSession("http://localhost:8000/api") as session:
        run = (
            user_data_pipeline(session)
            | task(write_to_file, join=True)
            > copy_to_db
        )
        await run()
```

--------------------------------

### Initializing pyper.Pipeline in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/Pipeline.md

This snippet shows the `__new__` method for `pyper.Pipeline`, which represents a data flow consisting of a series of tasks. It's important to note that direct instantiation of `Pipeline` is not recommended; users should typically use the `task` class instead.

```python
def __new__(cls, tasks: List[Task]) -> Pipeline:
```

--------------------------------

### Defining Tasks with Shared Resources (Arguments)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/AdvancedConcepts.md

This snippet illustrates how to define asynchronous functions (`list_user_ids`, `fetch_user_data`) that explicitly accept shared resources, such as an `aiohttp.ClientSession`, as arguments. This pattern allows resources to be managed externally and passed into tasks, promoting reusability and proper resource lifecycle management using Python's `with` syntax.

```python
from aiohttp import ClientSession
from pyper import task

async def list_user_ids(session: ClientSession) -> list[int]:
    async with session.get("/users") as r:
        return await r.json()

async def fetch_user_data(user_id: int, session: ClientSession) -> dict:
    async with session.get(f"/users/{user_id}") as r:
        return await r.json()
```

--------------------------------

### Parsing PGN Data and Extracting Game Details (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/Examples/ChessDataAnalysis.md

This snippet defines two functions: `_clean_opening_name` and `read_game_data`. `_clean_opening_name` extracts a simplified opening name from a chess.com ECO URL. `read_game_data` parses PGN text, extracting individual game headers, determining the player's color and score, and yielding a dictionary for each game with `color`, `score`, and `opening`.

```Python
def _clean_opening_name(eco_url: str):
    """Get a rough opening name from the chess.com ECO url."""
    name = eco_url.removeprefix("https://www.chess.com/openings/")
    return " ".join(name.split("-")[:2])


def read_game_data(pgn_text: str, player: str):
    """Read PGN data and generate game details (each PGN contains details for multiple games)."""
    pgn = io.StringIO(pgn_text)
    while (headers := chess.pgn.read_headers(pgn)) is not None:
        color = 'W' if headers["White"].lower() == player else 'B'
        
        if headers["Result"] == "1/2-1/2":
            score = 0.5
        elif (color == 'W' and headers["Result"] == "1-0") or (color == 'B' and headers["Result"] == "0-1"):
            score = 1
        else:
            score = 0
        
        yield {
            "color": color,
            "score": score,
            "opening": _clean_opening_name(headers["ECOUrl"]) 
        }
```

--------------------------------

### Fetching Raw Text Data from a URL (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/Examples/ChessDataAnalysis.md

This helper function fetches raw text data from a given URL using a `requests.Session` object. It includes a `User-Agent` header to mimic a web browser, which can be important for API etiquette and avoiding blocks. It returns the content of the response as plain text.

```Python
def fetch_text_data(url: str, session: requests.Session):
    """Fetch text data from a url."""
    r = session.get(url, headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"})
    return r.text
```

--------------------------------

### Executing a Pyper Pipeline and Iterating Outputs (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/CreatingPipelines.md

This snippet demonstrates how to execute a Pyper `Pipeline` by calling it like a function with input arguments. It shows how to iterate over the `Generator` object returned by the pipeline to access its outputs, illustrating the lazy evaluation mechanism.

```python
from pyper import task

def func(x: int):
    return x + 1

if __name__ == "__main__":
    pipeline = task(func)
    for output in pipeline(x=0):
        print(output)
        # > 1
```

--------------------------------

### Building an Asynchronous Pyper Pipeline in Python

Source: https://github.com/pyper-dev/pyper/blob/main/README.md

This snippet demonstrates how to construct an asynchronous data pipeline using `pyper.task`. It defines functions for data generation (`get_data`), asynchronous I/O (`step1`), synchronous I/O (`step2`), and CPU-bound computation (`step3`), then orchestrates them into a pipeline with specified worker counts and multiprocessing for `step3`. The `main` function executes the pipeline and aggregates results.

```Python
import asyncio
import time

from pyper import task


def get_data(limit: int):
    for i in range(limit):
        yield i


async def step1(data: int):
    await asyncio.sleep(1)
    print("Finished async wait", data)
    return data


def step2(data: int):
    time.sleep(1)
    print("Finished sync wait", data)
    return data


def step3(data: int):
    for i in range(10_000_000):
        _ = i*i
    print("Finished heavy computation", data)
    return data


async def main():
    # Define a pipeline of tasks using `pyper.task`
    pipeline = task(get_data, branch=True) \
        | task(step1, workers=20) \
        | task(step2, workers=20) \
        | task(step3, workers=20, multiprocess=True)

    # Call the pipeline
    total = 0
    async for output in pipeline(limit=20):
        total += output
    print("Total:", total)


if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Executing a Composed Pyper Pipeline

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/ComposingPipelines.md

This code snippet illustrates how to execute a previously composed Pyper pipeline. It shows how to call the `new_pipeline` with an input value and iterate over its outputs, demonstrating the flow of data through the chained tasks.

```Python
if __name__ == "__main__":
    for output in new_pipeline(4):
        print(output)
        #> 9
```

--------------------------------

### Creating a Basic Pipeline with `task` Decorator (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/CreatingPipelines.md

This snippet demonstrates how to create a basic Pyper `Pipeline` object by wrapping a standard Python function with the `task` decorator. It shows that the result of `task(func)` is an instance of `Pipeline`, confirming the successful creation of a single-task pipeline.

```python
from pyper import task, Pipeline

def func(x: int):
    return x + 1

pipeline = task(func)

assert isinstance(pipeline, Pipeline)
```

--------------------------------

### Calling pyper.Pipeline Instances in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/Pipeline.md

This snippet illustrates the `__call__` method, making a `Pipeline` object callable. It adopts the parameter specification of its first task and generates output from its last task, effectively executing the pipeline.

```python
def __call__(self, *args, **kwargs) -> Generator[Any, None, None]:
```

--------------------------------

### Creating and Running an Asynchronous Pyper Pipeline (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/CreatingPipelines.md

This snippet demonstrates how to define and execute an `AsyncPipeline` in Pyper using asynchronous functions. It shows the use of `async for` to iterate over the `AsyncGenerator` returned by the pipeline and `asyncio.run()` to execute the main asynchronous function, enabling non-blocking operations.

```python
import asyncio
from pyper import task

async def func(x: int):
    return x + 1

async def main():
    pipeline = task(func)
    async for output in pipeline(x=0):
        print(output)
        # > 1

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Creating a Single-Task Pipeline with Pyper

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/BasicConcepts.md

This snippet demonstrates how to create a basic Pyper pipeline by wrapping a standard Python function, `len_strings`, with the `task` class. The `pipeline` object then represents this single functional operation, taking two string inputs and returning an integer sum of their lengths.

```Python
from pyper import task

def len_strings(x: str, y: str) -> int:
    return len(x) + len(y)

pipeline = task(len_strings)
```

--------------------------------

### Creating a Large List Synchronously in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/AdvancedConcepts.md

This Python function demonstrates creating a large list of dictionaries synchronously. It highlights a scenario where subsequent tasks in a pipeline must wait for the entire list to be generated before proceeding, which can be inefficient for large datasets.

```python
def create_values_in_list() -> typing.List[dict]:
    return [{"data": i} for i in range(10_000_000)]
```

--------------------------------

### Initializing a pyper.task Pipeline in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/task.md

This is the constructor signature for `pyper.task`, used to initialize a Pipeline object with a single functional operation. Pipelines created this way can be composed into more complex pipelines with multiple tasks.

```python
def __new__(
    cls,
    func: Optional[Callable] = None,
    /,
    *,
    branch: bool = False,
    join: bool = False,
    workers: int = 1,
    throttle: int = 0,
    multiprocess: bool = False,
    bind: Optional[Tuple[Tuple[Any], Dict[str, Any]]] = None):
```

--------------------------------

### Composing an Outer Pipeline with a Nested Pyper Pipeline

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/ComposingPipelines.md

This snippet shows how to embed a previously defined pipeline (`download_files_from_source`) as a task within a larger, outer pipeline. It illustrates how to handle scenarios where an inner pipeline generates a batch of outputs that need to be processed together by subsequent tasks in the outer pipeline.

```Python
download_and_merge_files = (
    task(get_sources, branch=True)  # Return a list of sources
    | task(download_files_from_source)  # Return a batch of filepaths (as a generator)
    | task(sync_files, workers=5)  # Do something with each batch
)
```

--------------------------------

### Pyper task.bind Utility Method Signature (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/task.md

This snippet presents the static method signature for `task.bind`, a utility method used to create argument bindings for `pyper` tasks. It accepts arbitrary positional (`*args`) and keyword (`**kwargs`) arguments, which are then passed to `functools.partial` internally to pre-fill parameters for the bound task function.

```python
@staticmethod
def bind(*args, **kwargs):
```

--------------------------------

### Defining an Inner Nested Pyper Pipeline

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/ComposingPipelines.md

This code defines an inner Pyper pipeline, `download_files_from_source`, which simulates a multi-step file processing workflow. It demonstrates chaining tasks like listing, downloading, and decrypting files, with specified worker counts for concurrency.

```Python
download_files_from_source = (
    task(list_files, branch=True)  # Return a list of file info
    | task(download_file, workers=20)  # Return a filepath
    | task(decrypt_file, workers=5, multiprocess=True)  # Return a filepath
)
```

--------------------------------

### Creating Reusable Pipelines with Resource Factories

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/AdvancedConcepts.md

This snippet shows how to create a reusable Pyper pipeline by encapsulating it within a factory function (`user_data_pipeline`) that accepts the shared resource (e.g., `ClientSession`) as an argument. Inner functions (`list_user_ids`, `fetch_user_data`) capture the session from the outer scope, allowing the pipeline to be self-contained and modularly reused without redefining its internal structure each time.

```python
from aiohttp import ClientSession
from pyper import task, AsyncPipeline

def user_data_pipeline(session: ClientSession) -> AsyncPipeline:

    async def list_user_ids() -> list[int]:
        async with session.get("/users") as r:
            return await r.json()

    async def fetch_user_data(user_id: int) -> dict:
        async with session.get(f"/users/{user_id}") as r:
            return await r.json()
    
    return (
        task(list_user_ids, branch=True)
        | task(fetch_user_data, workers=10)
    )
```

--------------------------------

### Binding Shared Resources to Pyper Tasks

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/AdvancedConcepts.md

This snippet demonstrates how to bind a shared resource, like an `aiohttp.ClientSession`, to Pyper tasks using `task.bind`. The `session` object is created within an `async with` block in the `main` function, ensuring its proper lifecycle, and then explicitly passed to `fetch_user_data` via `task.bind(session=session)`, making the resource available to the task.

```python
async def main():
    async with ClientSession("http://localhost:8000/api") as session:
        user_data_pipeline = (
            task(list_user_ids, branch=True)
            | task(fetch_user_data, workers=10, bind=task.bind(session=session))
        )
        async for output in user_data_pipeline(session):
            print(output)
```

--------------------------------

### Composing Pyper Pipelines with the Pipe Operator

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/ComposingPipelines.md

This snippet demonstrates how to compose multiple Pyper tasks into a single pipeline using the `|` operator, which is syntactic sugar for the `Pipeline.pipe` method. It shows how individual tasks are chained together to form a new, combined pipeline.

```Python
from pyper import task, Pipeline

p1 = task(lambda x: x + 1)
p2 = task(lambda x: 2 * x)
p3 = task(lambda x: x - 1)

new_pipeline = p1 | p2 | p3
assert isinstance(new_pipeline, Pipeline)
# OR
new_pipeline = p1.pipe(p2).pipe(p3)
assert isinstance(new_pipeline, Pipeline)
```

--------------------------------

### Efficient Task Design: Separated IO and CPU Work (Good)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/AdvancedConcepts.md

This snippet shows the recommended pattern for separating IO-bound and CPU-bound work into distinct functions (`get_data` and `process_data`). By processing data in a separate, potentially multi-process task, the GIL is released, improving concurrency and overall pipeline efficiency. The `multiprocess=True` argument ensures CPU-bound work can run in parallel.

```python
def get_data(endpoint: str):
    # IO-bound work
    r = requests.get(endpoint)
    data = r.json()
    return data["results"]
    
def process_data(data):
    # CPU-bound work
    return ...

pipeline = (
    task(get_data, branch=True, workers=20)
    | task(process_data, workers=10, multiprocess=True)
)
```

--------------------------------

### Creating AsyncPipeline from Async Function - Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/ComposingPipelines.md

This snippet illustrates the creation of an `AsyncPipeline` instance by wrapping an asynchronous function `func` with `pyper.task`. It asserts that the resulting object is indeed an `AsyncPipeline`, highlighting Pyper's mechanism for handling asynchronous tasks.

```Python
from pyper import task, AsyncPipeline

async def func():
    return 1

assert isinstance(task(func), AsyncPipeline)
```

--------------------------------

### Building and Analyzing Chess Game Data with Pandas (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/Examples/ChessDataAnalysis.md

This function takes an iterable of game detail dictionaries and transforms them into a pandas DataFrame. It filters the data to include only games where the player played as White, then groups the data by chess opening to calculate the total number of games and the average score for each opening. The resulting DataFrame is sorted by the total number of games in descending order.

```Python
def build_df(data: typing.Iterable[dict]) -> pd.DataFrame:
    df = pd.DataFrame(data)
    df = df[df["color"] == 'W']
    df = df.groupby("opening").agg(total_games=("score", "count"), average_score=("score", "mean"))
    df = df.sort_values(by="total_games", ascending=False)
    return df
```

--------------------------------

### Consuming Pyper Pipeline Outputs with the Greater Than Operator

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/ComposingPipelines.md

This snippet demonstrates the `>` operator in Pyper, which pipes a pipeline's output directly into a consumer function, returning a callable that handles the 'run' operation. It also shows the equivalent `Pipeline.consume` method for directing pipeline results to a consumer.

```Python
if __name__ == "__main__":
    run = (
        task(step1, branch=True)
        | task(step2)
        > JsonFileWriter("data.json")
    )
    run(limit=10)
    # OR
    run = (
        task(step1, branch=True).pipe(
        task(step2)).consume(
        JsonFileWriter("data.json"))
    )
    run(limit=10)
```

--------------------------------

### Aggregating Streamed Inputs with `join` Parameter in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/task.md

The `join` parameter controls how a consumer task receives inputs from the previous task. When `join` is `False`, each individual output is processed separately. When `join` is `True`, the consumer task receives a stream of inputs, allowing for aggregation or batch processing.

```python
from typing import Iterable
from pyper import task

def create_data(x: int):
    return [x + 1, x + 2, x + 3]

def running_total(data: Iterable[int]):
    total = 0
    for item in data:
        total += item
        yield total

if __name__ == "__main__":
    pipeline = (
        task(create_data, branch=True)
        | task(running_total, branch=True, join=True)
    )
    for output in pipeline(0):
        print(output)
        #> 1
        #> 3
        #> 6
```

--------------------------------

### Composing pyper.Pipeline Objects with | Operator in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/Pipeline.md

This snippet shows the `__or__` method, which overloads the `|` operator. This allows users to use the `|` operator as syntactic sugar for the `Pipeline.pipe` method, providing a more concise way to compose pipelines.

```python
def __or__(self, other: Pipeline) -> Pipeline:
```

--------------------------------

### Composing pyper.Pipeline Objects with pipe in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/Pipeline.md

This snippet defines the `pipe` method, which enables the composition of two `Pipeline` objects. It returns a new `Pipeline` instance that combines the tasks from both original pipelines, facilitating sequential data processing.

```python
def pipe(self, other: Pipeline) -> Pipeline:
```

--------------------------------

### Inefficient Task Design: Combined IO and CPU Work (Bad)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/AdvancedConcepts.md

This snippet demonstrates an inefficient task design where IO-bound (network request) and CPU-bound (data processing) operations are combined within a single `get_data` function. This approach can harm concurrency by holding the Global Interpreter Lock (GIL) during CPU-intensive work, even when the network request is handled concurrently.

```python
def get_data(endpoint: str):
    # IO-bound work
    r = requests.get(endpoint)
    data = r.json()
    
    # CPU-bound work
    for item in data["results"]:
        yield process_data(item)

pipeline = task(get_data, branch=True, workers=20)
```

--------------------------------

### Attaching Consumer Functions to pyper.Pipeline with > Operator in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/Pipeline.md

This snippet defines the `__gt__` method, which overloads the `>` operator. This operator serves as syntactic sugar for the `Pipeline.consume` method, offering a more intuitive and concise syntax for attaching consumer functions to a pipeline.

```python
def __gt__(self, other: Callable) -> Callable:
```

--------------------------------

### Memory-Efficient Data Generation with Python Generators (Good)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/UserGuide/AdvancedConcepts.md

This snippet demonstrates the use of a Python generator function (`generate_values_lazily`) for lazy execution. By using the `yield` keyword, the function returns values one by one as they are requested, rather than allocating all 10 million values in memory at once. This approach is crucial for processing large datasets efficiently and preventing out-of-memory errors.

```python
import typing
from pyper import task

# Okay
def generate_values_lazily() -> typing.Iterable[dict]:
    for i in range(10_000_000):
        yield {"data": i}
```

--------------------------------

### Configuring Pyper Task for Multiprocessing (Python)

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/task.md

This snippet demonstrates how to configure a `pyper` task to run synchronous, CPU-bound functions using `multiprocessing.Process` workers by setting `multiprocess=True`. It shows a `slow_func` being processed in a pipeline with 20 workers, benefiting from parallel execution. Note that asynchronous tasks cannot use multiprocessing, and normal Python pickling restrictions apply to functions, arguments, and return values.

```python
from pyper import task

def slow_func(data: int):
    for i in range(1, 10_000_000):
        i *= i
    return data

if __name__ == "__main__":
    pipeline = (
        task(range, branch=True)
        | task(slow_func, workers=20, multiprocess=True)
    )
    for output in pipeline(20):
        print(output)
```

--------------------------------

### Attaching Consumer Functions to pyper.Pipeline in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/Pipeline.md

This snippet presents the `consume` method, designed to attach a consumer function to a `Pipeline`. This function will process the final output of the pipeline, allowing for actions like logging, saving, or further processing of the results.

```python
def consume(self, other: Callable) -> Callable:
```

--------------------------------

### Controlling Task Output Branching with `branch` Parameter in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/task.md

This snippet illustrates the effect of the `branch` parameter. When `branch` is `False`, the task returns a single value. When `branch` is `True`, the task is expected to return an `Iterable` or `AsyncIterable`, with each item from the iterable being treated as a separate output.

```python
from pyper import task

def create_data(x: int):
    return [x + 1, x + 2, x + 3]

if __name__ == "__main__":
    pipeline1 = task(create_data)
    for output in pipeline1(0):
        print(output)
        #> [1, 2, 3]

    pipeline2 = task(create_data, branch=True)
    for output in pipeline2(0):
        print(output)
        #> 1
        #> 2
        #> 3
```

--------------------------------

### Controlling Task Output Queue Size with `throttle` in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/task.md

The `throttle` parameter manages the maximum size of a task's output queue, preventing memory exhaustion when a fast producer generates data much quicker than a slow consumer can process it. This mechanism pauses the producer until the consumer catches up, ensuring efficient memory usage.

```python
import time
from pyper import task

def fast_producer():
    for i in range(1_000_000):
        yield i

def slow_consumer(data: int):
    time.sleep(10)
    return data

pipeline = (
    task(fast_producer, branch=True, throttle=5000)
    | task(slow_consumer)
)
```

--------------------------------

### Configuring Concurrent Task Execution with `workers` in Python

Source: https://github.com/pyper-dev/pyper/blob/main/docs/src/docs/ApiReference/task.md

The `workers` parameter specifies the number of concurrent workers that will execute the task. Setting `workers` greater than 1 enables parallel or concurrent processing, which can significantly improve performance for CPU-bound or I/O-bound tasks.

```python
import time
from pyper import task

def slow_func(data: int):
    time.sleep(2)
    return data

if __name__ == "__main__":
    pipeline = task(range, branch=True) | task(slow_func, workers=20)
    # Runs in ~2 seconds
    for output in pipeline(20):
        print(output)
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.