Am I managing asyncio tasks (python 3.9) in a proper way?

Question

I'm developing a scraper (python 3.9), I need to launch make_attempt() and, depending on the results of its execution, to launch additional tasks that should work concurrently.

First I create initial task and add it to a list where I store all asyncio tasks: self.data["worker"]["tasks"] Then I launch: await asyncio.gather(*self.data["worker"]["tasks"])

In make_attempt() I wait for results of POST-request to a server (I use aiohttp client), and, based on the result, I either add new tasks, or repeat make_attempt() after a small delay.

I stop the current task and remove it from the list with asyncio tasks, and then I add new tasks.

async def make_attempt(self):
    attempt: int = self.data["res"]["attempt"]

    await self.do_something()
    await sleep(1)

    for task in self.data["worker"]["tasks"]:
        print("Task name: %s" % task.get_name())
        if task.get_name() == str(attempt):
            task.cancel()
    self.data["worker"]["tasks"] = [task for task in self.data["worker"]["tasks"] if task.get_name() != str(attempt)]

    if 1 > 0:  # a condition to start make_attempt() again
        self.data["worker"]["tasks"].append(asyncio.create_task(self.make_attempt(), name=attempt))
        await asyncio.gather(*self.data["worker"]["tasks"])

async def run(self):
    self.data["worker"]["tasks"].append(asyncio.create_task(self.make_attempt(), name=self.data["res"]["attempt"]))
    await asyncio.gather(*self.data["worker"]["tasks"])

I'm new to asyncio so maybe you could point me to mistakes or could suggest a better implementation.

Upd. Here is a schema of what I want to achieve:

main_task should be run and if the result is ok, it should launch a couple of instances of another tasks (see loop2). When a result of a task in loop2 is obtained, a new subtask should be run. main_task should wait either for all tasks in loop2 to be finished or a Timeout to be fired. All tasks in loop2 should work concurrently.

UPD 2. This code produces RecursionError: maximum recursion depth exceeded while calling a Python object after about 1000 cycles of execution of the check_base_url() method.

class Scraper:
    def __init__(self, data):
        self.data = data

    async def get_response(self, session, url, method="get", *args, **kwargs) -> Union[
        ClientResponse, None]:
        for _ in range(0, 20):
            if method == "get":
                try:
                    response = await session.get(url, headers={}, proxy="_proxy", *args, **kwargs)
                    if response.status > 399:
                        raise ScraperError(response.status)
                    await sleep(0.1)
                    return response
                except (ClientError, ScraperError) as err:
                    await sleep(0.25)
                    continue
            else:
                try:
                    response = await session.post(url, headers={}, proxy="_proxy", *args, **kwargs)
                    if response.status > 399:
                        raise ScraperError(response.status)
                    await sleep(0.1)
                    return response
                except (ClientError, ScraperError) as err:
                    await sleep(0.25)
                    continue
        return None

    async def get_captcha(self) -> SolvedCaptcha:
        for _ in range(0, 20):
            captcha = await self.task_1()
            if captcha:
                continue

    async def final_task(self, url) -> bool:
        async with ClientSession(cookies={}) as sess:
            resp_step1: Union[ClientResponse, None] = await self.get_response(sess, "url", "post",
                                                                              data={})
            if resp_step1:
                resp_step2: Union[ClientResponse, None] = await self.get_response(sess, "url", "get")
                if resp_step2:
                    captcha: SolvedCaptcha = await self.get_captcha()
                    if captcha:
                        resp_captcha: Union[ClientResponse, None] = await self.get_response(sess, "url",
                                                                                            "post",
                                                                                            data={})
                        if resp_captcha:
                            if 2 > 1:
                                print("FINISHED")
                                return True
                        else:
                            return False
                    else:
                        return False
                else:
                    return False
            else:
                return False

    async def add_task_3(self) -> None:
        if 2 > 1:
            subtasks = [asyncio.create_task(self.final_task(self.data["res"]["slots_urls"][0]))]
            await asyncio.gather(*subtasks)
        else:
            await self.add_task_3()

    def parse(self, html: str, url: str) -> None:
        soup = BeautifulSoup(html, "lxml")
        # do parsing

    async def task_2(self, url) -> bool:
        async with ClientSession() as sess:
            resp: Union[ClientResponse, None] = await self.get_response(sess, url)
            if not resp:
                return False
            html = await resp.text()
            self.parse(html, url)

    async def add_task_2(self) -> None:
        if 2 > 1:
            subtasks = [asyncio.create_task(self.task_2(url)) for url in ["url1", "url2"]]
            await asyncio.gather(*subtasks)

    async def task_1(self) -> bool:
        self.data["res"]["captcha_requested"] += 1
        res = await self.captcha.task_1()
        if not res:
            return False
        return True

    async def add_task_1(self) -> None:
        if 2 > 1:
            subtasks = [asyncio.create_task(self.task_1()) for _ in range(0, 5)]
            await asyncio.gather(*subtasks)

    async def get_calendar_url(self, sess) -> bool:
        resp: Union[ClientResponse, None] = await self.get_response(sess, "url", method="post",
                                                                    data={})
        if not resp:
            return False
        else:
            return True

    async def check_base_url(self) -> bool:
        async with ClientSession() as session_0:
            return await self.get_calendar_url(session_0)

    async def schedule_tasks(self):
        def start_again() -> bool:
            if 2 > 1:
                return True
            return False

        res_base_url: bool = await self.check_base_url()
        if res_base_url:
            tasks = [asyncio.create_task(self.add_task_1()),
                     asyncio.create_task(self.add_task_2()),
                     asyncio.create_task(self.add_task_3())]
            await asyncio.gather(*tasks)
            if start_again():
                await sleep(0.1)
                await self.schedule_tasks()
        else:
            await self.schedule_tasks()

    async def run(self):
        await self.schedule_tasks()

user_10120928 · Answer

asyncio.gather won't update when you add new tasks. You should just create new tasks wherever you need them. Quick rule of thumb everything you aio.gather() no matter wherever you do that will execute in parallel. As an example there will be total of 9 some_async_task running in parallel

async def nested():
    tasks = [some_async_task(), some_async_task(), some_async_task()]
    await aio.gather(tasks)

async def main():
    tasks = [nested(), nested(), nested()]
    await aio.gather(tasks)

This is what you can do:

import asyncio as aio

async def task():
    ...
    await subtask()

async def subtask():
    # anything here
    pass



async def main():
   r = await aio_http_request()
   if should_create_tasks(r):
       tasks = [task(), task(), task()]
       await aio.gather(tasks)
   else:
       await aio.sleep(5)
       await make_attempt()

 
async def run():
   await make_attempt()

Here A, B, C execute concurrently. While Main Task -> Task -> Subtask are sequential.

+------------------------------------------------------------------------+  
|                                                                        |  
|                +-----------------+                                     |  
|                |                 |                                     |  
|                |    Main Task    |                                     |  
|                +-------+---------+                                     |  
|                        |                                               |  
|                        |                                               |  
|                        v                                               |  
|    +-+-------------+---+-------------++------+-------------+--------+  |  
|    | |      A      |   |      B      |       |      C      |        |  |  
|    | |  +-------+  |   |  +-------+  |       |  +-------+  |        |  |  
|    | |  |TASK   |  |   |  |TASK   |  |       |  |TASK   |  |        |  |  
|    | |  +---+---+  |   |  +---+---+  |       |  +---+---+  |        |  |  
|    | |      |      |   |      |      |       |      |      |        |  |  
|    | |      |      |   |      |      |       |      |      |        |  |  
|    | |      |      |   |      |      |       |      |      |        |  |  
|    | |      |      |   |      |      |       |      |      |        |  |  
|    | |  +---+---+  |   |  +---+---+  |       |  +---+---+  |        |  |  
|    | |  |subtask|  |   |  |subtask|  |       |  |subtask|  |        |  |  
|    | |  +-------+  |   |  +-------+  |       |  +-------+  |        |  |  
|    | |             |   |             |       |             |        |  |  
|    | +-------------+   +-------------+       +-------------+        |  |  
|    |                                                                |  |  
|    +----------------------------------------------------------------+  |  
|                                                                        |  
|                                                                        |  
|                                                                        |  
+------------------------------------------------------------------------+

Remember here SUBTASK of A can happen before TASK of B completes because they are not bound (they are concurrent).

user_15194429 · Answer

One approach is to to use asyncio.Queue to manage the pending jobs in your application. You can then create a number of worker tasks that receive jobs from that queue and can also add new jobs.

In the example the actual work (POST request and process some data) is simulated by using asyncio.sleep. Some jobs will produce new jobs while others will not.

Workers will pick up and work on jobs concurrently this way.

The Manager is in charge of creating the job queue and waiting for all items to be processed. After that it will cancel all workers and the program terminates.

Code

import asyncio
import random


class Worker:
    def __init__(self, num, target_q):
        # Worker number
        self.num = num
        # The job queue
        self.target_q = target_q
        # Create asyncio task
        self.task = asyncio.create_task(self.run())

    async def run(self):
        # Work on jobs until task is cancelled
        while True:
            print(f"Worker {self.num}: Waiting for new target")

            # Receive a new job from the queue
            target = await self.target_q.get()
            print(f"Worker {self.num}: Processing target {target}")

            try:
                # Simulating some work (e.g. a POST request)
                await asyncio.sleep(1.0)

                # Depending on the outcome, some new work results
                # 0-2 new targets are generated
                new_target_count = random.randint(0, 2)

                if new_target_count > 0:
                    print(
                        f"Worker {self.num}: Target {target} generating {new_target_count} more targets"
                    )
                    for _ in range(new_target_count):
                        # Create a new random target
                        new_target = random.randint(1, 10000)

                        # Put new targets into queue. This will wait if queue is
                        # currently full.
                        await self.target_q.put(new_target)

            finally:
                print(f"Worker {self.num}: Target {target} done")

                # Decrease queue count by one
                self.target_q.task_done()


class Manager:
    def __init__(self):
        # A common queue holding the jobs for the workers. It just stores
        # integers here but could hold any data.
        self.target_q = asyncio.Queue(10)
        # The list of workers
        self.workers = None

    async def run(self):
        # Create some initial work
        await self.target_q.put(1)

        # Create 3 workers
        self.workers = [Worker(num, self.target_q) for num in range(3)]

        # Wait until queue of unfinished tasks is empty
        await self.target_q.join()

        # Cancel other workers
        for worker in self.workers:
            worker.task.cancel()


def main():
    manager = Manager()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(manager.run())
    loop.close()


if __name__ == "__main__":
    main()

Sample outcome

$ python workers_test.py
Worker 0: Waiting for new target
Worker 0: Processing target 1
Worker 1: Waiting for new target
Worker 2: Waiting for new target
Worker 0: Target 1 generating 2 more targets
Worker 0: Target 1 done
Worker 0: Waiting for new target
Worker 0: Processing target 320
Worker 1: Processing target 5807
Worker 0: Target 320 done
Worker 0: Waiting for new target
Worker 1: Target 5807 done
Worker 1: Waiting for new target

user_1411668 · Answer

I am not sure this is what are you looking for . 1st. You don't have to store the coro. 2nd. Your code on step create_task throws the coro to the async loop without waiting but you call await asyncio.gather again.

if you just want make it run forever or quit on if condition please check out the code below.

async def make_attempt(self):
    await self.do_something()
    await asyncio.sleep(1)
    if 1 > 0:  # a condition to start make_attempt() again
        return await self.make_attempt()

async def run(self):
    await self.make_attempt()

Am I managing asyncio tasks (python 3.9) in a proper way?

3 Answers

Code

Sample outcome