FSM

StateGroup

StateGroup

A group of related states.

StateGroup provides a way to organize states logically and access them as class attributes.

Examples:

class OrderStates(StateGroup): waiting_name = State() waiting_phone = State() confirm = State()

Access states

OrderStates.waiting_name #

Get all states

OrderStates.all_states() # [State, State, State]

Check if state is in group

OrderStates.waiting_name in OrderStates.all_states() # True

all_states classmethod

all_states()

Get all states in this group.

Returns:

Type	Description
list[State]	List of all State objects in this group.

Source code in vkflow\app\fsm\state.py

@classmethod
def all_states(cls) -> list[State]:
    """
    Get all states in this group.

    Returns:
        List of all State objects in this group.
    """
    return list(cls.__states__.values())

get_state classmethod

get_state(name)

Get a state by its short name.

Parameters:

Name	Type	Description	Default
name	str	The short name of the state (without group prefix).	required

Returns:

Type	Description
State | None	The State object or None if not found.

Source code in vkflow\app\fsm\state.py

@classmethod
def get_state(cls, name: str) -> State | None:
    """
    Get a state by its short name.

    Args:
        name: The short name of the state (without group prefix).

    Returns:
        The State object or None if not found.
    """
    return cls.__states__.get(name)

__contains__ classmethod

__contains__(item)

Check if a state belongs to this group.

Parameters:

Name	Type	Description	Default
item	State | str	State object or state name string.	required

Returns:

Type	Description
bool	True if the state is in this group.

Source code in vkflow\app\fsm\state.py

@classmethod
def __contains__(cls, item: State | str) -> bool:
    """
    Check if a state belongs to this group.

    Args:
        item: State object or state name string.

    Returns:
        True if the state is in this group.
    """
    if isinstance(item, State):
        return item in cls.__states__.values()
    if isinstance(item, str):
        if ":" in item:
            group_name, state_name = item.split(":", 1)
            return group_name == cls.__group_name__ and state_name in cls.__states__
        return item in cls.__states__
    return False

State

State dataclass

State(name=None)

Represents a state in FSM.

Can be used as: - Standalone state: state = State("waiting_name") - Part of StateGroup: waiting_name = State() inside class

Examples:

Standalone

my_state = State("waiting_input")

In StateGroup

class OrderStates(StateGroup): waiting_name = State() waiting_phone = State()

Access

OrderStates.waiting_name.name # "OrderStates:waiting_name"

Initialize a State.

Parameters:

Name	Type	Description	Default
name	str | None	Optional explicit name. If not provided, will be set automatically when used in StateGroup.	None

Source code in vkflow\app\fsm\state.py

def __init__(self, name: str | None = None):
    """
    Initialize a State.

    Args:
        name: Optional explicit name. If not provided, will be set
              automatically when used in StateGroup.
    """
    self._name = name
    self._group = None

name property

name

Full state name: "GroupName:state_name" or just "state_name".

Returns:

Type	Description
str	The full qualified name of the state.

state property

state

Short state name (without group).

Returns:

Type	Description
str	The state name without group prefix.

group property

group

The StateGroup this state belongs to.

Returns:

Type	Description
type | None	The StateGroup class or None if standalone.

FSMRouter

FSMRouter

FSMRouter(storage, *, strategy=USER_CHAT, name=None)

Router for FSM state handlers.

FSMRouter provides a standalone way to register and dispatch FSM state handlers, independent of the Cog system.

Features: - Register handlers with @router.state() decorator - Before/after state hooks - Automatic argument injection - Process messages through registered handlers

Examples:

from vkflow.app.fsm import FSMRouter, MemoryStorage, StateGroup, State

class OrderStates(StateGroup): waiting_name = State() waiting_phone = State()

storage = MemoryStorage() router = FSMRouter(storage)

@router.state(OrderStates.waiting_name) async def handle_name(ctx, msg): await ctx.update_data(name=msg.msg.text) await ctx.set_state(OrderStates.waiting_phone) await msg.answer("Enter phone:")

@router.state(OrderStates.waiting_phone) async def handle_phone(ctx, msg): data = await ctx.finish() await msg.answer(f"Order: {data['name']}, {msg.msg.text}")

Register in app

app.include_fsm_router(router)

Initialize FSMRouter.

Parameters:

Name	Type	Description	Default
storage	BaseStorage	FSM storage backend	required
strategy	KeyStrategy | str	Key generation strategy	USER_CHAT
name	str | None	Optional router name for debugging	None

Source code in vkflow\app\fsm\router.py

def __init__(
    self,
    storage: BaseStorage,
    *,
    strategy: KeyStrategy | str = KeyStrategy.USER_CHAT,
    name: str | None = None,
):
    """
    Initialize FSMRouter.

    Args:
        storage: FSM storage backend
        strategy: Key generation strategy
        name: Optional router name for debugging
    """
    self.storage = storage
    self.strategy = KeyStrategy(strategy) if isinstance(strategy, str) else strategy
    self.name = name or self.__class__.__name__

    self._handlers: dict[str, StateHandler] = {}
    self._before_hooks: list[Handler] = []
    self._after_hooks: list[Handler] = []

state

state(state)

Decorator to register a state handler.

Parameters:

Name	Type	Description	Default
state	State | str	State object or state name string	required

Returns:

Type	Description
Callable[[Handler], Handler]	Decorator function

Examples:

@router.state(OrderStates.waiting_name) async def handle_name(ctx: fsm.Context, msg: NewMessage): await ctx.update_data(name=msg.msg.text) await ctx.set_state(OrderStates.waiting_phone)

Source code in vkflow\app\fsm\router.py

def state(
    self,
    state: State | str,
) -> typing.Callable[[Handler], Handler]:
    """
    Decorator to register a state handler.

    Args:
        state: State object or state name string

    Returns:
        Decorator function

    Examples:
        @router.state(OrderStates.waiting_name)
        async def handle_name(ctx: fsm.Context, msg: NewMessage):
            await ctx.update_data(name=msg.msg.text)
            await ctx.set_state(OrderStates.waiting_phone)
    """

    def decorator(func: Handler) -> Handler:
        state_name = state.name if hasattr(state, "name") else str(state)
        self._handlers[state_name] = StateHandler(
            handler=func,
            state=state,
        )
        return func

    return decorator

before_state

before_state()

Decorator for before-state hook.

Hook is called before any state handler. If it returns False, the state handler will be skipped.

Examples:

@router.before_state() async def log_before(ctx, msg): logger.info(f"Processing state for {msg.msg.from_id}")

Source code in vkflow\app\fsm\router.py

def before_state(self) -> typing.Callable[[Handler], Handler]:
    """
    Decorator for before-state hook.

    Hook is called before any state handler. If it returns False,
    the state handler will be skipped.

    Examples:
        @router.before_state()
        async def log_before(ctx, msg):
            logger.info(f"Processing state for {msg.msg.from_id}")
    """

    def decorator(func: Handler) -> Handler:
        self._before_hooks.append(func)
        return func

    return decorator

after_state

after_state()

Decorator for after-state hook.

Hook is called after state handler completes (success or failure).

Examples:

@router.after_state() async def log_after(ctx, msg): logger.info(f"Finished processing for {msg.msg.from_id}")

Source code in vkflow\app\fsm\router.py

def after_state(self) -> typing.Callable[[Handler], Handler]:
    """
    Decorator for after-state hook.

    Hook is called after state handler completes (success or failure).

    Examples:
        @router.after_state()
        async def log_after(ctx, msg):
            logger.info(f"Finished processing for {msg.msg.from_id}")
    """

    def decorator(func: Handler) -> Handler:
        self._after_hooks.append(func)
        return func

    return decorator

get_handler

get_handler(state)

Get handler for a state.

Parameters:

Name	Type	Description	Default
state	str	State name	required

Returns:

Type	Description
StateHandler | None	StateHandler or None if not found

Source code in vkflow\app\fsm\router.py

def get_handler(self, state: str) -> StateHandler | None:
    """
    Get handler for a state.

    Args:
        state: State name

    Returns:
        StateHandler or None if not found
    """
    return self._handlers.get(state)

get_states

get_states()

Get all registered state names.

Returns:

Type	Description
list[str]	List of state name strings

Source code in vkflow\app\fsm\router.py

def get_states(self) -> list[str]:
    """
    Get all registered state names.

    Returns:
        List of state name strings
    """
    return list(self._handlers.keys())

process async

process(message)

Process a message through FSM handlers.

Checks if user is in a registered state and invokes the corresponding handler.

Parameters:

Name	Type	Description	Default
message	NewMessage | CallbackButtonPressed	Message to process	required

Returns:

Type	Description
bool	True if a handler was invoked, False otherwise

Source code in vkflow\app\fsm\router.py

async def process(
    self,
    message: NewMessage | CallbackButtonPressed,
) -> bool:
    """
    Process a message through FSM handlers.

    Checks if user is in a registered state and invokes
    the corresponding handler.

    Args:
        message: Message to process

    Returns:
        True if a handler was invoked, False otherwise
    """
    fsm_ctx = FSMContext.from_message(self.storage, message, strategy=self.strategy)

    current_state = await fsm_ctx.get_state()
    if current_state is None:
        return False

    handler = self._handlers.get(current_state)
    if handler is None:
        return False

    for hook in self._before_hooks:
        try:
            result = await self._invoke_hook(hook, fsm_ctx, message)
            if result is False:
                return False
        except Exception:
            logger.exception("Error in before_state hook")
            raise

    try:
        await handler.invoke(fsm_ctx, message)
    except Exception:
        logger.exception(f"Error in FSM handler for state {current_state}")
        raise
    finally:
        for hook in self._after_hooks:
            try:
                await self._invoke_hook(hook, fsm_ctx, message)
            except Exception:
                logger.exception("Error in after_state hook")

    return True

include_router

include_router(router)

Include another router's handlers.

Parameters:

Name	Type	Description	Default
router	FSMRouter	Router to include	required

Source code in vkflow\app\fsm\router.py

def include_router(self, router: FSMRouter) -> None:
    """
    Include another router's handlers.

    Args:
        router: Router to include
    """
    self._handlers.update(router._handlers)
    self._before_hooks.extend(router._before_hooks)
    self._after_hooks.extend(router._after_hooks)

FSMContext

Context module-attribute

Context = FSMContext

StateFilter

StateFilter dataclass

StateFilter(state, storage, *, strategy=USER_CHAT)

Bases: BaseFilter

Filter for checking FSM state before command execution.

StateFilter integrates FSM with the existing command system. It checks if user is in a specific state before allowing the command to execute.

Examples:

Filter for specific state

@app.command("next", filter=StateFilter(OrderStates.waiting_name, storage)) async def handle_name(ctx: NewMessage): ...

Filter for any state in group

@app.command("cancel", filter=StateFilter.any(OrderStates, storage)) async def cancel(ctx: NewMessage): ...

Filter for no active state

@app.command("start", filter=StateFilter.none(storage)) async def start(ctx: NewMessage): ...

Combine with other filters

combined = StateFilter(MyStates.waiting, storage) & SomeOtherFilter()

Create a StateFilter for specific state(s).

Parameters:

Name	Type	Description	Default
state	State | str | list[State | str]	Single state or list of states to match	required
storage	BaseStorage	FSM storage backend	required
strategy	KeyStrategy | str	Key generation strategy	USER_CHAT

Source code in vkflow\app\fsm\filter.py

def __init__(
    self,
    state: State | str | list[State | str],
    storage: BaseStorage,
    *,
    strategy: KeyStrategy | str = KeyStrategy.USER_CHAT,
):
    """
    Create a StateFilter for specific state(s).

    Args:
        state: Single state or list of states to match
        storage: FSM storage backend
        strategy: Key generation strategy
    """
    states = list(state) if isinstance(state, (list, tuple)) else [state]

    object.__setattr__(self, "states", states)
    object.__setattr__(self, "storage", storage)
    object.__setattr__(self, "strategy", strategy)
    object.__setattr__(self, "_check_none", False)

    self.__post_init__()

any classmethod

any(group, storage, *, strategy=USER_CHAT)

Create filter that matches ANY state in a StateGroup.

Parameters:

Name	Type	Description	Default
group	type	StateGroup class	required
storage	BaseStorage	FSM storage backend	required
strategy	KeyStrategy | str	Key generation strategy	USER_CHAT

Returns:

Type	Description
StateFilter	StateFilter that passes if user is in any state of the group.

Examples:

@app.command("cancel", filter=StateFilter.any(OrderStates, storage)) async def cancel(ctx: NewMessage): # Works when user is in ANY OrderStates state ...

Source code in vkflow\app\fsm\filter.py

@classmethod
def any(
    cls,
    group: type,  # StateGroup
    storage: BaseStorage,
    *,
    strategy: KeyStrategy | str = KeyStrategy.USER_CHAT,
) -> StateFilter:
    """
    Create filter that matches ANY state in a StateGroup.

    Args:
        group: StateGroup class
        storage: FSM storage backend
        strategy: Key generation strategy

    Returns:
        StateFilter that passes if user is in any state of the group.

    Examples:
        @app.command("cancel", filter=StateFilter.any(OrderStates, storage))
        async def cancel(ctx: NewMessage):
            # Works when user is in ANY OrderStates state
            ...
    """
    return cls(
        state=list(group.__states__.values()),
        storage=storage,
        strategy=strategy,
    )

none classmethod

none(storage, *, strategy=USER_CHAT)

Create filter that matches when NO state is active.

Parameters:

Name	Type	Description	Default
storage	BaseStorage	FSM storage backend	required
strategy	KeyStrategy | str	Key generation strategy	USER_CHAT

Returns:

Type	Description
StateFilter	StateFilter that passes only if user has no active state.

Examples:

@app.command("start", filter=StateFilter.none(storage)) async def start(ctx: NewMessage): # Only works when user is NOT in any state ...

Source code in vkflow\app\fsm\filter.py

@classmethod
def none(
    cls,
    storage: BaseStorage,
    *,
    strategy: KeyStrategy | str = KeyStrategy.USER_CHAT,
) -> StateFilter:
    """
    Create filter that matches when NO state is active.

    Args:
        storage: FSM storage backend
        strategy: Key generation strategy

    Returns:
        StateFilter that passes only if user has no active state.

    Examples:
        @app.command("start", filter=StateFilter.none(storage))
        async def start(ctx: NewMessage):
            # Only works when user is NOT in any state
            ...
    """
    filter_obj = cls(
        state=[],
        storage=storage,
        strategy=strategy,
    )
    object.__setattr__(filter_obj, "_check_none", True)
    return filter_obj

make_decision async

make_decision(ctx, **kwargs)

Check if the current state matches filter criteria.

Raises:

Type	Description
StopCurrentHandlingError	If state doesn't match

Source code in vkflow\app\fsm\filter.py

async def make_decision(self, ctx: NewMessage, **kwargs) -> None:
    """
    Check if the current state matches filter criteria.

    Raises:
        StopCurrentHandlingError: If state doesn't match
    """
    fsm_ctx = FSMContext.from_message(self.storage, ctx, strategy=self.strategy)

    current_state = await fsm_ctx.get_state()

    if self._check_none:
        if current_state is not None:
            raise StopCurrentHandlingError()
        return

    if current_state is None:
        raise StopCurrentHandlingError()

    for state in self.states:
        state_name = state.name if hasattr(state, "name") else str(state)
        if current_state == state_name:
            return

    raise StopCurrentHandlingError()

MemoryStorage

MemoryStorage

MemoryStorage(ttl=None)

Bases: BaseStorage

In-memory FSM storage implementation.

Stores FSM state and data in Python dictionaries. Data is lost when the bot restarts.

Features: - Thread-safe via asyncio.Lock - Optional TTL for automatic state expiration - Zero dependencies

Best for: - Development and testing - Small bots where persistence isn't critical - Stateless deployments

Examples:

Basic usage

storage = MemoryStorage()

With TTL (states expire after 1 hour)

storage = MemoryStorage(ttl=3600)

Use with FSMRouter

router = FSMRouter(storage)

Initialize MemoryStorage.

Parameters:

Name	Type	Description	Default
ttl	float | None	Optional time-to-live in seconds for states. If set, states older than TTL will be automatically cleared on next access. None means no expiration.	None

Source code in vkflow\app\fsm\storage\memory.py

def __init__(self, ttl: float | None = None):
    """
    Initialize MemoryStorage.

    Args:
        ttl: Optional time-to-live in seconds for states.
             If set, states older than TTL will be automatically
             cleared on next access. None means no expiration.
    """
    self._states: dict[str, str] = {}
    self._data: dict[str, dict[str, typing.Any]] = defaultdict(dict)
    self._timestamps: dict[str, float] = {}
    self._lock = asyncio.Lock()
    self._ttl = ttl

get_states_count

get_states_count()

Get number of active states (for debugging).

Returns:

Type	Description
int	Number of stored states.

Source code in vkflow\app\fsm\storage\memory.py

def get_states_count(self) -> int:
    """
    Get number of active states (for debugging).

    Returns:
        Number of stored states.
    """
    return len(self._states)

get_keys

get_keys()

Get all stored keys (for debugging).

Returns:

Type	Description
list[str]	List of all keys with active states.

Source code in vkflow\app\fsm\storage\memory.py

def get_keys(self) -> list[str]:
    """
    Get all stored keys (for debugging).

    Returns:
        List of all keys with active states.
    """
    return list(self._states.keys())