async-lru-threadsafe 2.0.4

安装

pip install async-lru

使用

此软件包是Python内置的functools.lru_cache函数的asyncio版本。为了更好地处理异步行为，它还确保多个并发调用只调用一次包装函数，当该调用完成时，所有await都将收到该调用的结果。

import asyncio

import aiohttp
from async_lru import alru_cache


@alru_cache(maxsize=32)
async def get_pep(num):
    resource = 'https://pythonlang.cn/dev/peps/pep-%04d/' % num
    async with aiohttp.ClientSession() as session:
        try:
            async with session.get(resource) as s:
                return await s.read()
        except aiohttp.ClientError:
            return 'Not Found'


async def main():
    for n in 8, 290, 308, 320, 8, 218, 320, 279, 289, 320, 9991:
        pep = await get_pep(n)
        print(n, len(pep))

    print(get_pep.cache_info())
    # CacheInfo(hits=3, misses=8, maxsize=32, currsize=8)

    # closing is optional, but highly recommended
    await get_pep.cache_close()


asyncio.run(main())

通过接受ttl配置参数（默认关闭）支持TTL（生存时间，超时后过期）

@alru_cache(ttl=5)
async def func(arg):
    return arg * 2

库支持通过cache_invalidate()显式失效特定函数调用

@alru_cache(ttl=5)
async def func(arg1, arg2):
    return arg1 + arg2

func.cache_invalidate(1, arg2=2)

如果相应的参数集合已被缓存，则方法返回True，否则返回False。

需要Python 3.8+

感谢

此库由Ocean S.A.捐赠

感谢公司的贡献。