Skip to content

FSM

FSMContext 

FSMContext(storage: BaseStorage, key: StorageKey)

Initialize the f s m context.

Parameters:

Name Type Description Default
storage BaseStorage

BaseStorage instance to process.

 required
key StorageKey

Storage key.

 required
Source code in src/pyromax/fsm/context.py 
 9
10
11
12
13
14
15
16
17
18
def __init__(self, storage: BaseStorage, key: StorageKey) -> None:
    """Initialize the f s m context.

    :param storage: BaseStorage instance to process.
    :type storage: BaseStorage
    :param key: Storage key.
    :type key: StorageKey
    """
    self.storage = storage
    self.key = key

storage instance-attribute 

storage = storage

key instance-attribute 

key = key

set_state async 

set_state(state: StateType = None) -> None

Set state.

Parameters:

Name Type Description Default
state StateType

FSM state.

 None
Source code in src/pyromax/fsm/context.py 
20
21
22
23
24
25
26
async def set_state(self, state: StateType = None) -> None:
    """Set state.

    :param state: FSM state.
    :type state: StateType
    """
    await self.storage.set_state(key=self.key, state=state)

get_state async 

get_state() -> str | None

Retrieve state.

Returns:

Type Description
str | None

The resulting str | None value.
Source code in src/pyromax/fsm/context.py 
28
29
30
31
32
33
34
async def get_state(self) -> str | None:
    """Retrieve state.

    :returns: The resulting str | None value.
    :rtype: str | None
    """
    return await self.storage.get_state(key=self.key)

set_data async 

set_data(data: Mapping[str, Any]) -> None

Set data.

Parameters:

Name Type Description Default
data Mapping[str, Any]

Contextual data passed through the processing pipeline.

 required
Source code in src/pyromax/fsm/context.py 
36
37
38
39
40
41
42
async def set_data(self, data: Mapping[str, Any]) -> None:
    """Set data.

    :param data: Contextual data passed through the processing pipeline.
    :type data: Mapping[str, Any]
    """
    await self.storage.set_data(key=self.key, data=data)

get_data async 

get_data() -> dict[str, Any]

Retrieve data.

Returns:

Type Description
dict[str, Any]

The resulting dict[str, Any] value.
Source code in src/pyromax/fsm/context.py 
44
45
46
47
48
49
50
async def get_data(self) -> dict[str, Any]:
    """Retrieve data.

    :returns: The resulting dict[str, Any] value.
    :rtype: dict[str, Any]
    """
    return await self.storage.get_data(key=self.key)

get_value async 

get_value(key: str) -> Any | None

get_value(key: str, default: Any) -> Any

get_value(
    key: str, default: Any | None = None
) -> Any | None

Retrieve value.

Parameters:

Name Type Description Default
key str

Storage key.

 required
default Any | None

The default value.

 None

Returns:

Type Description
Any | None

The resulting Any | None value.
Source code in src/pyromax/fsm/context.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
async def get_value(self, key: str, default: Any | None = None) -> Any | None:
    """Retrieve value.

    :param key: Storage key.
    :type key: str
    :param default: The default value.
    :type default: Any | None
    :returns: The resulting Any | None value.
    :rtype: Any | None
    """
    return await self.storage.get_value(
        storage_key=self.key, dict_key=key, default=default
    )

update_data async 

update_data(
    data: Mapping[str, Any] | None = None, **kwargs: Any
) -> dict[str, Any]

Update data.

Parameters:

Name Type Description Default
data Mapping[str, Any] | None

Contextual data passed through the processing pipeline.

 None
kwargs Any

Keyword arguments forwarded to the wrapped callable.

 {}

Returns:

Type Description
dict[str, Any]

The resulting dict[str, Any] value.
Source code in src/pyromax/fsm/context.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
async def update_data(
    self,
    data: Mapping[str, Any] | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """Update data.

    :param data: Contextual data passed through the processing pipeline.
    :type data: Mapping[str, Any] | None
    :param kwargs: Keyword arguments forwarded to the wrapped callable.
    :type kwargs: Any
    :returns: The resulting dict[str, Any] value.
    :rtype: dict[str, Any]
    """
    if data:
        kwargs.update(data)
    return await self.storage.update_data(key=self.key, data=kwargs)

clear async 

clear() -> None

Clear.

Source code in src/pyromax/fsm/context.py 
108
109
110
111
112
async def clear(self) -> None:
    """Clear.
    """
    await self.set_state(state=None)
    await self.set_data({})

State 

State(
    state: str | None = None, group_name: str | None = None
)

State object

Initialize the state.

Parameters:

Name Type Description Default
state str | None

FSM state.

 None
group_name str | None

The group name value.

 None
Source code in src/pyromax/fsm/state.py 
19
20
21
22
23
24
25
26
27
28
29
def __init__(self, state: str | None = None, group_name: str | None = None) -> None:
    """Initialize the state.

    :param state: FSM state.
    :type state: str | None
    :param group_name: The group name value.
    :type group_name: str | None
    """
    self._state = state
    self._group_name = group_name
    self._group: type[StatesGroup] | None = None

group property 

group: type[StatesGroup]

Group.

Returns:

Type Description
'type[StatesGroup]'

The resulting 'type[StatesGroup]' value.

Raises:

Type Description
RuntimeError

If the requested action cannot be completed.

state property 

state: str | None

State.

Returns:

Type Description
str | None

The resulting str | None value.

__repr__ class-attribute instance-attribute 

__repr__ = __str__

set_parent 

set_parent(group: type[StatesGroup]) -> None

Set parent.

Parameters:

Name Type Description Default
group type[StatesGroup]

'type[StatesGroup]' instance to process.

 required

Raises:

Type Description
ValueError

If the requested action cannot be completed.
Source code in src/pyromax/fsm/state.py 
63
64
65
66
67
68
69
70
71
72
73
def set_parent(self, group: "type[StatesGroup]") -> None:
    """Set parent.

    :param group: 'type[StatesGroup]' instance to process.
    :type group: 'type[StatesGroup]'
    :raises ValueError: If the requested action cannot be completed.
    """
    if not issubclass(group, StatesGroup):
        msg = "Group must be subclass of StatesGroup"
        raise ValueError(msg)
    self._group = group

__set_name__ 

__set_name__(owner: type[StatesGroup], name: str) -> None

Set name.

Parameters:

Name Type Description Default
owner type[StatesGroup]

'type[StatesGroup]' instance to process.

 required
name str

The name value.

 required
Source code in src/pyromax/fsm/state.py 
75
76
77
78
79
80
81
82
83
84
85
def __set_name__(self, owner: "type[StatesGroup]", name: str) -> None:
    """Set name.

    :param owner: 'type[StatesGroup]' instance to process.
    :type owner: 'type[StatesGroup]'
    :param name: The name value.
    :type name: str
    """
    if self._state is None:
        self._state = name
    self.set_parent(owner)

__str__ 

__str__() -> str

Str.

Returns:

Type Description
str

The resulting str value.
Source code in src/pyromax/fsm/state.py 
87
88
89
90
91
92
93
def __str__(self) -> str:
    """Str.

    :returns: The resulting str value.
    :rtype: str
    """
    return f"<State '{self.state or ''}'>"

__call__ 

__call__(event: MaxObject, data: DataDict) -> bool

Invoke the state.

Parameters:

Name Type Description Default
event MaxObject

Incoming event to process.

 required
data DataDict

Contextual data passed through the processing pipeline.

 required

Returns:

Type Description
bool

True when the requested condition is satisfied; otherwise False.
Source code in src/pyromax/fsm/state.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def __call__(self, event: MaxObject, data: DataDict) -> bool:
    """Invoke the state.

    :param event: Incoming event to process.
    :type event: MaxObject
    :param data: Contextual data passed through the processing pipeline.
    :type data: DataDict
    :returns: True when the requested condition is satisfied; otherwise False.
    :rtype: bool
    """
    raw_state = data.get(RawState)
    if self.state == "*":
        return True
    return raw_state == self.state

__eq__ 

__eq__(other: object) -> bool

Eq.

Parameters:

Name Type Description Default
other object

object instance to process.

 required

Returns:

Type Description
bool

True when the requested condition is satisfied; otherwise False.
Source code in src/pyromax/fsm/state.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
def __eq__(self, other: object) -> bool:
    """Eq.

    :param other: object instance to process.
    :type other: object
    :returns: True when the requested condition is satisfied; otherwise False.
    :rtype: bool
    """
    if isinstance(other, self.__class__):
        return self.state == other.state
    if isinstance(other, str):
        return self.state == other
    return NotImplemented

__hash__ 

__hash__() -> int

Hash.

Returns:

Type Description
int

The resulting int value.
Source code in src/pyromax/fsm/state.py 
126
127
128
129
130
131
132
def __hash__(self) -> int:
    """Hash.

    :returns: The resulting int value.
    :rtype: int
    """
    return hash(self.state)

StatesGroup

get_root classmethod 

get_root() -> type[StatesGroup]

Retrieve root.

Returns:

Type Description
type['StatesGroup']

The resulting type['StatesGroup'] value.
Source code in src/pyromax/fsm/state.py 
275
276
277
278
279
280
281
282
283
284
@classmethod
def get_root(cls) -> type["StatesGroup"]:
    """Retrieve root.

    :returns: The resulting type['StatesGroup'] value.
    :rtype: type['StatesGroup']
    """
    if cls.__parent__ is None:
        return cls
    return cls.__parent__.get_root()

__call__ 

__call__(event: MaxObject, data: DataDict) -> bool

Invoke the states group.

Parameters:

Name Type Description Default
event MaxObject

Incoming event to process.

 required
data DataDict

Contextual data passed through the processing pipeline.

 required

Returns:

Type Description
bool

True when the requested condition is satisfied; otherwise False.
Source code in src/pyromax/fsm/state.py 
286
287
288
289
290
291
292
293
294
295
296
297
def __call__(self, event: MaxObject, data: DataDict) -> bool:
    """Invoke the states group.

    :param event: Incoming event to process.
    :type event: MaxObject
    :param data: Contextual data passed through the processing pipeline.
    :type data: DataDict
    :returns: True when the requested condition is satisfied; otherwise False.
    :rtype: bool
    """
    raw_state = data.get(RawState)
    return raw_state in type(self).__all_states_names__

__str__ 

__str__() -> str

Str.

Returns:

Type Description
str

The resulting str value.
Source code in src/pyromax/fsm/state.py 
299
300
301
302
303
304
305
def __str__(self) -> str:
    """Str.

    :returns: The resulting str value.
    :rtype: str
    """
    return f"StatesGroup {type(self).__full_group_name__}"

FSMStrategy

Bases: Enum

FSM strategy for storage key generation.

USER_IN_CHAT class-attribute instance-attribute 

USER_IN_CHAT = auto()

State will be stored for each user in chat.

CHAT class-attribute instance-attribute 

CHAT = auto()

State will be stored for each chat globally without separating by users.

GLOBAL_USER class-attribute instance-attribute 

GLOBAL_USER = auto()

State will be stored globally for each user globally.

BaseStorage

Bases: ABC

Base class for all FSM storages

set_state abstractmethod async 

set_state(key: StorageKey, state: StateType = None) -> None

Set state for specified key

Parameters:

Name Type Description Default
key StorageKey

storage key

 required
state StateType

new state

 None
Source code in src/pyromax/fsm/storage/base.py 
127
128
129
130
131
132
133
134
135
136
@abstractmethod
async def set_state(self, key: StorageKey, state: StateType = None) -> None:
    """Set state for specified key

    :param key: storage key
    :param state: new state

    :type key: StorageKey
    :type state: StateType
    """

get_state abstractmethod async 

get_state(key: StorageKey) -> str | None

Get key state

Parameters:

Name Type Description Default
key StorageKey

storage key

 required

Returns:

Type Description
str | None

The resulting str | None value.
Source code in src/pyromax/fsm/storage/base.py 
138
139
140
141
142
143
144
145
146
147
148
@abstractmethod
async def get_state(self, key: StorageKey) -> str | None:
    """Get key state

    :param key: storage key
    :return: current state

    :type key: StorageKey
    :returns: The resulting str | None value.
    :rtype: str | None
    """

set_data abstractmethod async 

set_data(key: StorageKey, data: Mapping[str, Any]) -> None

Write data (replace)

Parameters:

Name Type Description Default
key StorageKey

storage key

 required
data Mapping[str, Any]

new data

 required
Source code in src/pyromax/fsm/storage/base.py 
150
151
152
153
154
155
156
157
158
159
@abstractmethod
async def set_data(self, key: StorageKey, data: Mapping[str, Any]) -> None:
    """Write data (replace)

    :param key: storage key
    :param data: new data

    :type key: StorageKey
    :type data: Mapping[str, Any]
    """

get_data abstractmethod async 

get_data(key: StorageKey) -> dict[str, Any]

Get current data for key

Parameters:

Name Type Description Default
key StorageKey

storage key

 required

Returns:

Type Description
dict[str, Any]

The resulting dict[str, Any] value.
Source code in src/pyromax/fsm/storage/base.py 
161
162
163
164
165
166
167
168
169
170
171
@abstractmethod
async def get_data(self, key: StorageKey) -> dict[str, Any]:
    """Get current data for key

    :param key: storage key
    :return: current data

    :type key: StorageKey
    :returns: The resulting dict[str, Any] value.
    :rtype: dict[str, Any]
    """

get_value async 

get_value(
    storage_key: StorageKey, dict_key: str
) -> Any | None

get_value(
    storage_key: StorageKey, dict_key: str, default: Any
) -> Any

get_value(
    storage_key: StorageKey,
    dict_key: str,
    default: Any | None = None,
) -> Any | None

Retrieve value.

Parameters:

Name Type Description Default
storage_key StorageKey

StorageKey instance to process.

 required
dict_key str

The dict key value.

 required
default Any | None

The default value.

 None

Returns:

Type Description
Any | None

The resulting Any | None value.
Source code in src/pyromax/fsm/storage/base.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
async def get_value(
    self,
    storage_key: StorageKey,
    dict_key: str,
    default: Any | None = None,
) -> Any | None:
    """Retrieve value.

    :param storage_key: StorageKey instance to process.
    :type storage_key: StorageKey
    :param dict_key: The dict key value.
    :type dict_key: str
    :param default: The default value.
    :type default: Any | None
    :returns: The resulting Any | None value.
    :rtype: Any | None
    """
    data = await self.get_data(storage_key)
    return data.get(dict_key, default)

update_data async 

update_data(
    key: StorageKey, data: Mapping[str, Any]
) -> dict[str, Any]

Update date in the storage for key (like dict.update)

Parameters:

Name Type Description Default
key StorageKey

storage key

 required
data Mapping[str, Any]

partial data

 required

Returns:

Type Description
dict[str, Any]

The resulting dict[str, Any] value.
Source code in src/pyromax/fsm/storage/base.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
async def update_data(
    self, key: StorageKey, data: Mapping[str, Any]
) -> dict[str, Any]:
    """Update date in the storage for key (like dict.update)

    :param key: storage key
    :param data: partial data
    :return: new data

    :type key: StorageKey
    :type data: Mapping[str, Any]
    :returns: The resulting dict[str, Any] value.
    :rtype: dict[str, Any]
    """
    current_data = await self.get_data(key=key)
    current_data.update(data)
    await self.set_data(key=key, data=current_data)
    return current_data.copy()

close abstractmethod async 

close() -> None

Close storage (database connection, file or etc.)

Source code in src/pyromax/fsm/storage/base.py 
244
245
246
247
@abstractmethod
async def close(self) -> None:  # pragma: no cover
    """Close storage (database connection, file or etc.)
    """

MemoryStorage 

MemoryStorage()

Bases: BaseStorage

Default FSM storage; uses a regular :class:dict to store data and does not persist it across restarts.

.. warning::

This storage is not recommended for production use, as all data is lost
when the bot restarts

Initialize the memory storage.

Source code in src/pyromax/fsm/storage/memory.py 
36
37
38
39
40
41
def __init__(self) -> None:
    """Initialize the memory storage.
    """
    self.storage: defaultdict[StorageKey, MemoryStorageRecord] = defaultdict(
        MemoryStorageRecord,
    )

storage instance-attribute 

storage: defaultdict[StorageKey, MemoryStorageRecord] = (
    defaultdict(MemoryStorageRecord)
)

close async 

close() -> None

Close.

Source code in src/pyromax/fsm/storage/memory.py 
43
44
45
46
async def close(self) -> None:
    """Close.
    """
    pass

set_state async 

set_state(key: StorageKey, state: StateType = None) -> None

Set state.

Parameters:

Name Type Description Default
key StorageKey

Storage key.

 required
state StateType

FSM state.

 None
Source code in src/pyromax/fsm/storage/memory.py 
48
49
50
51
52
53
54
55
56
async def set_state(self, key: StorageKey, state: StateType = None) -> None:
    """Set state.

    :param key: Storage key.
    :type key: StorageKey
    :param state: FSM state.
    :type state: StateType
    """
    self.storage[key].state = state.state if isinstance(state, State) else state

get_state async 

get_state(key: StorageKey) -> str | None

Retrieve state.

Parameters:

Name Type Description Default
key StorageKey

Storage key.

 required

Returns:

Type Description
str | None

The resulting str | None value.
Source code in src/pyromax/fsm/storage/memory.py 
58
59
60
61
62
63
64
65
66
async def get_state(self, key: StorageKey) -> str | None:
    """Retrieve state.

    :param key: Storage key.
    :type key: StorageKey
    :returns: The resulting str | None value.
    :rtype: str | None
    """
    return self.storage[key].state

set_data async 

set_data(key: StorageKey, data: Mapping[str, Any]) -> None

Set data.

Parameters:

Name Type Description Default
key StorageKey

Storage key.

 required
data Mapping[str, Any]

Contextual data passed through the processing pipeline.

 required

Raises:

Type Description
DataNotDictLikeError

If the requested action cannot be completed.
Source code in src/pyromax/fsm/storage/memory.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
async def set_data(self, key: StorageKey, data: Mapping[str, Any]) -> None:
    """Set data.

    :param key: Storage key.
    :type key: StorageKey
    :param data: Contextual data passed through the processing pipeline.
    :type data: Mapping[str, Any]
    :raises DataNotDictLikeError: If the requested action cannot be completed.
    """
    if not isinstance(data, dict):
        msg = f"Data must be a dict or dict-like object, got {type(data).__name__}"
        raise DataNotDictLikeError(msg)
    self.storage[key].data = data.copy()

get_data async 

get_data(key: StorageKey) -> dict[str, Any]

Retrieve data.

Parameters:

Name Type Description Default
key StorageKey

Storage key.

 required

Returns:

Type Description
dict[str, Any]

The resulting dict[str, Any] value.
Source code in src/pyromax/fsm/storage/memory.py 
82
83
84
85
86
87
88
89
90
async def get_data(self, key: StorageKey) -> dict[str, Any]:
    """Retrieve data.

    :param key: Storage key.
    :type key: StorageKey
    :returns: The resulting dict[str, Any] value.
    :rtype: dict[str, Any]
    """
    return self.storage[key].data.copy()

get_value async 

get_value(
    storage_key: StorageKey, dict_key: str
) -> Any | None

get_value(
    storage_key: StorageKey, dict_key: str, default: Any
) -> Any

get_value(
    storage_key: StorageKey,
    dict_key: str,
    default: Any | None = None,
) -> Any | None

Retrieve value.

Parameters:

Name Type Description Default
storage_key StorageKey

StorageKey instance to process.

 required
dict_key str

The dict key value.

 required
default Any | None

The default value.

 None

Returns:

Type Description
Any | None

The resulting Any | None value.
Source code in src/pyromax/fsm/storage/memory.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
async def get_value(
    self,
    storage_key: StorageKey,
    dict_key: str,
    default: Any | None = None,
) -> Any | None:
    """Retrieve value.

    :param storage_key: StorageKey instance to process.
    :type storage_key: StorageKey
    :param dict_key: The dict key value.
    :type dict_key: str
    :param default: The default value.
    :type default: Any | None
    :returns: The resulting Any | None value.
    :rtype: Any | None
    """
    data = self.storage[storage_key].data
    return copy(data.get(dict_key, default))