corstream 0.1.0a1

A coroutine composition framework for declarative async pipelines in Python

CorStream

---

A coroutine composition framework for Python that enables declarative, streaming-style async pipelines.

📝 Table of Contents

• About
• Getting Started
• Deployment
• Usage
• Built Using
• TODO
• Contributing
• Authors
• Acknowledgments

🧐 About

CorStream is a Python library that lets you build elegant, composable pipelines using asynchronous iterables and coroutine-based operators. Think of it like a blend of asyncio, RxPy, and Unix pipes — but designed for readability, type safety, and modern Python development.

With CorStream, you can easily:

• Transform and filter async data flows
• Batch, throttle, or log stream data
• Apply async functions with concurrency control
• Collect or reduce outputs with simple syntax

🏁 Getting Started

These instructions will get you a copy of the project up and running on your local machine for development and testing purposes.

Prerequisites

You’ll need Python 3.9 or higher and Poetry installed:

python3 --version
# Should be 3.9+

curl -sSL https://install.python-poetry.org | python3 -

Installing

Clone the repo and install dependencies with Poetry:

git clone https://github.com/shivkun/corstream.git
cd corstream
poetry install

CorStream is also published on PyPI, you can install it directly with:

pip install corstream
# or use Poetry
poetry add corstream

You can now run tests, examples, or start building pipelines!

🔧 Running the tests

To run all tests:

poetry run pytest

Break down into end-to-end tests

Each operator and sink has its own test file under tests/.

Example:

tests/test_map.py          # Tests for .map()
tests/test_batch.py        # Tests for .batch()
tests/test_to_list.py      # Tests for .to_list()

And coding style tests

CorStream follows strict linting and formatting with black, mypy, and ruff:

poetry run black corstream/
poetry run ruff check corstream/
poetry run mypy corstream/

🎈 Usage

Here's a simple pipeline:

from corstream import Stream

async def get_email(user_id: int) -> str:
    return f"user{user_id}@example.com"

async def send_batch(batch: list[str]):
    print("Sending:", batch)

await (
    Stream
    .from_iterable(range(1, 11))
    .filter(lambda x: x % 2 == 0)
    .map_async(get_email, max_concurrency=3)
    .batch(5)
    .log("batch")
    .for_each(send_batch)
)

⛏️ Built Using

• Python 3.9+
• Poetry
• AsyncIO
• Pytest

✍️ Authors

• @shivkun — Design & Implementation

See also the list of contributors.

🎉 Acknowledgements

• Inspiration from functional programming and reactive streams
• Thanks to the maintainers of asyncio, RxPy, and toolz