fsm — FSM Manager and Garbage Collector

BaseStateManager

BaseStateManager

Manages namespaced state persistence for individual bot features.

Each feature is allocated a dedicated sub-key (namespace) within the underlying FSM storage (e.g., Redis). This abstraction facilitates independent data management and prevents cross-feature state corruption.

Attributes:

Name	Type	Description
state		The raw FSMContext instance for the current session.
storage_key		The resolved namespaced key used for persistent storage.

Source code in src/codex_bot/fsm/state_manager.py

class BaseStateManager:
    """Manages namespaced state persistence for individual bot features.

    Each feature is allocated a dedicated sub-key (namespace) within the
    underlying FSM storage (e.g., Redis). This abstraction facilitates
    independent data management and prevents cross-feature state corruption.

    Attributes:
        state: The raw `FSMContext` instance for the current session.
        storage_key: The resolved namespaced key used for persistent storage.
    """

    def __init__(self, state: FSMContext, feature_key: str) -> None:
        """Initializes the BaseStateManager.

        Args:
            state: Instance of `FSMContext` from the active handler.
            feature_key: Unique identifier for the feature (e.g., "market", "quest").
        """
        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

__init__(state, feature_key)

Initializes the BaseStateManager.

Parameters:

Name	Type	Description	Default
state	FSMContext	Instance of FSMContext from the active handler.	required
feature_key	str	Unique identifier for the feature (e.g., "market", "quest").	required

Source code in src/codex_bot/fsm/state_manager.py

def __init__(self, state: FSMContext, feature_key: str) -> None:
    """Initializes the BaseStateManager.

    Args:
        state: Instance of `FSMContext` from the active handler.
        feature_key: Unique identifier for the feature (e.g., "market", "quest").
    """
    self.state = state
    self.storage_key = f"draft:{feature_key}"

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

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

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

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

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

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 for which the Garbage Collector is active.

When a user transitions to a state registered here, the system can automatically delete messages tracked in the _garbage_messages FSM data key.

Example

GarbageStateRegistry.register(MyStates.input_form)
GarbageStateRegistry.register([MyStates.step1, MyStates.step2])

Source code in src/codex_bot/fsm/garbage_collector.py

class GarbageStateRegistry:
    """Registry of FSM states for which the Garbage Collector is active.

    When a user transitions to a state registered here, the system can automatically
    delete messages tracked in the ``_garbage_messages`` FSM data key.

    Example:
        ```python
        GarbageStateRegistry.register(MyStates.input_form)
        GarbageStateRegistry.register([MyStates.step1, MyStates.step2])
        ```
    """

    _states: set[str] = set()

    @classmethod
    def register(cls, state_or_group: RegisterableState) -> None:
        """Adds a state or a group of states to the registry for automatic cleaning.

        Args:
            state_or_group: String state, State object, StatesGroup class, or a collection of these.
        """
        if isinstance(state_or_group, str):
            cls._states.add(state_or_group)
        elif isinstance(state_or_group, State):
            if state_or_group.state:
                cls._states.add(state_or_group.state)
        elif isinstance(state_or_group, type) and issubclass(state_or_group, StatesGroup):
            # In AIogram 3.x, we iterate over class attributes to find States
            for attr_name in dir(state_or_group):
                attr = getattr(state_or_group, attr_name)
                if isinstance(attr, State) and attr.state:
                    cls._states.add(attr.state)
        elif isinstance(state_or_group, list | tuple | set):
            for item in state_or_group:
                cls.register(item)

    @classmethod
    def is_garbage(cls, state: str | None) -> bool:
        """Checks if a state is in the registry.

        Args:
            state: Full state name string (e.g. "MyStates:step1").

        Returns:
            ``True`` if the state is registered for GC.
        """
        return state in cls._states

    @classmethod
    def registered_states(cls) -> frozenset[str]:
        """Returns a read-only view of registered states."""
        return frozenset(cls._states)

Functions

is_garbage(state) classmethod

Checks if a state is in the registry.

Parameters:

Name	Type	Description	Default
state	str | None	Full state name string (e.g. "MyStates:step1").	required

Returns:

Type	Description
bool	True if the state is registered for GC.

Source code in src/codex_bot/fsm/garbage_collector.py

@classmethod
def is_garbage(cls, state: str | None) -> bool:
    """Checks if a state is in the registry.

    Args:
        state: Full state name string (e.g. "MyStates:step1").

    Returns:
        ``True`` if the state is registered for GC.
    """
    return state in cls._states

register(state_or_group) classmethod

Adds a state or a group of states to the registry for automatic cleaning.

Parameters:

Name	Type	Description	Default
state_or_group	RegisterableState	String state, State object, StatesGroup class, or a collection of these.	required

Source code in src/codex_bot/fsm/garbage_collector.py

@classmethod
def register(cls, state_or_group: RegisterableState) -> None:
    """Adds a state or a group of states to the registry for automatic cleaning.

    Args:
        state_or_group: String state, State object, StatesGroup class, or a collection of these.
    """
    if isinstance(state_or_group, str):
        cls._states.add(state_or_group)
    elif isinstance(state_or_group, State):
        if state_or_group.state:
            cls._states.add(state_or_group.state)
    elif isinstance(state_or_group, type) and issubclass(state_or_group, StatesGroup):
        # In AIogram 3.x, we iterate over class attributes to find States
        for attr_name in dir(state_or_group):
            attr = getattr(state_or_group, attr_name)
            if isinstance(attr, State) and attr.state:
                cls._states.add(attr.state)
    elif isinstance(state_or_group, list | tuple | set):
        for item in state_or_group:
            cls.register(item)

registered_states() classmethod

Returns a read-only view of registered states.

Source code in src/codex_bot/fsm/garbage_collector.py

@classmethod
def registered_states(cls) -> frozenset[str]:
    """Returns a read-only view of registered states."""
    return frozenset(cls._states)

IsGarbageStateFilter

IsGarbageStateFilter

Bases: Filter

Aiogram filter to check if the current user state is registered for GC.

Used by the internal garbage collection middleware/handlers to decide when to trigger the cleaning process.

Source code in src/codex_bot/fsm/garbage_collector.py

class IsGarbageStateFilter(Filter):
    """Aiogram filter to check if the current user state is registered for GC.

    Used by the internal garbage collection middleware/handlers to decide
    when to trigger the cleaning process.
    """

    async def __call__(self, event: TelegramObject, state: FSMContext) -> bool:
        """Checks the current state against the registry.

        Returns:
            ``True`` if current state is in GarbageStateRegistry.
        """
        current_state = await state.get_state()
        return GarbageStateRegistry.is_garbage(current_state)

Functions

__call__(event, state) async

Checks the current state against the registry.

Returns:

Type	Description
bool	True if current state is in GarbageStateRegistry.

Source code in src/codex_bot/fsm/garbage_collector.py

async def __call__(self, event: TelegramObject, state: FSMContext) -> bool:
    """Checks the current state against the registry.

    Returns:
        ``True`` if current state is in GarbageStateRegistry.
    """
    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)