How to run an asynchronous task on class level in Python?

Question

I want to start an asynchronous task/thread that is classed based. Like an class variable, meaning it is accessible to each instance of the class.

I tried to create a separate class which uses asyncio.

class yahoo_connection:

    def __init__(self):
        self.list_of_tick=[]
        self.format="%H   %M   %S"
        self.running=True
        asyncio.create_task(self.const_thread())

    async def const_thread(self):
        while self.running==True:
            print("current list: ",self.list_of_tick)
            if len(self.list_of_tick)!=0:
                idx=np.random.randint(len(self.list_of_tick)+1)
                self.format=self.list_of_tick[idx]
            await asyncio.sleep(3.5)

So this is the class that I planned to initialize as the class thread of my original class. However this code does not work.

I want to able to modify the self.list_of_tick in each instance the class that uses the class yahoo_connection as the class asynchronous thread.

Answer 1

If you want to modify list_of_tick from each asyncio coroutine/task, you have to make it class level field:

import asyncio


class Test:
    storage = []  # class level field

    def __init__(self, name):
        self.name = name  # instance level field

    async def do_work(self):
        """just add some digits to class level field"""
        for i in range(5):
            Test.storage.append(i)
            print(f"{self.name}: {Test.storage}")
            await asyncio.sleep(1)


async def async_main():
    """create three class instances and run do_work"""
    await asyncio.gather(*(Test(f"Cor-{i}").do_work() for i in range(3)))

if __name__ == '__main__':
    asyncio.run(async_main())

Output:

Cor-0: [0] Cor-1: [0, 0] Cor-2: [0, 0, 0] Cor-0: [0, 0, 0, 1] Cor-1: [0, 0, 0, 1, 1] Cor-2: [0, 0, 0, 1, 1, 1] Cor-0: [0, 0, 0, 1, 1, 1, 2] Cor-1: [0, 0, 0, 1, 1, 1, 2, 2] Cor-2: [0, 0, 0, 1, 1, 1, 2, 2, 2] Cor-0: [0, 0, 0, 1, 1, 1, 2, 2, 2, 3] Cor-1: [0, 0, 0, 1, 1, 1, 2, 2, 2, 3, 3] Cor-2: [0, 0, 0, 1, 1, 1, 2, 2, 2, 3, 3, 3] Cor-0: [0, 0, 0, 1, 1, 1, 2, 2, 2, 3, 3, 3, 4] Cor-1: [0, 0, 0, 1, 1, 1, 2, 2, 2, 3, 3, 3, 4, 4] Cor-2: [0, 0, 0, 1, 1, 1, 2, 2, 2, 3, 3, 3, 4, 4, 4]

But if I were your I would not do list_of_tick class level field, I would just keep it outside of class. But it is more my own taste, the both ways are right.

Edit:

I you want to use Test task inside another class you can do smth the following way:

import asyncio


class Test:
    storage = []  # class level field

    def __init__(self, name):
        self.name = name  # instance level field

    async def do_work(self):
        """just add some digits to class level field"""
        for i in range(3):
            Test.storage.append(i)
            print(f"{self.name}: {Test.storage}")
            await asyncio.sleep(1)


class TestWrapper:
    def __init__(self, n):
        self.n = n

    async def run(self):
        await asyncio.gather(*(Test(f"Cor-{i}").do_work() for i in range(self.n)))


if __name__ == '__main__':
    asyncio.run(TestWrapper(3).run())