Merge pull request #1369 from abdullaev388/master

CallbackData class added

examples/CallbackData_example.py

@@ -0,0 +1,86 @@
+# -*- coding: utf-8 -*-
+"""
+This Example will show you how to use CallbackData
+"""
+
+from telebot.callback_data import CallbackData, CallbackDataFilter
+from telebot import types, TeleBot
+from telebot.custom_filters import AdvancedCustomFilter
+
+API_TOKEN = ''
+PRODUCTS = [
+    {'id': '0', 'name': 'xiaomi mi 10', 'price': 400},
+    {'id': '1', 'name': 'samsung s20', 'price': 800},
+    {'id': '2', 'name': 'iphone 13', 'price': 1300}
+]
+
+bot = TeleBot(API_TOKEN)
+products_factory = CallbackData('product_id', prefix='products')
+
+
+def products_keyboard():
+    return types.InlineKeyboardMarkup(
+        keyboard=[
+            [
+                types.InlineKeyboardButton(
+                    text=product['name'],
+                    callback_data=products_factory.new(product_id=product["id"])
+                )
+            ]
+            for product in PRODUCTS
+        ]
+    )
+
+
+def back_keyboard():
+    return types.InlineKeyboardMarkup(
+        keyboard=[
+            [
+                types.InlineKeyboardButton(
+                    text='⬅',
+                    callback_data='back'
+                )
+            ]
+        ]
+    )
+
+
+class ProductsCallbackFilter(AdvancedCustomFilter):
+    key = 'config'
+
+    def check(self, call: types.CallbackQuery, config: CallbackDataFilter):
+        return config.check(query=call)
+
+
+@bot.message_handler(commands=['products'])
+def products_command_handler(message: types.Message):
+    bot.send_message(message.chat.id, 'Products:', reply_markup=products_keyboard())
+
+
+# Only product with field - product_id = 2
+@bot.callback_query_handler(func=None, config=products_factory.filter(product_id='2'))
+def product_one_callback(call: types.CallbackQuery):
+    bot.answer_callback_query(callback_query_id=call.id, text='Not available :(', show_alert=True)
+
+
+# Any other products
+@bot.callback_query_handler(func=None, config=products_factory.filter())
+def products_callback(call: types.CallbackQuery):
+    callback_data: dict = products_factory.parse(callback_data=call.data)
+    product_id = int(callback_data['product_id'])
+    product = PRODUCTS[product_id]
+
+    text = f"Product name: {product['name']}\n" \
+           f"Product price: {product['price']}"
+    bot.edit_message_text(chat_id=call.message.chat.id, message_id=call.message.message_id,
+                          text=text, reply_markup=back_keyboard())
+
+
+@bot.callback_query_handler(func=lambda c: c.data == 'back')
+def back_callback(call: types.CallbackQuery):
+    bot.edit_message_text(chat_id=call.message.chat.id, message_id=call.message.message_id,
+                          text='Products:', reply_markup=products_keyboard())
+
+
+bot.add_custom_filter(ProductsCallbackFilter())
+bot.infinity_polling()

telebot/callback_data.py

@@ -0,0 +1,115 @@
+import typing
+
+
+class CallbackDataFilter:
+
+    def __init__(self, factory, config: typing.Dict[str, str]):
+        self.config = config
+        self.factory = factory
+
+    def check(self, query):
+        """
+        Checks if query.data appropriates to specified config
+        :param query: telebot.types.CallbackQuery
+        :return: bool
+        """
+
+        try:
+            data = self.factory.parse(query.data)
+        except ValueError:
+            return False
+
+        for key, value in self.config.items():
+            if isinstance(value, (list, tuple, set, frozenset)):
+                if data.get(key) not in value:
+                    return False
+            elif data.get(key) != value:
+                return False
+        return True
+
+
+class CallbackData:
+    """
+    Callback data factory
+    This class will help you to work with CallbackQuery
+    """
+
+    def __init__(self, *parts, prefix: str, sep=':'):
+        if not isinstance(prefix, str):
+            raise TypeError(f'Prefix must be instance of str not {type(prefix).__name__}')
+        if not prefix:
+            raise ValueError("Prefix can't be empty")
+        if sep in prefix:
+            raise ValueError(f"Separator {sep!r} can't be used in prefix")
+
+        self.prefix = prefix
+        self.sep = sep
+
+        self._part_names = parts
+
+    def new(self, *args, **kwargs) -> str:
+        """
+        Generate callback data
+        :param args: positional parameters of CallbackData instance parts
+        :param kwargs: named parameters
+        :return: str
+        """
+        args = list(args)
+
+        data = [self.prefix]
+
+        for part in self._part_names:
+            value = kwargs.pop(part, None)
+            if value is None:
+                if args:
+                    value = args.pop(0)
+                else:
+                    raise ValueError(f'Value for {part!r} was not passed!')
+
+            if value is not None and not isinstance(value, str):
+                value = str(value)
+
+            if self.sep in value:
+                raise ValueError(f"Symbol {self.sep!r} is defined as the separator and can't be used in parts' values")
+
+            data.append(value)
+
+        if args or kwargs:
+            raise TypeError('Too many arguments were passed!')
+
+        callback_data = self.sep.join(data)
+
+        if len(callback_data.encode()) > 64:
+            raise ValueError('Resulted callback data is too long!')
+
+        return callback_data
+
+    def parse(self, callback_data: str) -> typing.Dict[str, str]:
+        """
+        Parse data from the callback data
+        :param callback_data: string, use to telebot.types.CallbackQuery to parse it from string to a dict
+        :return: dict parsed from callback data
+        """
+
+        prefix, *parts = callback_data.split(self.sep)
+        if prefix != self.prefix:
+            raise ValueError("Passed callback data can't be parsed with that prefix.")
+        elif len(parts) != len(self._part_names):
+            raise ValueError('Invalid parts count!')
+
+        result = {'@': prefix}
+        result.update(zip(self._part_names, parts))
+        return result
+
+    def filter(self, **config) -> CallbackDataFilter:
+        """
+        Generate filter
+
+        :param config: specified named parameters will be checked with CallbackQury.data
+        :return: CallbackDataFilter class
+        """
+
+        for key in config.keys():
+            if key not in self._part_names:
+                raise ValueError(f'Invalid field name {key!r}')
+        return CallbackDataFilter(self, config)