mirror of
https://github.com/aiogram/aiogram.git
synced 2026-04-08 16:37:47 +00:00
Add dynamic dataclass and class attribute resolvers
Introduced `dataclass_kwargs` to ensure compatibility with different Python versions and modular attribute handling. Added utilities for resolving class attributes dynamically, enhancing flexibility with MRO-based resolvers. Updated tests to verify new features and ensure proper functionality across various scenarios.
This commit is contained in:
parent
d36a10d428
commit
b146b7aa85
7 changed files with 349 additions and 13 deletions
|
|
@ -1,8 +1,9 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from typing import TYPE_CHECKING, Any, Dict, Optional
|
||||
from typing import TYPE_CHECKING, Any, Optional
|
||||
|
||||
from aiogram.utils.dataclass import dataclass_kwargs
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from aiogram.types import LinkPreviewOptions
|
||||
|
|
@ -28,13 +29,7 @@ class Default:
|
|||
return f"<{self}>"
|
||||
|
||||
|
||||
_dataclass_properties: Dict[str, Any] = {}
|
||||
if sys.version_info >= (3, 10):
|
||||
# Speedup attribute access for dataclasses in Python 3.10+
|
||||
_dataclass_properties.update({"slots": True, "kw_only": True})
|
||||
|
||||
|
||||
@dataclass(**_dataclass_properties)
|
||||
@dataclass(**dataclass_kwargs(slots=True, kw_only=True))
|
||||
class DefaultBotProperties:
|
||||
"""
|
||||
Default bot properties.
|
||||
|
|
|
|||
|
|
@ -20,6 +20,10 @@ from aiogram.fsm.context import FSMContext
|
|||
from aiogram.fsm.state import State
|
||||
from aiogram.fsm.storage.memory import MemoryStorageRecord
|
||||
from aiogram.types import TelegramObject, Update
|
||||
from aiogram.utils.class_attrs_resolver import (
|
||||
ClassAttrsResolver,
|
||||
get_sorted_mro_attrs_resolver,
|
||||
)
|
||||
|
||||
|
||||
class HistoryManager:
|
||||
|
|
@ -214,6 +218,15 @@ class SceneConfig:
|
|||
"""Reset scene history on enter"""
|
||||
callback_query_without_state: Optional[bool] = None
|
||||
"""Allow callback query without state"""
|
||||
attrs_resolver: ClassAttrsResolver = get_sorted_mro_attrs_resolver
|
||||
"""
|
||||
Attributes resolver.
|
||||
|
||||
.. danger::
|
||||
This attribute should only be changed when you know what you are doing.
|
||||
|
||||
.. versionadded:: 3.19.0
|
||||
"""
|
||||
|
||||
|
||||
async def _empty_handler(*args: Any, **kwargs: Any) -> None:
|
||||
|
|
@ -302,6 +315,7 @@ class Scene:
|
|||
reset_data_on_enter = kwargs.pop("reset_data_on_enter", None)
|
||||
reset_history_on_enter = kwargs.pop("reset_history_on_enter", None)
|
||||
callback_query_without_state = kwargs.pop("callback_query_without_state", None)
|
||||
attrs_resolver = kwargs.pop("attrs_resolver", None)
|
||||
|
||||
super().__init_subclass__(**kwargs)
|
||||
|
||||
|
|
@ -322,11 +336,13 @@ class Scene:
|
|||
reset_history_on_enter = parent_scene_config.reset_history_on_enter
|
||||
if callback_query_without_state is None:
|
||||
callback_query_without_state = parent_scene_config.callback_query_without_state
|
||||
if attrs_resolver is None:
|
||||
attrs_resolver = parent_scene_config.attrs_resolver
|
||||
|
||||
members: dict[str, Any] = {}
|
||||
for base in reversed(inspect.getmro(cls)):
|
||||
members.update(base.__dict__)
|
||||
for name, value in members.items():
|
||||
if attrs_resolver is None:
|
||||
attrs_resolver = get_sorted_mro_attrs_resolver
|
||||
|
||||
for name, value in attrs_resolver(cls):
|
||||
if scene_handlers := getattr(value, "__aiogram_handler__", None):
|
||||
handlers.extend(scene_handlers)
|
||||
if isinstance(value, ObserverDecorator):
|
||||
|
|
@ -349,6 +365,7 @@ class Scene:
|
|||
reset_data_on_enter=reset_data_on_enter,
|
||||
reset_history_on_enter=reset_history_on_enter,
|
||||
callback_query_without_state=callback_query_without_state,
|
||||
attrs_resolver=attrs_resolver,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
|
|
|
|||
86
aiogram/utils/class_attrs_resolver.py
Normal file
86
aiogram/utils/class_attrs_resolver.py
Normal file
|
|
@ -0,0 +1,86 @@
|
|||
import inspect
|
||||
from dataclasses import dataclass
|
||||
from operator import itemgetter
|
||||
from typing import Any, Generator, NamedTuple, Protocol
|
||||
|
||||
from aiogram.utils.dataclass import dataclass_kwargs
|
||||
|
||||
|
||||
class ClassAttrsResolver(Protocol):
|
||||
def __call__(self, cls: type) -> Generator[tuple[str, Any], None, None]: ...
|
||||
|
||||
|
||||
def inspect_members_resolver(cls: type) -> Generator[tuple[str, Any], None, None]:
|
||||
"""
|
||||
Inspects and resolves attributes of a given class.
|
||||
|
||||
This function uses the `inspect.getmembers` utility to yield all attributes of
|
||||
a provided class. The output is a generator that produces tuples containing
|
||||
attribute names and their corresponding values. This function is suitable for
|
||||
analyzing class attributes dynamically. However, it guarantees alphabetical
|
||||
order of attributes.
|
||||
|
||||
:param cls: The class for which the attributes will be resolved.
|
||||
:return: A generator yielding tuples containing attribute names and their values.
|
||||
"""
|
||||
yield from inspect.getmembers(cls)
|
||||
|
||||
|
||||
def get_reversed_mro_unique_attrs_resolver(cls: type) -> Generator[tuple[str, Any], None, None]:
|
||||
"""
|
||||
Resolve and yield attributes from the reversed method resolution order (MRO) of a given class.
|
||||
|
||||
This function iterates through the reversed MRO of a class and yields attributes
|
||||
that have not yet been encountered. It avoids duplicates by keeping track of
|
||||
attribute names that have already been processed.
|
||||
|
||||
:param cls: The class for which the attributes will be resolved.
|
||||
:return: A generator yielding tuples containing attribute names and their values.
|
||||
"""
|
||||
known_attrs = set()
|
||||
for base in reversed(inspect.getmro(cls)):
|
||||
for name, value in base.__dict__.items():
|
||||
if name in known_attrs:
|
||||
continue
|
||||
|
||||
yield name, value
|
||||
known_attrs.add(name)
|
||||
|
||||
|
||||
class _Position(NamedTuple):
|
||||
in_mro: int
|
||||
in_class: int
|
||||
|
||||
|
||||
@dataclass(**dataclass_kwargs(slots=True))
|
||||
class _AttributeContainer:
|
||||
position: _Position
|
||||
value: Any
|
||||
|
||||
def __lt__(self, other: "_AttributeContainer") -> bool:
|
||||
return self.position < other.position
|
||||
|
||||
|
||||
def get_sorted_mro_attrs_resolver(cls: type) -> Generator[tuple[str, Any], None, None]:
|
||||
"""
|
||||
Resolve and yield attributes from the method resolution order (MRO) of a given class.
|
||||
|
||||
Iterates through a class's method resolution order (MRO) and collects its attributes
|
||||
along with their respective positions in the MRO and the class hierarchy. This generator
|
||||
yields a tuple containing the name of each attribute and its associated value.
|
||||
|
||||
:param cls: The class for which the attributes will be resolved.
|
||||
:return: A generator yielding tuples containing attribute names and their values.
|
||||
"""
|
||||
attributes: dict[str, _AttributeContainer] = {}
|
||||
for position_in_mro, base in enumerate(inspect.getmro(cls)):
|
||||
for position_in_class, (name, value) in enumerate(vars(base).items()):
|
||||
position = _Position(position_in_mro, position_in_class)
|
||||
if attribute := attributes.get(name):
|
||||
attribute.position = position
|
||||
continue
|
||||
|
||||
attributes[name] = _AttributeContainer(value=value, position=position)
|
||||
|
||||
for name, attribute in sorted(attributes.items(), key=itemgetter(1)):
|
||||
yield name, attribute.value
|
||||
64
aiogram/utils/dataclass.py
Normal file
64
aiogram/utils/dataclass.py
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
"""
|
||||
This module contains utility functions for working with dataclasses in Python.
|
||||
|
||||
DO NOT USE THIS MODULE DIRECTLY. IT IS INTENDED FOR INTERNAL USE ONLY.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from typing import Any, Union
|
||||
|
||||
|
||||
def dataclass_kwargs(
|
||||
init: Union[bool, None] = None,
|
||||
repr: Union[bool, None] = None,
|
||||
eq: Union[bool, None] = None,
|
||||
order: Union[bool, None] = None,
|
||||
unsafe_hash: Union[bool, None] = None,
|
||||
frozen: Union[bool, None] = None,
|
||||
match_args: Union[bool, None] = None,
|
||||
kw_only: Union[bool, None] = None,
|
||||
slots: Union[bool, None] = None,
|
||||
weakref_slot: Union[bool, None] = None,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Generates a dictionary of keyword arguments that can be passed to a Python
|
||||
dataclass. This function allows specifying attributes related to the behavior
|
||||
and configuration of dataclasses, including attributes added in specific
|
||||
Python versions. This abstraction improves compatibility across different
|
||||
Python versions by ensuring only the parameters supported by the current
|
||||
version are included.
|
||||
|
||||
:return: A dictionary containing the specified dataclass configuration that
|
||||
dynamically adapts to the current Python version.
|
||||
"""
|
||||
params = {}
|
||||
|
||||
# All versions
|
||||
if init is not None:
|
||||
params["init"] = init
|
||||
if repr is not None:
|
||||
params["repr"] = repr
|
||||
if eq is not None:
|
||||
params["eq"] = eq
|
||||
if order is not None:
|
||||
params["order"] = order
|
||||
if unsafe_hash is not None:
|
||||
params["unsafe_hash"] = unsafe_hash
|
||||
if frozen is not None:
|
||||
params["frozen"] = frozen
|
||||
|
||||
# Added in 3.10
|
||||
if sys.version_info >= (3, 10):
|
||||
if match_args is not None:
|
||||
params["match_args"] = match_args
|
||||
if kw_only is not None:
|
||||
params["kw_only"] = kw_only
|
||||
if slots is not None:
|
||||
params["slots"] = slots
|
||||
|
||||
# Added in 3.11
|
||||
if sys.version_info >= (3, 11):
|
||||
if weakref_slot is not None:
|
||||
params["weakref_slot"] = weakref_slot
|
||||
|
||||
return params
|
||||
|
|
@ -1708,3 +1708,45 @@ class TestSceneHandlersOrdering:
|
|||
|
||||
assert collect_handler_names(Scene1) == ["handler1", "handler2"]
|
||||
assert collect_handler_names(Scene2) == ["handler2", "handler1"]
|
||||
|
||||
def test_inherited_handlers_follow_mro_order(self):
|
||||
class Scene1(Scene):
|
||||
@on.message()
|
||||
async def handler1(self, message: Message) -> None:
|
||||
pass
|
||||
|
||||
@on.message()
|
||||
async def handler2(self, message: Message) -> None:
|
||||
pass
|
||||
|
||||
class Scene2(Scene1):
|
||||
@on.message()
|
||||
async def handler3(self, message: Message) -> None:
|
||||
pass
|
||||
|
||||
@on.message()
|
||||
async def handler4(self, message: Message) -> None:
|
||||
pass
|
||||
|
||||
assert collect_handler_names(Scene2) == ["handler3", "handler4", "handler1", "handler2"]
|
||||
|
||||
def test_inherited_overwrite_follow_mro_order(self):
|
||||
class Scene1(Scene):
|
||||
@on.message()
|
||||
async def handler1(self, message: Message) -> None:
|
||||
pass
|
||||
|
||||
@on.message()
|
||||
async def handler2(self, message: Message) -> None:
|
||||
pass
|
||||
|
||||
class Scene2(Scene1):
|
||||
@on.message()
|
||||
async def handler2(self, message: Message) -> None:
|
||||
pass
|
||||
|
||||
@on.message()
|
||||
async def handler3(self, message: Message) -> None:
|
||||
pass
|
||||
|
||||
assert collect_handler_names(Scene2) == ["handler3", "handler1", "handler2"]
|
||||
|
|
|
|||
81
tests/test_utils/test_class_attrs_resolver.py
Normal file
81
tests/test_utils/test_class_attrs_resolver.py
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
import pytest
|
||||
|
||||
from aiogram.utils.class_attrs_resolver import (
|
||||
get_reversed_mro_unique_attrs_resolver,
|
||||
get_sorted_mro_attrs_resolver,
|
||||
inspect_members_resolver,
|
||||
)
|
||||
|
||||
|
||||
class SimpleClass1:
|
||||
def method1(self):
|
||||
pass
|
||||
|
||||
def method2(self):
|
||||
pass
|
||||
|
||||
|
||||
class SimpleClass2:
|
||||
def method2(self):
|
||||
pass
|
||||
|
||||
def method1(self):
|
||||
pass
|
||||
|
||||
|
||||
class InheritedClass1(SimpleClass1):
|
||||
def method3(self):
|
||||
pass
|
||||
|
||||
def method4(self):
|
||||
pass
|
||||
|
||||
|
||||
class InheritedClass2(SimpleClass1):
|
||||
def method2(self):
|
||||
pass
|
||||
|
||||
def method3(self):
|
||||
pass
|
||||
|
||||
|
||||
class TestClassAttrsResolver:
|
||||
@pytest.mark.parametrize(
|
||||
"cls, resolver, expected",
|
||||
[
|
||||
# inspect_members_resolver
|
||||
(SimpleClass1, inspect_members_resolver, ["method1", "method2"]),
|
||||
(SimpleClass2, inspect_members_resolver, ["method1", "method2"]),
|
||||
(
|
||||
InheritedClass1,
|
||||
inspect_members_resolver,
|
||||
["method1", "method2", "method3", "method4"],
|
||||
),
|
||||
(InheritedClass2, inspect_members_resolver, ["method1", "method2", "method3"]),
|
||||
# get_reversed_mro_unique_attrs_resolver
|
||||
(SimpleClass1, get_reversed_mro_unique_attrs_resolver, ["method1", "method2"]),
|
||||
(SimpleClass2, get_reversed_mro_unique_attrs_resolver, ["method2", "method1"]),
|
||||
(
|
||||
InheritedClass1,
|
||||
get_reversed_mro_unique_attrs_resolver,
|
||||
["method1", "method2", "method3", "method4"],
|
||||
),
|
||||
(
|
||||
InheritedClass2,
|
||||
get_reversed_mro_unique_attrs_resolver,
|
||||
["method1", "method2", "method3"],
|
||||
),
|
||||
# get_sorted_mro_attrs_resolver
|
||||
(SimpleClass1, get_sorted_mro_attrs_resolver, ["method1", "method2"]),
|
||||
(SimpleClass2, get_sorted_mro_attrs_resolver, ["method2", "method1"]),
|
||||
(
|
||||
InheritedClass1,
|
||||
get_sorted_mro_attrs_resolver,
|
||||
["method3", "method4", "method1", "method2"],
|
||||
),
|
||||
(InheritedClass2, get_sorted_mro_attrs_resolver, ["method3", "method1", "method2"]),
|
||||
],
|
||||
)
|
||||
def test_resolve_class_attrs(self, cls, resolver, expected):
|
||||
names = [name for name, _ in resolver(cls) if not name.startswith("__")]
|
||||
assert names == expected
|
||||
51
tests/test_utils/test_dataclass.py
Normal file
51
tests/test_utils/test_dataclass.py
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from aiogram.utils.dataclass import dataclass_kwargs
|
||||
|
||||
ALL_VERSIONS = {
|
||||
"init": True,
|
||||
"repr": True,
|
||||
"eq": True,
|
||||
"order": True,
|
||||
"unsafe_hash": True,
|
||||
"frozen": True,
|
||||
}
|
||||
ADDED_IN_3_10 = {"match_args": True, "kw_only": True, "slots": True}
|
||||
ADDED_IN_3_11 = {"weakref_slot": True}
|
||||
|
||||
PY_310 = {**ALL_VERSIONS, **ADDED_IN_3_10}
|
||||
PY_311 = {**PY_310, **ADDED_IN_3_11}
|
||||
LATEST_PY = PY_311
|
||||
|
||||
|
||||
class TestDataclassKwargs:
|
||||
@pytest.mark.parametrize(
|
||||
"py_version,expected",
|
||||
[
|
||||
((3, 9, 0), ALL_VERSIONS),
|
||||
((3, 9, 2), ALL_VERSIONS),
|
||||
((3, 10, 2), PY_310),
|
||||
((3, 11, 0), PY_311),
|
||||
((4, 13, 0), LATEST_PY),
|
||||
],
|
||||
)
|
||||
def test_dataclass_kwargs(self, py_version, expected):
|
||||
with patch("sys.version_info", py_version):
|
||||
|
||||
assert (
|
||||
dataclass_kwargs(
|
||||
init=True,
|
||||
repr=True,
|
||||
eq=True,
|
||||
order=True,
|
||||
unsafe_hash=True,
|
||||
frozen=True,
|
||||
match_args=True,
|
||||
kw_only=True,
|
||||
slots=True,
|
||||
weakref_slot=True,
|
||||
)
|
||||
== expected
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue