Python asyncio list does not processing asyncrounouslly because of what?

Question

[Михаил Марков]

А есть мысли, почему вот такой код

async def main(api_id, api_hash, tg_session, chat_id, db_session):
    async with TelegramClient(StringSession(tg_session), api_id, api_hash) as client:
        await client.start()
        chan = await client.get_entity(int(chat_id))
        events = await client.get_admin_log(chan)
        print("start for loop")

        for event in events:
            print(event.user_id)
            await asyncio.sleep(5)
            print(f"{event.user_id} awaited")
            continue

asyncio.run(main(api_id, api_hash, tg_session, parsing_chat_id, Session))

даёт вот такой вывод?
start for loop
6550133864
6550133864 awaited
5171756347
5171756347 awaited
2002370110
2002370110 awaited
1682226622

Для сравнения, нормальный пример нормального кода

async def bar(item: int) -> int:
    print("Started")
    await asyncio.sleep(5)
    print("Finished")
    return item ** 2

async def foo():
    items = range(1, 10)
    tasks = [bar(item) for item in items]
    new_items = await asyncio.gather(*tasks)
    return new_items

if __name__ == '__main__':
    results = asyncio.run(foo())
    print(results)

выводит нормальный вывод:
Started
...
Started
Finished
...
Finished
[1, 4, 9, 16, 25, 36, 49, 64, 81]

Comments

[Михаил Кростелев]

не совсем понятно что смущает.
Что вы ожидали от первого кода?

start for loop
6550133864
5171756347
2002370110
6550133864 awaited
5171756347 awaited
2002370110 awaited

Что-то такое?

[Михаил Марков]

Михаил Кростелев, да, естественно

Answer 1

[Михаил Марков]

Вопрос говно, питон говно, я искал подобный go синтаксис с вейтгруппами, а нашёл вот что:

# Python 3.11+ ONLY!
async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(some_coro(...))
        task2 = tg.create_task(another_coro(...))
    print("Both tasks have completed now.")

вот тут: https://stackoverflow.com/a/42246632
Вопрос можно закрывать за неимением ответа в дискуссии.

Answer 2

[Сергей Соловьев]

В первом случае ты грубо говоря все запускаешь синхронно: синхронно итерируешься по списку, await'ишь - все последовательно

Во втором случае, ты запускаешь множество асинхронных генераторов. Каждый запускается, пишет Started, достигает await и управление переходит следующему генератору. Т.е. получается - сначала все доходят до await (печатают Started), после ставятся в конец и, когда все Started уже написали, начинается исполнение того, что после await

Comments

[Михаил Марков]

Всё отлично, Вы всё понимаете, явно работаете с асинхронным кодом и очень хорошо понимаете все концепции asyncio.

А скажите, почему я из такой штуки

async def handle_event(event):
    print(event.user_id)
    await asyncio.sleep(5)
    print(f"{event.user_id} awaited")

def main():
    events = await client.get_admin_log(chan)
    print("start for loop")

    for event in events:
        asyncio.create_task(handle_event(event))

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

получаю только

start for loop
292054100
419898034
694851875
743258597
...

без ...743258597 awaited?

Т.е. цикл завершается, а вместе с ним завершается и программа

[Сергей Соловьев]

asyncio.create_task создает таску - это такая штука, которая не блокирует работу. Т.е. ее можно запустить и приложение продолжит работать.

В данном случае, создаются таски, они отрабатывают до первого await, и на этом моменте приложение закрывается. Созданные таски - не заставляют приложение ждать их окончания (они засыпают на 5 секунд). Можешь добавить условный sleep(10) в конце, чтобы проверить

[Михаил Марков]

Да, но вот в чём дело: мне надо именно подождать, пока все таски закончат своё выполнение.
А ещё лучше не создавать отдельную функцию, но это мелочи.
На самом деле, я ждал этой функциональности от функции
loop.run_until_complete(main())

[Сергей Соловьев]

зачем тогда create_task ? здесь нужен gather

def main():
    events = await client.get_admin_log(chan)
    print("start for loop")

   
    await asyncio.gather(handle_event(event))

[Михаил Марков]

Сергей Соловьев, если имелось в виду

events = await client.get_admin_log(chan)
print("start for loop")

for event in events:
    await asyncio.gather(handle_event(event))

то у меня ничего не вышло, он как писал последовательно
566534996
566534996 awaited
раз в 5 секунд, так и пишет

[Михаил Марков]

events = await client.get_admin_log(chan)
print("start for loop")
tasks = [handle_event(event) for event in events]
await asyncio.gather(*tasks)

есть вот такое решение, но из-за оператора * и распаковки списка как аргумент функции, я не думаю, что этот код можно считать хорошим, читаемым, очевидным или понятным. Может быть, есть ещё один способ?

[Сергей Соловьев]

Михаил Марков,

events = await client.get_admin_log(chan)
print("start for loop")
await asyncio.gather(handle_event(event) for event in events)

[Сергей Соловьев]

Михаил Марков, он хороший, читаемый, очевидный и понятный

[Михаил Марков]

Сергей Соловьев, даже не знаю, что Вам сказать...

RuntimeError: Task got bad yield: <coroutine object handle_event at 0x7f9e576c12a0>

и такая штука

await asyncio.gather([handle_event(event) for event in events])

тоже выдаёт
TypeError: unhashable type: 'list'

Нет, в принципе, код со списком и звёздочкой работает, хотя я и ожидал другого. На SO говорят, что моя проблема решена с питона версии 3.11, а это, можно сказать, неприемлемый способ решения. Спасибо за попытку.

[Сергей Соловьев]

Михаил Марков, зачем [] ?

[Михаил Марков]

Сергей Соловьев, ну типа там можно вот так вот сделать

await asyncio.gather(*[handle_event(event) for event in events])

но по мне так ещё хуже

[Сергей Соловьев]

Михаил Марков, почему не так?

await asyncio.gather(handle_event(event) for event in events)

[Михаил Марков]

RuntimeError: Task got bad yield: <coroutine object handle_event at 0x7f9e576c12a0>

[Сергей Соловьев]

Михаил Марков, ну тогда

await asyncio.gather(*[handle_event(event) for event in events])

Answer 3

[Antonio Solo]

думаю что async for поможет

async for event in client.iter_admin_log(chan):
      print(event.user_id)
      await asyncio.sleep(5)
      print(f"{event.user_id} awaited")

https://superfastpython.com/asyncio-async-for/

Comments

[Михаил Марков]

Нет, это так не работает:

start for loop
419485747
419485747 awaited
1601871914
1601871914 awaited

Но спасибо за попытку