1
"""Repository for work with redis."""
7
from typing import Any, Hashable, Optional
10
from starlette.datastructures import URL
21
self, dsn: URL, prefix: Optional[str] = None, expire: Optional[int] = None
23
"""Init repository object."""
32
cls, dsn: URL, prefix: Optional[str] = None, expire: Optional[int] = None
34
"""Init repository object."""
36
repo = cls(dsn=dsn, prefix=prefix, expire=expire)
37
repo.redis = await aioredis.create_redis_pool(repo.dsn)
40
async def close(self) -> None:
41
"""Close connection to redis."""
44
await self.redis.wait_closed()
46
def __key(self, arg: Hashable) -> str:
47
if self.prefix is not None:
48
_prefix = self.prefix + self.delimiter
52
return _prefix + hashlib.md5(pickle.dumps(arg)).hexdigest()
54
async def get(self, key: Hashable, default: Any = None) -> Any:
55
"""Get value from redis."""
57
cached_data = await self.redis.get(self.__key(key))
58
if cached_data is None:
61
return pickle.loads(cached_data)
64
self, key: Hashable, value: Any, expire: Optional[int] = None
66
"""Set value into redis."""
70
await self.redis.set(self.__key(key), pickle.dumps(value), expire=expire)
72
async def delete(self, key: Hashable) -> None:
73
"""Remove value from redis."""
75
await self.redis.delete(self.__key(key))
77
async def rget(self, key: Hashable, default: Any = None) -> Any:
78
"""Get value and remove it from redis."""
80
value = await self.get(key, default)
81
await self.delete(key)