Перейти к содержанию

Package: maxapi.context

maxapi.context

State()

Представляет отдельное состояние в FSM-группе.

При использовании внутри StatesGroup, автоматически присваивает уникальное имя в формате 'ИмяКласса:имя_поля'.

Source code in maxapi/context/state_machine.py 
11
12
def __init__(self) -> None:
    self.name: Optional[str] = None

StatesGroup

Базовый класс для описания группы состояний FSM.

Атрибуты должны быть экземплярами State. Метод states() возвращает список всех состояний в виде строк.

states() classmethod

Получить список всех состояний в формате 'ИмяКласса:имя_состояния'.

Returns:

Type Description
List[str]

Список строковых представлений состояний
Source code in maxapi/context/state_machine.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@classmethod
def states(cls) -> List[str]:
    """
    Получить список всех состояний в формате 'ИмяКласса:имя_состояния'.

    Returns:
        Список строковых представлений состояний
    """

    return [
        str(getattr(cls, attr))
        for attr in dir(cls)
        if isinstance(getattr(cls, attr), State)
    ]

BaseContext(chat_id, user_id, **kwargs)

Bases: ABC

Абстрактный базовый класс для контекста хранения данных пользователя.

Source code in maxapi/context/base.py 
14
15
16
17
18
def __init__(
    self, chat_id: Optional[int], user_id: Optional[int], **kwargs: Any
) -> None:
    self.chat_id = chat_id
    self.user_id = user_id

get_data() abstractmethod async

Возвращает текущий контекст данных.

Source code in maxapi/context/base.py 
20
21
22
23
@abstractmethod
async def get_data(self) -> Dict[str, Any]:
    """Возвращает текущий контекст данных."""
    pass

set_data(data) abstractmethod async

Полностью заменяет контекст данных.

Source code in maxapi/context/base.py 
25
26
27
28
@abstractmethod
async def set_data(self, data: Dict[str, Any]) -> None:
    """Полностью заменяет контекст данных."""
    pass

update_data(**kwargs) abstractmethod async

Обновляет контекст данных новыми значениями.

Source code in maxapi/context/base.py 
30
31
32
33
@abstractmethod
async def update_data(self, **kwargs: Any) -> None:
    """Обновляет контекст данных новыми значениями."""
    pass

set_state(state=None) abstractmethod async

Устанавливает новое состояние.

Source code in maxapi/context/base.py 
35
36
37
38
39
40
@abstractmethod
async def set_state(
    self, state: Optional[Union[State, str]] = None
) -> None:
    """Устанавливает новое состояние."""
    pass

get_state() abstractmethod async

Возвращает текущее состояние.

Source code in maxapi/context/base.py 
42
43
44
45
@abstractmethod
async def get_state(self) -> Optional[Union[State, str]]:
    """Возвращает текущее состояние."""
    pass

clear() abstractmethod async

Очищает контекст и сбрасывает состояние.

Source code in maxapi/context/base.py 
47
48
49
50
@abstractmethod
async def clear(self) -> None:
    """Очищает контекст и сбрасывает состояние."""
    pass

MemoryContext(chat_id, user_id, **kwargs)

Bases: BaseContext

Контекст хранения данных пользователя в оперативной памяти.

Source code in maxapi/context/context.py 
14
15
16
17
18
19
20
def __init__(
    self, chat_id: Optional[int], user_id: Optional[int], **kwargs: Any
) -> None:
    super().__init__(chat_id, user_id, **kwargs)
    self._context: Dict[str, Any] = {}
    self._state: State | str | None = None
    self._lock = asyncio.Lock()

get_data() async

Возвращает текущий контекст данных.

Returns:

Type Description
Dict[str, Any]

Словарь с данными контекста
Source code in maxapi/context/context.py 
22
23
24
25
26
27
28
29
30
31
async def get_data(self) -> Dict[str, Any]:
    """
    Возвращает текущий контекст данных.

    Returns:
        Словарь с данными контекста
    """

    async with self._lock:
        return self._context

set_data(data) async

Полностью заменяет контекст данных.

Parameters:

Name Type Description Default
data Dict[str, Any]

Новый словарь контекста

 required
Source code in maxapi/context/context.py 
33
34
35
36
37
38
39
40
41
42
async def set_data(self, data: Dict[str, Any]) -> None:
    """
    Полностью заменяет контекст данных.

    Args:
        data: Новый словарь контекста
    """

    async with self._lock:
        self._context = data

update_data(**kwargs) async

Обновляет контекст данных новыми значениями.

Parameters:

Name Type Description Default
**kwargs Any

Пары ключ-значение для обновления

 {}
Source code in maxapi/context/context.py 
44
45
46
47
48
49
50
51
52
53
async def update_data(self, **kwargs: Any) -> None:
    """
    Обновляет контекст данных новыми значениями.

    Args:
        **kwargs: Пары ключ-значение для обновления
    """

    async with self._lock:
        self._context.update(kwargs)

set_state(state=None) async

Устанавливает новое состояние.

Parameters:

Name Type Description Default
state Optional[Union[State, str]]

Новое состояние или None для сброса

 None
Source code in maxapi/context/context.py 
55
56
57
58
59
60
61
62
63
64
65
66
async def set_state(
    self, state: Optional[Union[State, str]] = None
) -> None:
    """
    Устанавливает новое состояние.

    Args:
        state: Новое состояние или None для сброса
    """

    async with self._lock:
        self._state = state

get_state() async

Возвращает текущее состояние.

Returns:

Type Description
Optional[Union[State, str]]

Текущее состояние или None
Source code in maxapi/context/context.py 
68
69
70
71
72
73
74
75
76
77
async def get_state(self) -> Optional[Union[State, str]]:
    """
    Возвращает текущее состояние.

    Returns:
        Текущее состояние или None
    """

    async with self._lock:
        return self._state

clear() async

Очищает контекст и сбрасывает состояние.

Source code in maxapi/context/context.py 
79
80
81
82
83
84
85
86
async def clear(self) -> None:
    """
    Очищает контекст и сбрасывает состояние.
    """

    async with self._lock:
        self._state = None
        self._context = {}

RedisContext(chat_id, user_id, redis_client, key_prefix='maxapi', **kwargs)

Bases: BaseContext

Контекст хранения данных пользователя в Redis. Требует установленной библиотеки redis: pip install redis

Source code in maxapi/context/context.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(
    self,
    chat_id: Optional[int],
    user_id: Optional[int],
    redis_client: Any,  # redis.asyncio.Redis
    key_prefix: str = "maxapi",
    **kwargs: Any,
) -> None:
    super().__init__(chat_id, user_id, **kwargs)
    self.redis = redis_client
    self.prefix = f"{key_prefix}:{chat_id}:{user_id}"
    self.data_key = f"{self.prefix}:data"
    self.state_key = f"{self.prefix}:state"

update_data(**kwargs) async

Атомарно обновляет данные

Source code in maxapi/context/context.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
async def update_data(self, **kwargs: Any) -> None:
    """
    Атомарно обновляет данные
    """
    lua_script = """
    local data = redis.call('get', KEYS[1])
    local decoded = {}
    if data then
        decoded = cjson.decode(data)
    end
    local updates = cjson.decode(ARGV[1])
    for k, v in pairs(updates) do
        decoded[k] = v
    end
    redis.call('set', KEYS[1], cjson.encode(decoded))
    return redis.status_reply("OK")
    """
    await self.redis.eval(lua_script, 1, self.data_key, json.dumps(kwargs))

base

BaseContext(chat_id, user_id, **kwargs)

Bases: ABC

Абстрактный базовый класс для контекста хранения данных пользователя.

Source code in maxapi/context/base.py 
14
15
16
17
18
def __init__(
    self, chat_id: Optional[int], user_id: Optional[int], **kwargs: Any
) -> None:
    self.chat_id = chat_id
    self.user_id = user_id
get_data() abstractmethod async

Возвращает текущий контекст данных.

Source code in maxapi/context/base.py 
20
21
22
23
@abstractmethod
async def get_data(self) -> Dict[str, Any]:
    """Возвращает текущий контекст данных."""
    pass
set_data(data) abstractmethod async

Полностью заменяет контекст данных.

Source code in maxapi/context/base.py 
25
26
27
28
@abstractmethod
async def set_data(self, data: Dict[str, Any]) -> None:
    """Полностью заменяет контекст данных."""
    pass
update_data(**kwargs) abstractmethod async

Обновляет контекст данных новыми значениями.

Source code in maxapi/context/base.py 
30
31
32
33
@abstractmethod
async def update_data(self, **kwargs: Any) -> None:
    """Обновляет контекст данных новыми значениями."""
    pass
set_state(state=None) abstractmethod async

Устанавливает новое состояние.

Source code in maxapi/context/base.py 
35
36
37
38
39
40
@abstractmethod
async def set_state(
    self, state: Optional[Union[State, str]] = None
) -> None:
    """Устанавливает новое состояние."""
    pass
get_state() abstractmethod async

Возвращает текущее состояние.

Source code in maxapi/context/base.py 
42
43
44
45
@abstractmethod
async def get_state(self) -> Optional[Union[State, str]]:
    """Возвращает текущее состояние."""
    pass
clear() abstractmethod async

Очищает контекст и сбрасывает состояние.

Source code in maxapi/context/base.py 
47
48
49
50
@abstractmethod
async def clear(self) -> None:
    """Очищает контекст и сбрасывает состояние."""
    pass

context

MemoryContext(chat_id, user_id, **kwargs)

Bases: BaseContext

Контекст хранения данных пользователя в оперативной памяти.

Source code in maxapi/context/context.py 
14
15
16
17
18
19
20
def __init__(
    self, chat_id: Optional[int], user_id: Optional[int], **kwargs: Any
) -> None:
    super().__init__(chat_id, user_id, **kwargs)
    self._context: Dict[str, Any] = {}
    self._state: State | str | None = None
    self._lock = asyncio.Lock()
get_data() async

Возвращает текущий контекст данных.

Returns:

Type Description
Dict[str, Any]

Словарь с данными контекста
Source code in maxapi/context/context.py 
22
23
24
25
26
27
28
29
30
31
async def get_data(self) -> Dict[str, Any]:
    """
    Возвращает текущий контекст данных.

    Returns:
        Словарь с данными контекста
    """

    async with self._lock:
        return self._context
set_data(data) async

Полностью заменяет контекст данных.

Parameters:

Name Type Description Default
data Dict[str, Any]

Новый словарь контекста

 required
Source code in maxapi/context/context.py 
33
34
35
36
37
38
39
40
41
42
async def set_data(self, data: Dict[str, Any]) -> None:
    """
    Полностью заменяет контекст данных.

    Args:
        data: Новый словарь контекста
    """

    async with self._lock:
        self._context = data
update_data(**kwargs) async

Обновляет контекст данных новыми значениями.

Parameters:

Name Type Description Default
**kwargs Any

Пары ключ-значение для обновления

 {}
Source code in maxapi/context/context.py 
44
45
46
47
48
49
50
51
52
53
async def update_data(self, **kwargs: Any) -> None:
    """
    Обновляет контекст данных новыми значениями.

    Args:
        **kwargs: Пары ключ-значение для обновления
    """

    async with self._lock:
        self._context.update(kwargs)
set_state(state=None) async

Устанавливает новое состояние.

Parameters:

Name Type Description Default
state Optional[Union[State, str]]

Новое состояние или None для сброса

 None
Source code in maxapi/context/context.py 
55
56
57
58
59
60
61
62
63
64
65
66
async def set_state(
    self, state: Optional[Union[State, str]] = None
) -> None:
    """
    Устанавливает новое состояние.

    Args:
        state: Новое состояние или None для сброса
    """

    async with self._lock:
        self._state = state
get_state() async

Возвращает текущее состояние.

Returns:

Type Description
Optional[Union[State, str]]

Текущее состояние или None
Source code in maxapi/context/context.py 
68
69
70
71
72
73
74
75
76
77
async def get_state(self) -> Optional[Union[State, str]]:
    """
    Возвращает текущее состояние.

    Returns:
        Текущее состояние или None
    """

    async with self._lock:
        return self._state
clear() async

Очищает контекст и сбрасывает состояние.

Source code in maxapi/context/context.py 
79
80
81
82
83
84
85
86
async def clear(self) -> None:
    """
    Очищает контекст и сбрасывает состояние.
    """

    async with self._lock:
        self._state = None
        self._context = {}

RedisContext(chat_id, user_id, redis_client, key_prefix='maxapi', **kwargs)

Bases: BaseContext

Контекст хранения данных пользователя в Redis. Требует установленной библиотеки redis: pip install redis

Source code in maxapi/context/context.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def __init__(
    self,
    chat_id: Optional[int],
    user_id: Optional[int],
    redis_client: Any,  # redis.asyncio.Redis
    key_prefix: str = "maxapi",
    **kwargs: Any,
) -> None:
    super().__init__(chat_id, user_id, **kwargs)
    self.redis = redis_client
    self.prefix = f"{key_prefix}:{chat_id}:{user_id}"
    self.data_key = f"{self.prefix}:data"
    self.state_key = f"{self.prefix}:state"
update_data(**kwargs) async

Атомарно обновляет данные

Source code in maxapi/context/context.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
async def update_data(self, **kwargs: Any) -> None:
    """
    Атомарно обновляет данные
    """
    lua_script = """
    local data = redis.call('get', KEYS[1])
    local decoded = {}
    if data then
        decoded = cjson.decode(data)
    end
    local updates = cjson.decode(ARGV[1])
    for k, v in pairs(updates) do
        decoded[k] = v
    end
    redis.call('set', KEYS[1], cjson.encode(decoded))
    return redis.status_reply("OK")
    """
    await self.redis.eval(lua_script, 1, self.data_key, json.dumps(kwargs))

state_machine

State()

Представляет отдельное состояние в FSM-группе.

При использовании внутри StatesGroup, автоматически присваивает уникальное имя в формате 'ИмяКласса:имя_поля'.

Source code in maxapi/context/state_machine.py 
11
12
def __init__(self) -> None:
    self.name: Optional[str] = None

StatesGroup

Базовый класс для описания группы состояний FSM.

Атрибуты должны быть экземплярами State. Метод states() возвращает список всех состояний в виде строк.

states() classmethod

Получить список всех состояний в формате 'ИмяКласса:имя_состояния'.

Returns:

Type Description
List[str]

Список строковых представлений состояний
Source code in maxapi/context/state_machine.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
@classmethod
def states(cls) -> List[str]:
    """
    Получить список всех состояний в формате 'ИмяКласса:имя_состояния'.

    Returns:
        Список строковых представлений состояний
    """

    return [
        str(getattr(cls, attr))
        for attr in dir(cls)
        if isinstance(getattr(cls, attr), State)
    ]