How do I cancel asyncio coroutines on pressing the button again?

Question

How could I cancel tasks that are already running when I click the button again?

The code I've sketched out, but it certainly doesn't work

class TasksManager:
    def __init__(self) -> None:
        self.state = False
        self.list = [self.task_1, self.task_2, self.task_3]
 
    async def task_1(self):
        for i in range(10):
            await asyncio.sleep(1)
            print("Task 1: ", i)
    
    async def task_2(self):
        for i in range(10):
            await asyncio.sleep(1)
            print("Task 2: ", i)
 
    async def task_3(self):
        for i in range(10):
            await asyncio.sleep(1)
            print("Task 3: ", i)
 
    async def click(self, e: ft.TapEvent):
        if not self.state:
            self.state = True
            list = [asyncio.create_task(task()) for task in self.list]
            done, pending = await asyncio.wait(list, return_when=asyncio.FIRST_COMPLETED)
            print(done, pending)
        else:
            # здесь нужно отменить корутины
            for task in pending:
                task.cancel()
 
def main(page: ft.Page):
    manager = TasksManager()
    button = ft.TextButton("Click", on_click=manager.click)
    page.add(button)
    page.update()
 
ft.app(main)

Tasks must be forcibly stopped

Answer 1

While more details would be helpful, there are several issues in click. However, I'm not familiar with Flet.

1. When you create a task, you need to save it off. This is mentioned in Python's documentation for asyncio.create_task:

   Important: Save a reference to the result of this function, to avoid a task disappearing mid-execution. The event loop only keeps weak references to tasks. A task that isn’t referenced elsewhere may get garbage collected at any time, even before it’s done. For reliable “fire-and-forget” background tasks, gather them in a collection:

2. You are calling asyncio.wait, which will block until the tasks are complete or canceled. This is where more details about Flet and the use case are important. Is click launched in its own task? If not, two calls to click will wait for the first one to complete before the second one is executed.

3. You are not saving the tasks inside the class, so there is no way for subsequent calls to click to cancel the tasks created in prior executions.

---

Try something like this:

class TaskManager:
    def __init__(self) -> None:
        self.state = False
        self.__tasks_to_create = [self.task_1, self.task_2, self.task_3]
        self.__tasks: list[asyncio.Task] = []
 
    ...
 
    async def click(self, e: ft.TapEvent):
        if not self.state:
            self.state = True
            self.__tasks = [asyncio.create_task(task()) for task in self.__tasks_to_create]

    async def cancel_tasks(self, e: ft.TapEvent):
        for task in self.__tasks:
            task.cancel()

        self.__tasks = []
 
def main(page: ft.Page):
    manager = TaskManager()
    button = ft.TextButton("Click", on_click=manager.click)
    cancel_button = ft.TextButton("Cancel", on_click=manager.cancel_tasks)
    page.add(button)
    page.add(cancel_button)
    page.update()
 
ft.app(main)