运行 class 初始化程序中的周期性后台函数

Question

我正在尝试连接一些需要使用 asyncio 的身份验证令牌的 Api。

我在 ApiClient class 中有一个获取此令牌的方法。

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)

问题是，每小时我都需要调用此函数以维护有效令牌，因为它每小时刷新一次（没有这个我将在一小时后得到 401）。

如何让此方法从 ApiClient 初始化开始每小时在后台调用一次？

(_set_auth_token方法发出HTTP请求然后self._auth_token = res.id_token)

Answer 1

要使用 asyncio 库，您的程序需要运行在 asyncio 事件循环中。假设已经是这种情况，您需要使用 asyncio.create_task:

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

请注意，在调用 ApiClient() 时，身份验证令牌将不可用，它不会早于第一个 await，在后台任务有机会运行。要解决这个问题，您可以使 _set_async_token public 并显式等待它：

client = ApiClient()
await client.set_async_token()

为了使使用更符合人体工程学，您可以实现异步上下文管理器。例如：

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()

然后在 async with:

中使用 ApiClient

async with ApiClient() as client:
    ...

这样做的好处是一旦不再需要客户端就可以可靠地取消任务。由于 Python 是垃圾收集的并且不支持确定性析构函数，如果任务是在构造函数中创建的，这将是不可能的。

运行 class 初始化程序中的周期性后台函数

Running periodically background function in class initializer

multithreading

background-process

python-3.x

python-asyncio