feat(storage): add get_value method to RedisStorage

This commit is contained in:
Вадим Христенко 2025-04-07 17:18:43 +00:00
parent 49e9b34b3f
commit 82da8dde1e

View file

@ -1,6 +1,6 @@
import json
from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator, Callable, Dict, Optional, cast
from typing import Any, AsyncGenerator, Callable, Dict, Optional, cast, overload
from redis.asyncio.client import Redis
from redis.asyncio.connection import ConnectionPool
@ -127,6 +127,18 @@ class RedisStorage(BaseStorage):
value = value.decode("utf-8")
return cast(Dict[str, Any], self.json_loads(value))
@overload
async def get_value(self, storage_key: StorageKey, dict_key: str) -> Optional[Any]: ...
@overload
async def get_value(self, storage_key: StorageKey, dict_key: str, default: Any) -> Any: ...
async def get_value(
self, storage_key: StorageKey, dict_key: str, default: Optional[Any] = None
) -> Optional[Any]:
data = await self.get_data(storage_key)
return data.get(dict_key, default)
class RedisEventIsolation(BaseEventIsolation):
def __init__(