mirror of
https://github.com/aiogram/aiogram.git
synced 2026-04-08 16:37:47 +00:00
Add function get_value to BaseStorage (#1594)
* Add function `get_value` to all built-in storage implementations, `FSMContext` and `SceneWizard` (#1431) * Fix type hints * Split up return statements in `get_value` functions * Implement `get_value` method in `BaseStorage` and remove redundant implementations
This commit is contained in:
parent
fd014d2026
commit
592267dd99
8 changed files with 120 additions and 4 deletions
|
|
@ -1,4 +1,4 @@
|
|||
from typing import Any, Dict, Optional
|
||||
from typing import Any, Dict, Optional, overload
|
||||
|
||||
from aiogram.fsm.storage.base import BaseStorage, StateType, StorageKey
|
||||
|
||||
|
|
@ -20,6 +20,15 @@ class FSMContext:
|
|||
async def get_data(self) -> Dict[str, Any]:
|
||||
return await self.storage.get_data(key=self.key)
|
||||
|
||||
@overload
|
||||
async def get_value(self, key: str) -> Optional[Any]: ...
|
||||
|
||||
@overload
|
||||
async def get_value(self, key: str, default: Any) -> Any: ...
|
||||
|
||||
async def get_value(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
|
||||
return await self.storage.get_value(storage_key=self.key, dict_key=key, default=default)
|
||||
|
||||
async def update_data(
|
||||
self, data: Optional[Dict[str, Any]] = None, **kwargs: Any
|
||||
) -> Dict[str, Any]:
|
||||
|
|
|
|||
|
|
@ -4,7 +4,7 @@ import inspect
|
|||
from collections import defaultdict
|
||||
from dataclasses import dataclass, replace
|
||||
from enum import Enum, auto
|
||||
from typing import Any, ClassVar, Dict, List, Optional, Tuple, Type, Union
|
||||
from typing import Any, ClassVar, Dict, List, Optional, Tuple, Type, Union, overload
|
||||
|
||||
from typing_extensions import Self
|
||||
|
||||
|
|
@ -572,6 +572,32 @@ class SceneWizard:
|
|||
"""
|
||||
return await self.state.get_data()
|
||||
|
||||
@overload
|
||||
async def get_value(self, key: str) -> Optional[Any]:
|
||||
"""
|
||||
This method returns the value from key in the data of the current state.
|
||||
|
||||
:param key: The keyname of the item you want to return the value from.
|
||||
|
||||
:return: A dictionary containing the data stored in the scene state.
|
||||
"""
|
||||
pass
|
||||
|
||||
@overload
|
||||
async def get_value(self, key: str, default: Any) -> Any:
|
||||
"""
|
||||
This method returns the value from key in the data of the current state.
|
||||
|
||||
:param key: The keyname of the item you want to return the value from.
|
||||
:param default: Default value to return, if ``key`` was not found.
|
||||
|
||||
:return: A dictionary containing the data stored in the scene state.
|
||||
"""
|
||||
pass
|
||||
|
||||
async def get_value(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
|
||||
return await self.state.get_value(key, default)
|
||||
|
||||
async def update_data(
|
||||
self, data: Optional[Dict[str, Any]] = None, **kwargs: Any
|
||||
) -> Dict[str, Any]:
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
from abc import ABC, abstractmethod
|
||||
from contextlib import asynccontextmanager
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, AsyncGenerator, Dict, Literal, Optional, Union
|
||||
from typing import Any, AsyncGenerator, Dict, Literal, Optional, Union, overload
|
||||
|
||||
from aiogram.fsm.state import State
|
||||
|
||||
|
|
@ -144,6 +144,35 @@ class BaseStorage(ABC):
|
|||
"""
|
||||
pass
|
||||
|
||||
@overload
|
||||
async def get_value(self, storage_key: StorageKey, dict_key: str) -> Optional[Any]:
|
||||
"""
|
||||
Get single value from data by key
|
||||
|
||||
:param storage_key: storage key
|
||||
:param dict_key: value key
|
||||
:return: value stored in key of dict or ``None``
|
||||
"""
|
||||
pass
|
||||
|
||||
@overload
|
||||
async def get_value(self, storage_key: StorageKey, dict_key: str, default: Any) -> Any:
|
||||
"""
|
||||
Get single value from data by key
|
||||
|
||||
:param storage_key: storage key
|
||||
:param dict_key: value key
|
||||
:param default: default value to return
|
||||
:return: value stored in key of dict or default
|
||||
"""
|
||||
pass
|
||||
|
||||
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)
|
||||
|
||||
async def update_data(self, key: StorageKey, data: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""
|
||||
Update date in the storage for key (like dict.update)
|
||||
|
|
|
|||
|
|
@ -1,8 +1,9 @@
|
|||
from asyncio import Lock
|
||||
from collections import defaultdict
|
||||
from contextlib import asynccontextmanager
|
||||
from copy import copy
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, AsyncGenerator, DefaultDict, Dict, Hashable, Optional
|
||||
from typing import Any, AsyncGenerator, DefaultDict, Dict, Hashable, Optional, overload
|
||||
|
||||
from aiogram.fsm.state import State
|
||||
from aiogram.fsm.storage.base import (
|
||||
|
|
@ -49,6 +50,18 @@ class MemoryStorage(BaseStorage):
|
|||
async def get_data(self, key: StorageKey) -> Dict[str, Any]:
|
||||
return self.storage[key].data.copy()
|
||||
|
||||
@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 = self.storage[storage_key].data
|
||||
return copy(data.get(dict_key, default))
|
||||
|
||||
|
||||
class DisabledEventIsolation(BaseEventIsolation):
|
||||
@asynccontextmanager
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue