Running periodically background function in class initializer

Question

I am trying to interface some Api which requires an auth token using asyncio.

I have a method inside the ApiClient class for obtaining this token.

class ApiClient:    
    def __init__(self):
        self._auth_token = None
        # how to invoke _keep_auth_token_alive in the background?

    async def _keep_auth_token_alive(self):
        while True:
            await self._set_auth_token()
            await asyncio.sleep(3600)        

The problem is, that every hour I need to recall this function in order to maintain a valid token because it refreshes every hour (without this I will get 401 after one hour).

How can I make this method invoke every hour in the background starting at the initializing of ApiClient?

(The _set_auth_token method makes an HTTP request and then self._auth_token = res.id_token)

Answer 1

To use an asyncio library, your program needs to run inside the asyncio event loop. Assuming that's already the case, you need to use asyncio.create_task:

class ApiClient:    
    def __init__(self):
        self._auth_token = None
        asyncio.create_task(self._keep_auth_token_alive())

Note that the auth token will not be available upon a call to ApiClient(), it will get filled no earlier than the first await, after the background task has had a chance to run. To fix that, you can make _set_async_token public and await it explicitly:

client = ApiClient()
await client.set_async_token()

To make the usage more ergonomic, you can implement an async context manager. For example:

class ApiClient:    
    def __init__(self):
        self._auth_token = None

    async def __aenter__(self):
        await self._set_auth_token()
        self._keepalive_task = asyncio.create_task(self._keep_auth_token_alive())

    async def __aexit__(self, *_):
        self._keepalive_task.cancel()

    async def _keep_auth_token_alive(self):
        # modified to sleep first and then re-acquire the token
        while True:
            await asyncio.sleep(3600)    
            await self._set_auth_token()

Then you use ApiClient inside an async with:

async with ApiClient() as client:
    ...

This has the advantage of reliably canceling the task once the client is no longer needed. Since Python is garbage-collected and doesn't support deterministic destructors, this would not be possible if the task were created in the constructor.