Step

Represent a step in a Telegram bot command.

This is the base class for all user-defined steps.

Source code in django_telegram_app/bot/base.py

class Step:
    """Represent a step in a Telegram bot command.

    This is the base class for all user-defined steps.
    """

    def __init__(self, command: BaseBotCommand, unique_id: str | None = None, translate: bool | None = None):
        """Initialize the step.

        Args:
            command: The command this step belongs to.
            unique_id: An optional unique identifier for the step. If not provided, the class name will be used.
            translate: Whether to activate translation for this step. If None, the command's translate setting will be
                       used. If True or False, it will override the command's setting.
        """
        self.command = command
        self.unique_id = unique_id
        self.translate = translate

    def __call__(self, telegram_update: TelegramUpdate):
        """Execute the step.

        This activates the appropriate translation based on the user's language code.
        """
        if self.translate is None:
            should_translate = self.command.translate
        else:
            should_translate = self.translate

        translation_override = override(telegram_update.language_code) if should_translate else nullcontext()
        with translation_override:
            return self.handle(telegram_update)

    def handle(self, telegram_update: TelegramUpdate):
        """Handle the step."""
        raise NotImplementedError("This method should be overridden by subclasses.")

    def callback_to(self, step: StepRef, original_data: dict | None = None, **kwargs: Any) -> str:
        """Create a callback that routes to a specific step."""
        original_data = original_data or {}
        data = {**original_data, **kwargs}
        return self.command._create_callback("go_to", step, **data)

    def next_step_callback(self, original_data: dict | None = None, **kwargs):
        """Create a callback to advance to the next step."""
        return self._create_callback("next_step", original_data, **kwargs)

    def previous_step_callback(self, steps_back: int, original_data: dict | None = None, **kwargs):
        """Create a callback to return to the previous step."""
        kwargs["_steps_back"] = steps_back
        return self._create_callback("previous_step", original_data, **kwargs)

    def current_step_callback(self, original_data: dict | None = None, **kwargs):
        """Create a callback to reload the current step with the provided data."""
        return self.callback_to(self.name, original_data, **kwargs)

    def cancel_callback(self, original_data: dict | None = None, **kwargs):
        """Create a callback to cancel the command."""
        return self._create_callback("cancel", original_data, **kwargs)

    def get_callback_data(self, telegram_update: TelegramUpdate):
        """Get callback data from the telegram_update.

        If the update is a message and not a command, check if we are waiting for user input.
        If so, retrieve the callback data using the waiting_for token and store the message text
        in the appropriate key in the callback data.

        Otherwise, retrieve the callback data using the callback token from the update.
        If no callback token is provided, return default callback data.
        """
        if not telegram_update.callback_data and telegram_update.is_message() and not telegram_update.is_command():
            waiting_for = self.command.settings.data.get("_waiting_for", None)
            if waiting_for:
                callback_token = waiting_for
                callback_data = self.command.get_callback_data(callback_token)
                key = callback_data["_message_key"]  # Move the message_text to this key
                callback_data[key] = telegram_update.message_text.strip()
                return callback_data

        callback_token = telegram_update.callback_data
        return self.command.get_callback_data(callback_token)

    def add_waiting_for(self, message_key: str, data: dict[str, Any] | None = None):
        """Add waiting_for to the command settings.

        The message_key will be used to store the user input in the callback data of the next step.
        """
        data = data or {}
        self.command.settings.data["_waiting_for"] = self.next_step_callback(data, _message_key=message_key)
        self.command.settings.save()

    @property
    def name(self):
        """Return the name of the step.

        This is either the unique_id provided during initialization or the class name of the step.
        """
        return self.unique_id or type(self).__name__

    def _create_callback(self, action: str, original_data: dict | None = None, **kwargs):
        """Create callback data for the current step and return the token."""
        original_data = original_data or {}
        data = {**original_data, **kwargs}
        return self.command._create_callback(action, self.name, **data)

    def __str__(self):
        """Return the name of the step."""
        return self.name

name property

Return the name of the step.

This is either the unique_id provided during initialization or the class name of the step.

__call__(telegram_update)

Execute the step.

This activates the appropriate translation based on the user's language code.

Source code in django_telegram_app/bot/base.py

def __call__(self, telegram_update: TelegramUpdate):
    """Execute the step.

    This activates the appropriate translation based on the user's language code.
    """
    if self.translate is None:
        should_translate = self.command.translate
    else:
        should_translate = self.translate

    translation_override = override(telegram_update.language_code) if should_translate else nullcontext()
    with translation_override:
        return self.handle(telegram_update)

__init__(command, unique_id=None, translate=None)

Initialize the step.

Parameters:

Name	Type	Description	Default
command	BaseBotCommand	The command this step belongs to.	required
unique_id	str | None	An optional unique identifier for the step. If not provided, the class name will be used.	None
translate	bool | None	Whether to activate translation for this step. If None, the command's translate setting will be used. If True or False, it will override the command's setting.	None

Source code in django_telegram_app/bot/base.py

def __init__(self, command: BaseBotCommand, unique_id: str | None = None, translate: bool | None = None):
    """Initialize the step.

    Args:
        command: The command this step belongs to.
        unique_id: An optional unique identifier for the step. If not provided, the class name will be used.
        translate: Whether to activate translation for this step. If None, the command's translate setting will be
                   used. If True or False, it will override the command's setting.
    """
    self.command = command
    self.unique_id = unique_id
    self.translate = translate

__str__()

Return the name of the step.

Source code in django_telegram_app/bot/base.py

def __str__(self):
    """Return the name of the step."""
    return self.name

add_waiting_for(message_key, data=None)

Add waiting_for to the command settings.

The message_key will be used to store the user input in the callback data of the next step.

Source code in django_telegram_app/bot/base.py

def add_waiting_for(self, message_key: str, data: dict[str, Any] | None = None):
    """Add waiting_for to the command settings.

    The message_key will be used to store the user input in the callback data of the next step.
    """
    data = data or {}
    self.command.settings.data["_waiting_for"] = self.next_step_callback(data, _message_key=message_key)
    self.command.settings.save()

callback_to(step, original_data=None, **kwargs)

Create a callback that routes to a specific step.

Source code in django_telegram_app/bot/base.py

def callback_to(self, step: StepRef, original_data: dict | None = None, **kwargs: Any) -> str:
    """Create a callback that routes to a specific step."""
    original_data = original_data or {}
    data = {**original_data, **kwargs}
    return self.command._create_callback("go_to", step, **data)

cancel_callback(original_data=None, **kwargs)

Create a callback to cancel the command.

Source code in django_telegram_app/bot/base.py

def cancel_callback(self, original_data: dict | None = None, **kwargs):
    """Create a callback to cancel the command."""
    return self._create_callback("cancel", original_data, **kwargs)

current_step_callback(original_data=None, **kwargs)

Create a callback to reload the current step with the provided data.

Source code in django_telegram_app/bot/base.py

def current_step_callback(self, original_data: dict | None = None, **kwargs):
    """Create a callback to reload the current step with the provided data."""
    return self.callback_to(self.name, original_data, **kwargs)

get_callback_data(telegram_update)

Get callback data from the telegram_update.

If the update is a message and not a command, check if we are waiting for user input. If so, retrieve the callback data using the waiting_for token and store the message text in the appropriate key in the callback data.

Otherwise, retrieve the callback data using the callback token from the update. If no callback token is provided, return default callback data.

Source code in django_telegram_app/bot/base.py

def get_callback_data(self, telegram_update: TelegramUpdate):
    """Get callback data from the telegram_update.

    If the update is a message and not a command, check if we are waiting for user input.
    If so, retrieve the callback data using the waiting_for token and store the message text
    in the appropriate key in the callback data.

    Otherwise, retrieve the callback data using the callback token from the update.
    If no callback token is provided, return default callback data.
    """
    if not telegram_update.callback_data and telegram_update.is_message() and not telegram_update.is_command():
        waiting_for = self.command.settings.data.get("_waiting_for", None)
        if waiting_for:
            callback_token = waiting_for
            callback_data = self.command.get_callback_data(callback_token)
            key = callback_data["_message_key"]  # Move the message_text to this key
            callback_data[key] = telegram_update.message_text.strip()
            return callback_data

    callback_token = telegram_update.callback_data
    return self.command.get_callback_data(callback_token)

handle(telegram_update)

Handle the step.

Source code in django_telegram_app/bot/base.py

def handle(self, telegram_update: TelegramUpdate):
    """Handle the step."""
    raise NotImplementedError("This method should be overridden by subclasses.")

next_step_callback(original_data=None, **kwargs)

Create a callback to advance to the next step.

Source code in django_telegram_app/bot/base.py

def next_step_callback(self, original_data: dict | None = None, **kwargs):
    """Create a callback to advance to the next step."""
    return self._create_callback("next_step", original_data, **kwargs)

previous_step_callback(steps_back, original_data=None, **kwargs)

Create a callback to return to the previous step.

Source code in django_telegram_app/bot/base.py

def previous_step_callback(self, steps_back: int, original_data: dict | None = None, **kwargs):
    """Create a callback to return to the previous step."""
    kwargs["_steps_back"] = steps_back
    return self._create_callback("previous_step", original_data, **kwargs)