Skip to content

BaseBotCommand

Represent a base Telegram bot command.

This is the base class for all user-defined Telegram bot commands.

Source code in django_telegram_app/bot/base.py 
 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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class BaseBotCommand:
    """Represent a base Telegram bot command.

    This is the base class for all user-defined Telegram bot commands.
    """

    description: str = ""
    exclude_from_help: bool = False
    translate: bool = True

    def __init__(self, settings: AbstractTelegramSettings):
        """Initialize the command."""
        self.settings = settings

    def start(self, telegram_update: TelegramUpdate):
        """Start the command."""
        logging.info(f"Starting {self.get_name()} for {self.settings}")
        self._clear_state()
        return self.steps[0](telegram_update)

    def finish(self, current_step_name: str, telegram_update: TelegramUpdate):
        """Finish the command and clear all data."""
        logging.info(f"Finishing the command at step {current_step_name}")
        self._clear_state()
        self._clear_callback_data(telegram_update)

    def cancel(self, current_step_name: str, telegram_update: TelegramUpdate):
        """Cancel the command and clear all data."""
        from django_telegram_app.bot.bot import send_message

        logging.info(f"Canceled the command at step {current_step_name}")
        data = self.get_callback_data(telegram_update.callback_data)
        cancel_text = data.get("cancel_text", _("Command canceled."))
        send_message(cancel_text, self.settings.chat_id)
        return self.finish(current_step_name, telegram_update)

    def next_step(self, current_step_name: str, telegram_update: TelegramUpdate):
        """Proceed to the next step in the command."""
        next_index = self._steps_to_str().index(current_step_name) + 1
        if next_index < len(self.steps):
            next_step = self.steps[next_index]
            return next_step(telegram_update)
        self.finish(current_step_name, telegram_update)

    def previous_step(self, current_step_name: str, telegram_update: TelegramUpdate):
        """Return to the previous step in the command."""
        data = self.get_callback_data(telegram_update.callback_data)
        steps_back = int(data.get("_steps_back", 1))
        previous_index = self._steps_to_str().index(current_step_name) - steps_back
        if previous_index >= 0:
            previous_step = self.steps[previous_index]
            return previous_step(telegram_update)

    def current_step(self, current_step_name: str, telegram_update: TelegramUpdate):
        """Reload the current step."""
        current_index = self._steps_to_str().index(current_step_name)
        current_step = self.steps[current_index]
        return current_step(telegram_update)

    def create_callback(self, step_name: str, action: str, **kwargs):
        """Create callback data for the current command and return the token."""
        if not kwargs:
            kwargs = self._get_default_callback_data()
        if "correlation_key" not in kwargs:
            kwargs.update(self._get_default_callback_data())
        callback_data = CallbackData(command=self.get_command_string(), step=step_name, action=action, data=kwargs)
        callback_data.save()
        return str(callback_data.token)

    def get_callback(self, token: str):
        """Return the callback for the given token."""
        return CallbackData.objects.get(token=token)

    def get_callback_data(self, callback_token: str) -> dict[str, Any]:
        """Get callback data from the callback token.

        If the callback token is not provided, return default callback data.
        """
        if not callback_token:
            return self._get_default_callback_data()
        callback_data = self.get_callback(callback_token)
        return callback_data.data

    @property
    def steps(self) -> Sequence[Step]:
        """Return the steps of the command."""
        raise NotImplementedError("Subclasses must implement this method")

    @classmethod
    def get_name(cls):
        """Return the name of the command.

        By default this is the lowercased last part of the module name.
        """
        return cls.__module__.split(".")[-1].lower()

    @classmethod
    def get_command_string(cls):
        """Return the command string."""
        return f"/{cls.get_name()}"

    def _get_default_callback_data(self):
        """Return a dictionary with correlation key as default callback data."""
        return {"correlation_key": str(uuid.uuid4())}

    def _clear_state(self):
        """Clear the command state."""
        self.settings.data = {}
        self.settings.save()

    def _clear_callback_data(self, telegram_update: TelegramUpdate):
        """Clear callback data for the current command."""
        step_data = self.get_callback_data(telegram_update.callback_data)
        correlation_key = step_data.get("correlation_key", "non_existent_key")
        CallbackData.objects.filter(data__correlation_key=correlation_key).delete()

    def _steps_to_str(self):
        return [step.name for step in self.steps]

steps property

Return the steps of the command.

__init__(settings)

Initialize the command.

Source code in django_telegram_app/bot/base.py 
33
34
35
def __init__(self, settings: AbstractTelegramSettings):
    """Initialize the command."""
    self.settings = settings

cancel(current_step_name, telegram_update)

Cancel the command and clear all data.

Source code in django_telegram_app/bot/base.py 
49
50
51
52
53
54
55
56
57
def cancel(self, current_step_name: str, telegram_update: TelegramUpdate):
    """Cancel the command and clear all data."""
    from django_telegram_app.bot.bot import send_message

    logging.info(f"Canceled the command at step {current_step_name}")
    data = self.get_callback_data(telegram_update.callback_data)
    cancel_text = data.get("cancel_text", _("Command canceled."))
    send_message(cancel_text, self.settings.chat_id)
    return self.finish(current_step_name, telegram_update)

create_callback(step_name, action, **kwargs)

Create callback data for the current command and return the token.

Source code in django_telegram_app/bot/base.py 
82
83
84
85
86
87
88
89
90
def create_callback(self, step_name: str, action: str, **kwargs):
    """Create callback data for the current command and return the token."""
    if not kwargs:
        kwargs = self._get_default_callback_data()
    if "correlation_key" not in kwargs:
        kwargs.update(self._get_default_callback_data())
    callback_data = CallbackData(command=self.get_command_string(), step=step_name, action=action, data=kwargs)
    callback_data.save()
    return str(callback_data.token)

current_step(current_step_name, telegram_update)

Reload the current step.

Source code in django_telegram_app/bot/base.py 
76
77
78
79
80
def current_step(self, current_step_name: str, telegram_update: TelegramUpdate):
    """Reload the current step."""
    current_index = self._steps_to_str().index(current_step_name)
    current_step = self.steps[current_index]
    return current_step(telegram_update)

finish(current_step_name, telegram_update)

Finish the command and clear all data.

Source code in django_telegram_app/bot/base.py 
43
44
45
46
47
def finish(self, current_step_name: str, telegram_update: TelegramUpdate):
    """Finish the command and clear all data."""
    logging.info(f"Finishing the command at step {current_step_name}")
    self._clear_state()
    self._clear_callback_data(telegram_update)

get_callback(token)

Return the callback for the given token.

Source code in django_telegram_app/bot/base.py 
92
93
94
def get_callback(self, token: str):
    """Return the callback for the given token."""
    return CallbackData.objects.get(token=token)

get_callback_data(callback_token)

Get callback data from the callback token.

If the callback token is not provided, return default callback data.

Source code in django_telegram_app/bot/base.py 
 96
 97
 98
 99
100
101
102
103
104
def get_callback_data(self, callback_token: str) -> dict[str, Any]:
    """Get callback data from the callback token.

    If the callback token is not provided, return default callback data.
    """
    if not callback_token:
        return self._get_default_callback_data()
    callback_data = self.get_callback(callback_token)
    return callback_data.data

get_command_string() classmethod

Return the command string.

Source code in django_telegram_app/bot/base.py 
119
120
121
122
@classmethod
def get_command_string(cls):
    """Return the command string."""
    return f"/{cls.get_name()}"

get_name() classmethod

Return the name of the command.

By default this is the lowercased last part of the module name.

Source code in django_telegram_app/bot/base.py 
111
112
113
114
115
116
117
@classmethod
def get_name(cls):
    """Return the name of the command.

    By default this is the lowercased last part of the module name.
    """
    return cls.__module__.split(".")[-1].lower()

next_step(current_step_name, telegram_update)

Proceed to the next step in the command.

Source code in django_telegram_app/bot/base.py 
59
60
61
62
63
64
65
def next_step(self, current_step_name: str, telegram_update: TelegramUpdate):
    """Proceed to the next step in the command."""
    next_index = self._steps_to_str().index(current_step_name) + 1
    if next_index < len(self.steps):
        next_step = self.steps[next_index]
        return next_step(telegram_update)
    self.finish(current_step_name, telegram_update)

previous_step(current_step_name, telegram_update)

Return to the previous step in the command.

Source code in django_telegram_app/bot/base.py 
67
68
69
70
71
72
73
74
def previous_step(self, current_step_name: str, telegram_update: TelegramUpdate):
    """Return to the previous step in the command."""
    data = self.get_callback_data(telegram_update.callback_data)
    steps_back = int(data.get("_steps_back", 1))
    previous_index = self._steps_to_str().index(current_step_name) - steps_back
    if previous_index >= 0:
        previous_step = self.steps[previous_index]
        return previous_step(telegram_update)

start(telegram_update)

Start the command.

Source code in django_telegram_app/bot/base.py 
37
38
39
40
41
def start(self, telegram_update: TelegramUpdate):
    """Start the command."""
    logging.info(f"Starting {self.get_name()} for {self.settings}")
    self._clear_state()
    return self.steps[0](telegram_update)