Mypy fails when using an async function for `map`

[lindycoder]

Hello,

Just started looking into this tool, it's great! This might be an error/misunderstanding on my part, if so i'm sorry.

Issue

I tried to naively map an async function on a Stream, but i get this error on the line | pipe.map(consumer)

error: Argument 1 has incompatible type "Callable[[str, VarArg(Any)], Coroutine[Any, Any, str]]"; expected "Callable[[str, VarArg(str)], Awaitable[Never]]"  [arg-type]

Here's a minimal reproduction code

from collections.abc import AsyncGenerator
from typing import Any

from aiostream import stream, pipe, Stream


async def producer() -> AsyncGenerator[str, None]:
    yield "a"

async def consumer(value: str, *_: Any) -> str:
    return value + "!"

iterator: Stream[str] = (
    stream.iterate(producer())
    | pipe.map(consumer)
)

Fixes

If i use amap it works
If I make consumer sync, it works (but I can't, just wanted to say the typing is accurate for sync functions)

Notes

Sure i can use amap but it seems like map is suppose to handle both to make the code simpler, which is a nice idea, but it looks like a typing issue.

Versions / config

Versions:

• Python 3.12.8
• mypy 1.15.0
• aiostream 0.6.4

[tool.mypy]
namespace_packages=true
explicit_package_bases=true
strict=true
show_error_codes=true

[vxgmichel]

Hello @lindycoder and thanks for the report!

First thing to note is that aiostream is no longer compatible with mypy, since mypy 1.7: #105

However, the example you posted does not pass the typing checks using mypy 1.6 either, although it does pass when using the latest pyright instead.

I would like to fix the typing for mypy but it turns out this is really hard to do for a project like aiostream. Most of the API predates the support for type hints, which means it hasn't been designed to ease the work of the type checkers. In particular, taking an argument which is either a generic async callback or a generic sync callback is really hard to support. I also did my best to type the operator decorator correctly by defining a custom protocol like so:

aiostream/aiostream/core.py

Lines 255 to 287 in 8201a79

	class Operator(Protocol[P, T]):
	def __call__(self, *args: P.args, **kwargs: P.kwargs) -> Stream[T]: ...
	@staticmethod
	def raw(*args: P.args, **kwargs: P.kwargs) -> AsyncIterator[T]: ...
	class PipableOperator(Protocol[A, P, T]):
	def __call__(
	self, source: AsyncIterable[A], /, *args: P.args, **kwargs: P.kwargs
	) -> Stream[T]: ...
	@staticmethod
	def raw(
	source: AsyncIterable[A], /, *args: P.args, **kwargs: P.kwargs
	) -> AsyncIterator[T]: ...
	@staticmethod
	def pipe(
	*args: P.args, **kwargs: P.kwargs
	) -> Callable[[AsyncIterable[A]], Stream[T]]: ...
	class SourcesOperator(Protocol[P, T]):
	def __call__(self, *args: P.args, **kwargs: P.kwargs) -> Stream[T]: ...
	@staticmethod
	def raw(*args: P.args, **kwargs: P.kwargs) -> AsyncIterator[T]: ...
	@staticmethod
	def pipe(
	*args: P.args, **kwargs: P.kwargs
	) -> Callable[[AsyncIterable[Any]], Stream[T]]: ...

but it still has a couple of issues and limitations.

Sorry I cannot help more than that. If you ever find a way to improve the typing in this project, contributions are more than welcome :)