Skip to content

fsm — FSM Manager and Garbage Collector

BaseStateManager

BaseStateManager

Base FSM state manager for a feature (draft).

Isolates specific feature data under the draft:<feature_key> key within the user's general FSM storage. This prevents collisions between different features operating in the same FSM session.

Parameters:

Name Type Description Default
state FSMContext

FSM context of the current user.

 required
feature_key str

Unique feature key (e.g., "booking", "profile").

 required
Example 
manager = BaseStateManager(state, feature_key="booking")
await manager.update(date="2024-01-15", time="14:00")
payload = await manager.get_payload()
# {"date": "2024-01-15", "time": "14:00"}
await manager.clear()
Source code in src/codex_bot/fsm/state_manager.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class BaseStateManager:
    """
    Base FSM state manager for a feature (draft).

    Isolates specific feature data under the `draft:<feature_key>` key
    within the user's general FSM storage. This prevents collisions
    between different features operating in the same FSM session.

    Args:
        state: FSM context of the current user.
        feature_key: Unique feature key (e.g., `"booking"`, `"profile"`).

    Example:
        ```python
        manager = BaseStateManager(state, feature_key="booking")
        await manager.update(date="2024-01-15", time="14:00")
        payload = await manager.get_payload()
        # {"date": "2024-01-15", "time": "14:00"}
        await manager.clear()
        ```
    """

    def __init__(self, state: FSMContext, feature_key: str) -> None:
        self.state = state
        self.storage_key = f"draft:{feature_key}"

    async def get_payload(self) -> dict[str, Any]:
        """
        Returns all data of the feature draft.

        Returns:
            Dictionary with current data. Empty dictionary if no data exists.
        """
        result = await StateHelper.get_value(self.state, self.storage_key, {})
        return dict(result) if isinstance(result, dict) else {}

    async def update(self, **kwargs: Any) -> dict[str, Any]:
        """
        Updates the draft with the passed fields (partial update).

        Args:
            **kwargs: Fields to update.

        Returns:
            Updated dictionary with draft data.

        Example:
            ```python
            await manager.update(name="Alice", age=30)
            ```
        """
        current = await self.get_payload()
        current.update(kwargs)
        await StateHelper.update_value(self.state, self.storage_key, current)
        return current

    async def clear(self) -> None:
        """Completely removes the draft key from FSM storage.

        Unlike resetting to ``{key: {}}``, it completely removes the key,
        leaving no empty "zombie dictionaries" in Redis.
        Does not affect data of other features in the same FSM storage.
        """
        await StateHelper.clear_key(self.state, self.storage_key)

    async def set_value(self, key: str, value: Any) -> None:
        """
        Sets one specific value.

        Args:
            key: Field key.
            value: Value to save.
        """
        await self.update(**{key: value})

    async def get_value(self, key: str, default: Any = None) -> Any:
        """
        Returns a specific value from the draft.

        Args:
            key: Field key.
            default: Default value if the key is not found.

        Returns:
            Field value or default.
        """
        payload = await self.get_payload()
        return payload.get(key, default)

Functions

clear() async

Completely removes the draft key from FSM storage.

Unlike resetting to {key: {}}, it completely removes the key, leaving no empty "zombie dictionaries" in Redis. Does not affect data of other features in the same FSM storage.

Source code in src/codex_bot/fsm/state_manager.py 
73
74
75
76
77
78
79
80
async def clear(self) -> None:
    """Completely removes the draft key from FSM storage.

    Unlike resetting to ``{key: {}}``, it completely removes the key,
    leaving no empty "zombie dictionaries" in Redis.
    Does not affect data of other features in the same FSM storage.
    """
    await StateHelper.clear_key(self.state, self.storage_key)

get_payload() async

Returns all data of the feature draft.

Returns:

Type Description
dict[str, Any]

Dictionary with current data. Empty dictionary if no data exists.
Source code in src/codex_bot/fsm/state_manager.py 
43
44
45
46
47
48
49
50
51
async def get_payload(self) -> dict[str, Any]:
    """
    Returns all data of the feature draft.

    Returns:
        Dictionary with current data. Empty dictionary if no data exists.
    """
    result = await StateHelper.get_value(self.state, self.storage_key, {})
    return dict(result) if isinstance(result, dict) else {}

get_value(key, default=None) async

Returns a specific value from the draft.

Parameters:

Name Type Description Default
key str

Field key.

 required
default Any

Default value if the key is not found.

 None

Returns:

Type Description
Any

Field value or default.
Source code in src/codex_bot/fsm/state_manager.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
async def get_value(self, key: str, default: Any = None) -> Any:
    """
    Returns a specific value from the draft.

    Args:
        key: Field key.
        default: Default value if the key is not found.

    Returns:
        Field value or default.
    """
    payload = await self.get_payload()
    return payload.get(key, default)

set_value(key, value) async

Sets one specific value.

Parameters:

Name Type Description Default
key str

Field key.

 required
value Any

Value to save.

 required
Source code in src/codex_bot/fsm/state_manager.py 
82
83
84
85
86
87
88
89
90
async def set_value(self, key: str, value: Any) -> None:
    """
    Sets one specific value.

    Args:
        key: Field key.
        value: Value to save.
    """
    await self.update(**{key: value})

update(**kwargs) async

Updates the draft with the passed fields (partial update).

Parameters:

Name Type Description Default
**kwargs Any

Fields to update.

 {}

Returns:

Type Description
dict[str, Any]

Updated dictionary with draft data.
Example 
await manager.update(name="Alice", age=30)
Source code in src/codex_bot/fsm/state_manager.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
async def update(self, **kwargs: Any) -> dict[str, Any]:
    """
    Updates the draft with the passed fields (partial update).

    Args:
        **kwargs: Fields to update.

    Returns:
        Updated dictionary with draft data.

    Example:
        ```python
        await manager.update(name="Alice", age=30)
        ```
    """
    current = await self.get_payload()
    current.update(kwargs)
    await StateHelper.update_value(self.state, self.storage_key, current)
    return current

GarbageStateRegistry

GarbageStateRegistry

Registry of FSM states where text messages are considered garbage.

Global registry: states are registered once at startup, the filter checks the user's current state with each message.

Supports registration of: - a single State object - an entire StatesGroup (all states in the group) - a string (state name) - a list / tuple / set of the above types

Example 
GarbageStateRegistry.register(MyFeatureStates)          # entire group
GarbageStateRegistry.register(MyFeatureStates.waiting)  # single state
GarbageStateRegistry.register(["Feature1:main", "Feature2:step1"])
Source code in src/codex_bot/fsm/garbage_collector.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class GarbageStateRegistry:
    """
    Registry of FSM states where text messages are considered garbage.

    Global registry: states are registered once at startup,
    the filter checks the user's current state with each message.

    Supports registration of:
    - a single `State` object
    - an entire `StatesGroup` (all states in the group)
    - a string (state name)
    - a list / tuple / set of the above types

    Example:
        ```python
        GarbageStateRegistry.register(MyFeatureStates)          # entire group
        GarbageStateRegistry.register(MyFeatureStates.waiting)  # single state
        GarbageStateRegistry.register(["Feature1:main", "Feature2:step1"])
        ```
    """

    _states: set[str] = set()

    @classmethod
    def register(
        cls,
        state: State | StatesGroup | type[StatesGroup] | str | list[Any] | tuple[Any, ...] | set[Any],
    ) -> None:
        """
        Registers state(s) as garbage.

        Args:
            state: State, StatesGroup, string, or an iterable of them.
        """
        if isinstance(state, list | tuple | set):
            for s in state:
                cls.register(s)
            return

        # StatesGroup class (not instance) — take all its states
        if isinstance(state, type) and issubclass(state, StatesGroup):
            for s in state.__all_states__:
                cls.register(s)
            return

        # StatesGroup instance — take __all_states__ (Aiogram 3)
        if hasattr(state, "__all_states__"):
            for s in state.__all_states__:
                cls.register(s)
            return

        # State object or string
        state_name = state.state if isinstance(state, State) else str(state)
        if state_name:
            cls._states.add(state_name)

    @classmethod
    def is_garbage(cls, state_name: str | None) -> bool:
        """
        Checks if a state is registered as garbage.

        Args:
            state_name: String name of the current FSM state.

        Returns:
            True if the state is registered as garbage.
        """
        if state_name is None:
            return False
        return state_name in cls._states

    @classmethod
    def registered_states(cls) -> frozenset[str]:
        """
        Returns all registered garbage states (read-only).

        Returns:
            Frozenset of string state names.
        """
        return frozenset(cls._states)

Functions

is_garbage(state_name) classmethod

Checks if a state is registered as garbage.

Parameters:

Name Type Description Default
state_name str | None

String name of the current FSM state.

 required

Returns:

Type Description
bool

True if the state is registered as garbage.
Source code in src/codex_bot/fsm/garbage_collector.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
@classmethod
def is_garbage(cls, state_name: str | None) -> bool:
    """
    Checks if a state is registered as garbage.

    Args:
        state_name: String name of the current FSM state.

    Returns:
        True if the state is registered as garbage.
    """
    if state_name is None:
        return False
    return state_name in cls._states

register(state) classmethod

Registers state(s) as garbage.

Parameters:

Name Type Description Default
state State | StatesGroup | type[StatesGroup] | str | list[Any] | tuple[Any, ...] | set[Any]

State, StatesGroup, string, or an iterable of them.

 required
Source code in src/codex_bot/fsm/garbage_collector.py 
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@classmethod
def register(
    cls,
    state: State | StatesGroup | type[StatesGroup] | str | list[Any] | tuple[Any, ...] | set[Any],
) -> None:
    """
    Registers state(s) as garbage.

    Args:
        state: State, StatesGroup, string, or an iterable of them.
    """
    if isinstance(state, list | tuple | set):
        for s in state:
            cls.register(s)
        return

    # StatesGroup class (not instance) — take all its states
    if isinstance(state, type) and issubclass(state, StatesGroup):
        for s in state.__all_states__:
            cls.register(s)
        return

    # StatesGroup instance — take __all_states__ (Aiogram 3)
    if hasattr(state, "__all_states__"):
        for s in state.__all_states__:
            cls.register(s)
        return

    # State object or string
    state_name = state.state if isinstance(state, State) else str(state)
    if state_name:
        cls._states.add(state_name)

registered_states() classmethod

Returns all registered garbage states (read-only).

Returns:

Type Description
frozenset[str]

Frozenset of string state names.
Source code in src/codex_bot/fsm/garbage_collector.py 
106
107
108
109
110
111
112
113
114
@classmethod
def registered_states(cls) -> frozenset[str]:
    """
    Returns all registered garbage states (read-only).

    Returns:
        Frozenset of string state names.
    """
    return frozenset(cls._states)

IsGarbageStateFilter

IsGarbageStateFilter

Bases: Filter

Aiogram filter: True if the user's current FSM state is garbage.

Used in common_fsm_router for automatic deletion of unwanted text messages.

Example 
@router.message(F.text, IsGarbageStateFilter())
async def delete_garbage(message: Message, state: FSMContext):
    await message.delete()
Source code in src/codex_bot/fsm/garbage_collector.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class IsGarbageStateFilter(Filter):
    """
    Aiogram filter: True if the user's current FSM state is garbage.

    Used in common_fsm_router for automatic deletion of unwanted text messages.

    Example:
        ```python
        @router.message(F.text, IsGarbageStateFilter())
        async def delete_garbage(message: Message, state: FSMContext):
            await message.delete()
        ```
    """

    async def __call__(self, message: Message, state: FSMContext) -> bool:
        """
        Args:
            message: Incoming message.
            state: User's FSM context.

        Returns:
            True if the current state is registered as garbage.
        """
        current_state = await state.get_state()
        return GarbageStateRegistry.is_garbage(current_state)

Functions

__call__(message, state) async

Parameters:

Name Type Description Default
message Message

Incoming message.

 required
state FSMContext

User's FSM context.

 required

Returns:

Type Description
bool

True if the current state is registered as garbage.
Source code in src/codex_bot/fsm/garbage_collector.py 
131
132
133
134
135
136
137
138
139
140
141
async def __call__(self, message: Message, state: FSMContext) -> bool:
    """
    Args:
        message: Incoming message.
        state: User's FSM context.

    Returns:
        True if the current state is registered as garbage.
    """
    current_state = await state.get_state()
    return GarbageStateRegistry.is_garbage(current_state)

common_fsm_router

A ready-to-use router with a garbage-collector handler. Connect it to the main dispatcher:

from codex_bot.fsm import common_fsm_router
dp.include_router(common_fsm_router)