Pipeable Operator Gives No Explanation of Issue about aiostream HOT 2 CLOSED

Hi @t-mart, thanks for your report :)

I tried to reproduce your problem, and I did get a proper exception out of it. For instance, consider the following code:

import asyncio
from aiostream import operator


@operator(pipable=True)
async def some_pipable_operator(source, arg):
    await asyncio.sleep(1)


async def main():
    some_pipable_operator("test")

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

Just like in your snippet, regenerate is incorrectly instantiated:

• the first argument is not an async iterable
• the second argument is missing

In this case it's the first error that is reported first:

Traceback (most recent call last):
  File "test.py", line 15, in <module>
    asyncio.run(main())
  File "/home/vinmic/miniconda/lib/python3.7/asyncio/runners.py", line 43, in run
    return loop.run_until_complete(main)
  File "/home/vinmic/miniconda/lib/python3.7/asyncio/base_events.py", line 584, in run_until_complete
    return future.result()
  File "test.py", line 12, in main
    some_pipable_operator("test")
  File "/home/vinmic/aiostream/aiostream/core.py", line 284, in init
    assert_async_iterable(args[0])
  File "/home/vinmic/aiostream/aiostream/aiter_utils.py", line 61, in assert_async_iterable
    f"{type(obj).__name__!r} object is not async iterable")
TypeError: 'str' object is not async iterable

For your information, the proper usage of user-defined pipable operator is:

import asyncio
from aiostream import stream, operator


@operator(pipable=True)
async def some_pipable_operator(source, arg):
    assert await source == "a"
    yield arg


async def main():
    result = await (stream.just("a") | some_pipable_operator.pipe("b"))
    assert result == "b"

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

However, it seems to me that you don't need a user-defined pipable operator. Instead, you may simply use stream.map:

from functools import partial
from aiostream import stream, pipe

async def watch(app: web.Application) -> None:
    await (
        stream.merge(*[awatch(p) for p in watch_paths]) 
        | pipe.map(partial(regenerate, app=app), task_limit=1)
    )

async def regenerate(
    changedset: Set[Tuple[Change, str]], app: web.Application
) -> None:
    [...]

Hope that helps :)

from aiostream.

Fantastic. Thank you for your help.

I wasn't getting the exception because, while I did create_task all this code, I wasn't awaiting it, and that maybe caused the exception to not bubble up properly. (It's amazing it was running at all. Maybe aiohttp application.on_startup was helping me in some way I don't understand.) EDIT: Or it was because I didn't debug=True the loop...no matter. It's not important. For many Python programmers, thinking asynchronously will take some getting-used-to.

Sorry to stuff up your issues with a non-issue. I'd definitely be in favor of a help channel of some kind, whether that be gitter or IRC or discord etc.

from aiostream.