Не шарю в асинхронности и очереди в python, что читать, смотреть?

Question

[fotis06]

Написал скрипт на python, решил, что он может понадобиться кому-то ещё, запихал его в fastapi приложение. Проблема в том, что сам скрипт требует около 2-3 минут на выполнение и с несколькими пользователями одновременно это всё точно не будет работать. Что вы можете посоветовать, чтобы разобраться во всём этом?

Answer 1

[chemdev]

Если выполняется так долго, я бы сделал следующим образом:
Создал бы приложение celery с задачей, сделал бы ручку апи, которая запускает задачу и отдает ссылку на результат, ссылка на результат отдает либо "еще выполняется", либо результат.
Либо, вместо ссылки на результат, можно отправлять его на email.
Я бы написал docker-compose.yml с контейнерами api, nginx, celery, redis, но можно развернуть все любым удобным способом, это не обязательно.

Comments

[Everything_is_bad]

какой-то докер головного мозга, часто достаточно просто вызывать create_task
и celery тоже часто оверхед, есть куча более простых аналогов, а тут еще и asyncio

[chemdev]

Everything_is_bad, как я уже упомянул в ответе:

но можно развернуть все любым удобным способом, это не обязательно

Просто когда нужно, чтобы сообща работали 3-4 процесса, докер очень удобно лично мне.

[fotis06]

chemdev, прикрутил celery, все работает как я и хотел, осталось разобраться в докере. Спасибо!

Answer 2

[Everything_is_bad]

Что вы можете посоветовать, чтобы разобраться во всём этом?

начни с чтение оф доков по библиотеки asyncio

Answer 3

[alekssamos]

В этом коде задачи будут добавляться в очередь без ожидания и выполняться по одной.

import logging
import asyncio

q = asyncio.Queue()


async def a_difficult_task(some_arg):
    print(f"task {some_arg} started")
    await asyncio.sleep(120)
    print(f"task {some_arg} ended")


async def worker():
    while True:
        try:
            some_arg = await q.get()
            await a_difficult_task(some_arg=some_arg)
        except (KeyboardInterrupt, SystemExit, SystemError):
            break
        except Exception:
            logging.exception("...")
        finally:
            q.task_done()


async def main():
    asyncio.create_task(worker())
    print("Adding tasks to queue")
    for x in range(1, 7):
        await q.put( x )
    print("Tasks added to queue")
    while True:
        await asyncio.sleep(1)


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