-
Notifications
You must be signed in to change notification settings - Fork 156
/
multicached_decorator.py
45 lines (31 loc) · 1.03 KB
/
multicached_decorator.py
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())