Ultimately I want to pass this into asyncio.gather() but I'm stuck on how.
#🔒 How do you transform an async generator into a coroutine?
granite frost
amber kernelBOT
Ultimately I want to pass this into `asyncio.gather()` but I'm stuck on how.
@granite frost
Python help channel opened
wheat turret
Wrap it in a coroutine that collects it
!pip asyncstdlib has some functions that may be useful.
amber kernelBOT
wheat turret
Example of a function: ```py
async def alist(aiter: AsyncIterable[T]) -> list[T]:
return [item async for item in aiter]
it collects an async iterator (or generator) to a list
granite frost
I think this worked:
from asyncstdlib import list as alist
async def async_generator_to_coroutine(async_generator):
return alist(async_generator)
tasks.append(async_generator_to_coroutine(item))
You need to await alist
granite frost
i think tasks.append should handle that
wheat turret
no, it will make the task return a coroutine
granite frost
i mean results = await asyncio.gather(*tasks)
wheat turret
Why not just append alist(item) directly?
granite frost
i think i'm just realizing that now
how can you tell if something is a async_generator specifically
wheat turret
it's marked async and has yield
the type checker knows. hover over it or pass it to typing.reveal_type
You can also use inspect.isasyncgenerator(func)
!d inspect.isasyncgenfunction
amber kernelBOT
silk storm
this whole question seems odd; what is the actual use case?
wheat turret
yeah, using inspect is odd for adding tasks.
if you await a coroutine and get a coroutine back, you fucked up
Can you share all your code?
!paste
amber kernelBOT
granite frost
high level i have implemented some utility functions for mangaging concurrency by way of Promise.map_series, Promise.reduce_series and Promise.parallel
which lets me do stuff like:
tasks.append(asyncio.create_task(Promise.series([
Promise.parallel([
Promise.series([
lambda subproduct: fetch(session, "get", "/api/sub-product/" + subproduct["id"]),
lambda data: data.json()
], subproduct),
that looks more JS than Python
granite frost
yes
silk storm
I come from a very different school of thought in this matter; not sure I can help
granite frost
well i think my core issue is fixed with alist
now I just need a condition for identifying a async_generator_asend
so instanceof? probably?
granite frost
async_generator_asend
silk storm
it's not something you can import
wheat turret
try the types module
wheat turret
Why do you need to check the type?
silk storm
is that Promise type your code?
granite frost
yes
granite frost
Why do you need to check the type?
to conditionally wrap the async_generator_asend in alist() if it's an async generator
wheat turret
Why would it not return an async generator?
granite frost
instead of async_generator_asend you mean?
wheat turret
async def will_always_be_async_generator():
if False:
yield
That will always be a generator. It's just empty.
silk storm
sorry to say but this looks to me like you're making this unnecessarily complicated
wheat turret
Can you share your code so we can get a better idea of what you're actually trying to do?
silk storm
yes please
wheat turret
^
silk storm
this doesn't yet answer the question of what you're actually trying to accomplish
wheat turret
just the lambdas
silk storm
?
wheat turret
I see no generators in your code
granite frost
comes from the .__anext__() hack to get async lambdas (https://stackoverflow.com/a/66330279)
wheat turret
I think you should reconsider what you're doing just for the DX
granite frost
i'm thoroughly enjoying the dx 🙂
i could not use lambdas and write actual functions and just pass their names in
wheat turret
calling __anext__() doesn't return a generator.
granite frost
oh maybe the for _ in "_" bit then
wheat turret
yes, that creates a generator
the object you call anext on is a generator
specifically, it's async _ in ...
asyncstdlib has a function to convert a regular iterable into an async iterable
granite frost
so i could get rid of the dumb hack?
granite frost
i think i'll still have the problem of needing to use the await keyword within the lambdas which isn't allowed
but yeah to your point i could just not use lambdas
wheat turret
async def do_thing(session):
foos = await sdk.foo.get_all_foos(session)
for foo in foos:
bars = await sdk.bar.get_all_bars_by_foo_name(session, foo["name"])
for bar in bars:
results = await acsdk.bar.delete_bar_by_id(session, bar["id"])
print(results)
async def main():
session = init_session(api_key)
await do_thing(session)
you could even make this a list comprehension if you wanted
granite frost
sure but now it's all in series
wheat turret
you never called parallel anyway
granite frost
i do in other places in the code
wheat turret
The way you're calling parallel isn't how generators work
silk storm
here you want the for foo in foos part to work in parallel, right?
granite frost
i'm mostly just handling coroutines, i'm only using generators as a part of the async lambda hack
wheat turret
You'll probably want to replace inspect.isasyncgenfunction with inspect.isasyncgen
generator functions return generators
wheat turret
don't use __anext__
granite frost
ah! using __aiter__ may have done it
wheat turret
You can also do hasattr(item, "__anext__") to check if something is an async iterator
silk storm
I don't think I'm any closer to understanding what this enormous hack intends to accomplish
wheat turret
Also hasattr(item, "__aiter__")
granite frost
python doesn't really seem to have a good answer to managing concurrency, that's the problem
wheat turret
async iterators are also async iterables
silk storm
@granite frost that's a really broad topic; what solutions have you researched?
have you tried asyncio.gather() or asyncio.TaskGroup yet?
in short, what are you missing?
wheat turret
Try doing this. ```py
for item in iterable:
if inspect.isasyncgen(item):
tasks.append(alist(item))
else:
tasks.append(item)
yep that's what i have
wheat turret
btw lambdas can't be generators
your lambda with get_all_foos is actually returning tuple[Session, generator_send]
granite frost
yeah what is a generator_send
silk storm
I'm bugging out since I'm not getting any answers
wheat turret
I'm curious how you were able to call __anext__ on a non-async generator
granite frost
silk storm
what I've been asking is what you're doing these things for?
wheat turret
ok, simply awaiting something inside the generator will make it async
wheat turret
I think they're making a toy
silk storm
toy?
wheat turret
for fun
silk storm
eh
granite frost
it's for interacting with an api, it's a whole bunch of fetch requests
wheat turret
Just use a normal async function
granite frost
yeah
wheat turret
you're adding unneeded complexity
wheat turret
!pip aiolimit has some useful functions
amber kernelBOT
granite frost
im guessing you meant itertools
amber kernelBOT
wheat turret
neat
granite frost
cheers! thanks for the help!
