zh217/aiochan

CSP-style concurrency for Python

aiochan's Issues

FYI: pygolang

Hello up there.

No bug on aiochan side found but just wanted to let you know that a similar project exists with support for channels and goroutines: pygolang.

Thanks.

Version on pypi doesn't support Python versions >= 3.10

But fixes for 3.10 and above seem to be merged in the repo.
A new release on pypi would be appreciated!

How do I dynamically chain ops

Hi, great library, thanks.

I'm creating a pipeline, where pipeline tasks are user defined and can be dynamic. e.g. below are tasks that user wish to pass to aiochan,

test.py

gen = [1, 2, 3]
download = DownloadSomethingFromInternet() # it contains 'execute' method
batchify = ("group", 2) # group 2 downloads
proceess = SomeProcessing() # it contains 'execute' method

p = AioChanPipeline.run(gen, [download,  batchify,  process])

for d in p:
   ....

server.py

class AioChanPipeline:

    async def run(self, gen, stages):
        pipeline_stream = ac.from_iter(gen)

        for stage in stages:
            if isinstance(stage, tuple):
                pipeline_stream = getattr(pipeline_stream, stage[0])(**stage[1])
            else:
                pipeline_stream = pipeline_stream.async_pipe_unordered(stage.execute)

        async for data in pipeline_stream:
            yield data

Im running above test.py using @pytest.mark.asyncio and it works If I exclude last process stage. I think after getattr(pipeline_stream, stage[0])(**stage[1]) call something went wrong.

so do you any way to pipe operations without method call?

Apply type hints

What do you think about adding type hints for channels for better control?