Indeed - but then maybe we could aim for a stdlib addition of something similar - maybe even via PEPifying a standard behavior, implementing that in a published lib, and them go for adoption.
It turns out I am halfway across a “Pipe” class, which emerged here a couple months ago - but aiming at different things: a way to control(and throtle) the parallelization of asyncio tasks -
The proponent (@missing-stardust at Add task pipeline to asyncio with capped parallelism and lazy input reading ) problem was not about having a nice syntax - but actually a nice pattern to do that,
It turns out I have a nive “utility box for asyncio” package that is a lot less toyich than that linked code, meaning: this one is published, and I’d use it in prod: GitHub - jsbueno/extraasync: Extra utilities for Python asynchronous programming
So, after some invitation, Matin did contribute his code, I started the fiddling around it, got his contributions - and them had to pause due to (life) - I came back a couple weeks ago, and decided for a different approach than his initial nested-function based one. And, having a class for the Pipe will allow me to do all the niceties customizing (existing) operators such as seen in this thread.
One of Martin’s core ideas is to be agnostic to the source being either a sync or asynchronous generator. And them, comes one pattern he found missing: the ability to parallelize / throttle the item generation in this source itself, which is orthogonal to having operators.
(Indeed, Python can always build a readable, very nice pipeline just adding a sequence of callables as args to an orchestrator - readable enough, without even needing the operator overriding part)
Feel free to join the discussion in the project there - I will likely resume working on it this week.