Decorators

aiocache comes with a couple of decorators for caching results from asynchronous functions. Do not use the decorator in synchronous functions, it may lead to unexpected behavior.

cached

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import asyncio

from collections import namedtuple
import redis.asyncio as redis

from aiocache import cached, Cache
from aiocache.serializers import PickleSerializer

Result = namedtuple('Result', "content, status")


@cached(
    ttl=10, cache=Cache.REDIS, key_builder=lambda *args, **kw: "key",
    serializer=PickleSerializer(), namespace="main", client=redis.Redis())
async def cached_call():
    return Result("content", 200)


async def test_cached():
    async with Cache(Cache.REDIS, namespace="main", client=redis.Redis()) as cache:
        await cached_call()
        exists = await cache.exists("key")
        assert exists is True
        await cache.delete("key")


if __name__ == "__main__":
    asyncio.run(test_cached())

multi_cached

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
import asyncio

import redis.asyncio as redis

from aiocache import multi_cached, Cache

DICT = {
    'a': "Z",
    'b': "Y",
    'c': "X",
    'd': "W"
}

cache = Cache(Cache.REDIS, namespace="main", client=redis.Redis())


@multi_cached("ids", cache=Cache.REDIS, namespace="main", client=cache.client)
async def multi_cached_ids(ids=None):
    return {id_: DICT[id_] for id_ in ids}


@multi_cached("keys", cache=Cache.REDIS, namespace="main", client=cache.client)
async def multi_cached_keys(keys=None):
    return {id_: DICT[id_] for id_ in keys}


async def test_multi_cached():
    await multi_cached_ids(ids=("a", "b"))
    await multi_cached_ids(ids=("a", "c"))
    await multi_cached_keys(keys=("d",))

    assert await cache.exists("a")
    assert await cache.exists("b")
    assert await cache.exists("c")
    assert await cache.exists("d")

    await cache.delete("a")
    await cache.delete("b")
    await cache.delete("c")
    await cache.delete("d")
    await cache.close()


if __name__ == "__main__":
    asyncio.run(test_multi_cached())