Exploring Finite State Machine in Aiogram 3: A Powerful Tool for Telegram Bot Development

Introduction

In the realm of Telegram bot development, the quest for efficiency and robustness is an ongoing endeavor. With each iteration, developers seek tools and frameworks that streamline the development process while ensuring the scalability and maintainability of their bots. In this pursuit, the integration of Finite State Machines (FSMs) has emerged as a powerful technique, providing a structured approach to managing bot behavior.

In this article, we delve into the integration of Finite State Machines within the Aiogram 3 framework, a powerful option for Telegram bot development in Python. Aiogram, renowned for its simplicity and flexibility, empowers developers to build feature-rich bots effortlessly. By using FSM, it can become even more seamless, offering developers a structured approach to handling complex bot behavior.

Understanding FSM in bots development

A Finite State Machine (FSM) is a computational model used to describe the behavior of a system by dividing it into a finite number of states, along with transitions between these states based on certain conditions or events.

At its core, an FSM consists of the following components:

1. States: These represent the distinct conditions or modes that the system can be in at any given time. Each state encapsulates a specific set of behaviors or actions that the system can perform while in that state.
2. Transitions: These define the conditions or events that trigger a change in state. When a transition occurs, the system moves from one state to another, potentially altering its behavior or internal configuration.
3. Inputs/Events: These are the stimuli or signals that drive the transitions between states. Inputs can come from various sources, such as user interactions, sensor readings, or external events.

In Aiogram 3, the integration of Finite State Machines offers developers a remarkable opportunity to manage the intricate flow of user inputs with ease. By guiding users through a sequence of defined states and handling each state separately, FSMs empower developers to create Telegram bots that gracefully navigate through complex interaction scenarios. This capability allows for the systematic handling of user inputs, ensuring a seamless conversational experience for bot users.

State group schema and sequential flow

To better comprehend the theoretical aspects of Finite State Machines (FSM) in Aiogram, let's embark on a practical journey. Imagine we're creating a form within our Telegram bot where we sequentially ask users a series of questions, concluding with a confirmation of the collected answers. In Aiogram, FSMs are represented as StatesGroups, serving as the foundation for organizing bot behavior into distinct states and managing transitions between them.

The schema below illustrates the sequential flow of our form within the FSM framework:

Starting with the creation of a StatesGroup, we establish a structured environment for managing the flow of user interactions. Each state within the StatesGroup corresponds to a specific stage in our form, guiding users through a sequence of questions and responses. As users progress through the form, transitions between states occur based on their inputs, ensuring a smooth and intuitive user experience.

Now, let's delve into the implementation details. Below, you'll find the code snippet illustrating the creation of the StatesGroup with Aiogram:

class About(StatesGroup):
    first_name = State()
    last_name = State()
    age = State()
    hobbies = State()

class Address(StatesGroup):
    city = State()
    street = State()
    house = State()

class Contact(StatesGroup):
    phone = State()
    email = State()

class Form(StatesGroup):
    about = About
    address = Address
    contact = Contact
    confirm = State()        

After defining the StatesGroup for our form, the next step is to establish handlers to manage user interactions. Typically, a conversation with the bot initiates with the /start command. Therefore, our first task is to handle this command and transition the user to the initial state of our form, Form.About.first_name.

Below is a snippet showcasing the implementation of the handler for the /start command and the transition operation to the first state:

@form_router.message(CommandStart())
async def command_start(message: Message, state: FSMContext) -> None:
    await state.set_state(Form.about.first_name)
    await message.answer(
        "Hello! What's your name?",
        reply_markup=ReplyKeyboardMarkup(
            keyboard=[
                [KeyboardButton(text="Skip")],
                [KeyboardButton(text="Cancel")],
            ],
            resize_keyboard=True,
        ),
    )        

We have added two buttons for this question - "Skip" and "Cancel", we will cover them later. After the user answers the first question, we need to store the answer in the storage and move on to the next step.

Code snippet to handler first state and transition to second state:

@form_router.message(Form.about.first_name)
async def process_first_name(message: Message, state: FSMContext) -> None:
    await state.update_data(first_name=message.text)
    await state.set_state(Form.about.last_name)
    await message.answer(
        "What's your last name?",
        reply_markup=ReplyKeyboardMarkup(
            keyboard=[
                [
                    KeyboardButton(text="Skip"),
                    KeyboardButton(text="Go back")
                ],
                [KeyboardButton(text="Cancel")],
            ],
            resize_keyboard=True,
        ),
    )        

In this handler, we've successfully stored the user's input for the first question and will now progress to the next question. Additionally, we'll incorporate an "Go back" button later, enabling users to make transition to the previous question. Following a consistent sequence, each subsequent handler will entail the same actions: saving the user's response, transitioning to the subsequent step, and posing the subsequent question. Upon reaching the final state, Form.confirm, users will be presented with all collected information for review.

Code snippet for email (last question) handler:

@form_router.message(Form.contact.email)
async def process_email(message: Message, state: FSMContext) -> None:
    await state.update_data(email=message.text)
    await state.set_state(Form.confirm)
    data: Dict[str, Any] = await state.get_data()
    await message.answer(
        f"Please confirm your data:\n"
        f"Name: {html.quote(data['first_name'])} "
        f"{html.quote(data['last_name'])}\n"
        f"Age: {html.quote(data['age'])}\n"
        f"Hobbies: {html.quote(data['hobbies'])}\n"
        f"City: {html.quote(data['city'])}\n"
        f"Street: {html.quote(data['street'])}\n"
        f"House: {html.quote(data['house'])}\n"
        f"Phone: {html.quote(data['phone'])}\n"
        f"Email: {html.quote(data['email'])}",
        reply_markup=ReplyKeyboardMarkup(
            keyboard=[
                [KeyboardButton(text="Approve"), KeyboardButton(text="Disapprove")],
                [KeyboardButton(text="Go back")],
                [KeyboardButton(text="Cancel")],
            ],
            resize_keyboard=True,
        ),
    )        

With the implementation of the described logic, our bot is now equipped to traverse through all defined states, guiding users through each step of the form.

Presentation of bot flow:

Transition to previous flow and flow cancelation

Incorporating various options to enhance user flexibility during Finite State Machine (FSM) integration in Aiogram is crucial for providing a smoother user experience. Among these options is the ability to terminate the flow, facilitated by the addition of a "Cancel" button. This button should be accessible from any state within the FSM, necessitating a single handler to manage its functionality. This handler will be responsible for clearing the current state and sending an appropriate message to the user, indicating the cancellation of the flow.

Code snippet for "Cancel" button handler:

@form_router.message(Command("cancel"))
@form_router.message(F.text.casefold() == "cancel")
async def cancel_handler(message: Message, state: FSMContext) -> None:
    """
    Allow user to cancel any action
    """
    current_state = await state.get_state()
    if current_state is None:
        return

    logging.info("Cancelling state %r", current_state)
    await state.clear()
    await message.answer(
        "Cancelled.",
        reply_markup=ReplyKeyboardRemove(),
    )
        

To introduce the "Go back" button functionality, enabling users to transition to the previous state, we need to implement an additional tool - a list containing information about the states.

For this, let's outline a data structure representing the state information. Code snippet for this:

@dataclass
class State:
    state_name: str
    state_question: str
    state_in_memory_name: str
    state_corresponding_button: str
    keyboard_buttons: list[list[KeyboardButton]]        

Each element of the list will be represented as dataclass with state name, question for this state, in memory representative value, corresponding button (we will need it for future step) and keyboard buttons that need to be sent to user with state question.

Code snippet for first state implementation in dataclass format:

State(
    "Form.About:first_name",
    "What's your first name?",
    "first_name",
    "First name",
    [
        [KeyboardButton(text="Skip")],
        [KeyboardButton(text="Cancel")],
    ],
),        

All following states will have similar structure, but with additional "Go back" button. Code snippet example for following states:

State(
    "Form.About:last_name",
    "What's your last name?",
    "last_name",
    "Last name",
    [
        [KeyboardButton(text="Skip"), KeyboardButton(text="Go back")],
        [KeyboardButton(text="Cancel")],
    ],
),        

Code snippet for last state:

State(
    "Form:confirm",
    "",
    "",
    "",
    [
        [KeyboardButton(text="Approve"),
         KeyboardButton(text="Disapprove")],
        [KeyboardButton(text="Go back")],
        [KeyboardButton(text="Cancel")],
    ],
),        

This list of states will allow us to track the state by name and manage all of its attributes to better understand it lets review code snippet for the "Go back" button handler:

@form_router.message(Command("go back"))
@form_router.message(F.text.casefold() == "go back")
async def go_back_handler(message: Message, state: FSMContext) -> None:
    current_state = await state.get_state()
    logging.info("Going back from %r", current_state)
    (
        previous_state,
        state_message,
        keyboard_buttons
    ) = get_previous_state(current_state)
    if previous_state is None:
        await message.answer(
            "You are already at the first step.",
        )
    else:
        await state.set_state(previous_state)
        await message.answer(
            state_message,
            reply_markup=ReplyKeyboardMarkup(
                keyboard=keyboard_buttons,
                resize_keyboard=True,
            ),
        )
        

In this handler we get the current_state and based on that we get the previous state information using the get_previous_state method.

Code snippet for get_r

def get_previous_state(current_state: str) -> str:
    current_state_index = next(
        (i for i, obj in enumerate(STATES_LIST) if obj.state_name == current_state),
        None,
    )
    previous_state_index = current_state_index - 1
    if previous_state_index < 0:
        return None, None
    return (
        STATES_LIST[previous_state_index].state_name,
        STATES_LIST[previous_state_index].state_question,
        STATES_LIST[previous_state_index].keyboard_buttons,
    )        

This method uses created STATES_LIST to receive current state (the state from which the user wants to go back) index and based on this returns information about previous state name, questions and keyboard buttons. After this we only need to set this state and send proper message to user. Implementation of this flow shown bellow.

Transition to any state in flow

Now when we can cancel flow and use transition to previous state we can try to implement even more complicated operations. For this lets implement confirmation for our form. Previously we have added "Approve" and "Disapprove" buttons for final step. Handler for "Approve" button is very simple.

@form_router.message(Form.confirm, F.text.casefold() == "approve")
async def process_confirm(message: Message, state: FSMContext) -> None:
    await state.clear()
    await message.answer(
        "Thank you for your Form!",
        reply_markup=ReplyKeyboardRemove(),
    )        

We will not save any customers data in this example, so all we need - close state and send appropriate message. To implement "Disapprove" function lets add another state - "confirm_reject", after it our StatesGroup looks like this:

class Form(StatesGroup):
    about = About
    address = Address
    contact = Contact
    confirm = State()
    confirm_reject = State()        

Handler for "Disapprove" button will set this state and send message with proposition to change any of inputed data.

@form_router.message(Form.confirm, F.text.casefold() == "disapprove")
async def process_dont_confirm(message: Message, state: FSMContext) -> None:
    await state.set_state(Form.confirm_reject)
    await message.answer(
        "Which data is incorrect?",
        reply_markup=ReplyKeyboardMarkup(
            keyboard=[
                [
                    KeyboardButton(text="First name"),
                    KeyboardButton(text="Last name"),
                    KeyboardButton(text="Age"),
                ],
                [
                    KeyboardButton(text="Hobbies"),
                    KeyboardButton(text="City"),
                    KeyboardButton(text="Street"),
                ],
                [
                    KeyboardButton(text="House"),
                    KeyboardButton(text="Phone"),
                    KeyboardButton(text="Email"),
                ],
                [
                    KeyboardButton(text="Cancel"),
                ],
            ],
            resize_keyboard=True,
        ),
    )        

Lets also add handler for "confirm_reject" state:

@form_router.message(Form.confirm_reject, F.text.casefold() != "cancel")
async def process_reject(message: Message, state: FSMContext) -> None:
    required_state_index = next(
        (
            i
            for i, obj in enumerate(STATES_LIST)
            if obj.state_corresponding_button == message.text
        ),
        None,
    )
    if required_state_index is None:
        await message.answer(
            "I don't understand you. Please, choose one of the options.",
        )
    else:
        required_state = STATES_LIST[required_state_index]
        await state.update_data(reject=True)
        await state.set_state(required_state.state_name)
        await message.answer(
            required_state.state_question,
            reply_markup=ReplyKeyboardMarkup(
                keyboard=required_state.keyboard_buttons,
                resize_keyboard=True,
            ),
        )        

In this handler firstly we validate if user selected one of available buttons, for this we are trying to find state by state_corresponding_button value in STATES_LIST.

After this we need to receive required_state information, make transition to this step and send proper message. But logically - after user will answer to this question we should send him to confirmation step again, for this we need to add "reject" value to states data and also update our states handlers to handle this value. Updated "first_name" state handler

@form_router.message(Form.about.first_name)
async def process_first_name(message: Message, state: FSMContext) -> None:
    await state.update_data(first_name=message.text)
    data = await state.get_data()
    reject = data.get("reject", False)
    if reject:
        await update_on_reject(message, state)
    else:
        await state.set_state(Form.about.last_name)
        await message.answer(
            "What's your last name?",
            reply_markup=ReplyKeyboardMarkup(
                keyboard=[
                    [
                        KeyboardButton(text="Skip"),
                        KeyboardButton(text="Go back")
                    ],
                    [KeyboardButton(text="Cancel")],
                ],
                resize_keyboard=True,
            ),
        )        

As you can see we added only if else statement to handler reject, in case if reject is equal to True we are calling update_on_reject method:

async def update_on_reject(message: Message, state: FSMContext) -> str:
    await state.set_state(Form.confirm)
    data: Dict[str, Any] = await state.get_data()
    await message.answer(
        f"Please confirm your data:\n"
        f"Name: {html.quote(data['first_name'])} "
        f"{html.quote(data['last_name'])}\n"
        f"Age: {html.quote(data['age'])}\n"
        f"Hobbies: {html.quote(data['hobbies'])}\n"
        f"City: {html.quote(data['city'])}\n"
        f"Street: {html.quote(data['street'])}\n"
        f"House: {html.quote(data['house'])}\n"
        f"Phone: {html.quote(data['phone'])}\n"
        f"Email: {html.quote(data['email'])}",
        reply_markup=ReplyKeyboardMarkup(
            keyboard=[
                [
                    KeyboardButton(text="Approve"),
                    KeyboardButton(text="Disapprove")
                ],
                [KeyboardButton(text="Go back")],
                [KeyboardButton(text="Cancel")],
            ],
            resize_keyboard=True,
        ),
    )        

This method will make transition to confirmation state and send confirmation message again.

On our state schema this operation is illustrated in two lines, firstly we perform transition from "Confirmation" to "First name" (with red arrow) and then back to "Confirmation" (with purple arrow). In Telegram flow it will looks like this:

This article was created by the SP-Lutsk LLC team. We are dedicated to producing quality content and hope you found this article informative and engaging. Stay tuned for more updates!

?SP-Lutsk — 2024