pythonfilterchatbothandleraiogram

How to use BoundFilter in aiogram 3.x


I'm trying to use a custom type of filter like BoundFilter in aiogram 2.x for filtering not allowed characters.

But recommended

class AllowedCharactersFilter(BoundFilter):
    async def check(self, message: types.Message):
        allowed_characters = set('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890')
        entered_text = message.text
        return all(char in allowed_characters for char in entered_text)

doesn't work. Because in aiogram 3.4 neither

from aiogram.dispather.filters import BoundFilter

nor

from aiogram.filters import BoundFilter

work.

Is BoundFilter to be imported in some another fashion?

I found that probably the solution is possible through using a BaseFilter imported from aiogram.filters. With creation of new directory under filters.

But a description of the procedure is not clear. Can comebody shed a light on the subject: How to filter unallowed characters in handlers of aiogram 3.4?


Solution

  • it seems 3.x has Magic Filters and you may need only F.text.isalnum()

    from aiogram import F
    
    @dp.message(F.text.isalnum())
    async def my_handler(message: Message):
        #... code ...
    

    EDIT:

    I tested it and isalnum() allows only for messages with alphanumeric chars - but it means also native chars like ąść. Your function doesn't allow for native chars like ąść


    And if you want to build filter then documentation shows aiogram.filters.base.Filter as base class for own filter.

    #import string
    from aiogram.filters import Filter
    
    class MyFilter(Filter):
    
        def __init__(self) -> None:
            #self.allowed_characters = set(string.ascii_letters + string.digits)
            self.allowed_characters = set('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890')
    
        async def __call__(self, message: Message) -> bool:
            #not_allowed = set(message.text) - self.allowed_characters
            #return len(not_allowed) == 0
            return all(char in self.allowed_characters for char in message.text)
    
    
    @dp.message(MyFilter())
    async def my_handler(message: Message):
        #... code ...
    

    EDIT:

    Minimal working code which I used for tests:

    import asyncio
    import logging
    import sys
    from os import getenv
    
    import aiogram
    print('aiogram.__version__:', aiogram.__version__)
    
    from aiogram import Bot, Dispatcher, html
    from aiogram.client.default import DefaultBotProperties
    from aiogram.enums import ParseMode
    from aiogram.filters import CommandStart, Filter
    from aiogram.types import Message
    from aiogram import F
    
    TOKEN = getenv("TELEGRAM_TOKEN")
    
    dp = Dispatcher()
    
    class MyFilter(Filter):
    
        def __init__(self) -> None:
            #self.allowed_characters = set(string.ascii_letters + string.digits)
            self.allowed_characters = set('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890')
    
        async def __call__(self, message: Message) -> bool:
            #not_allowed = set(message.text) - self.allowed_characters
            #return len(not_allowed) == 0
            return all(char in self.allowed_characters for char in message.text)
    
    
    @dp.message(CommandStart())
    async def command_start_handler(message: Message) -> None:
        await message.answer(f"Hello, {html.bold(message.from_user.full_name)}!")
    
    
    #@dp.message(F.text.isalnum())   # use one of them
    @dp.message(MyFilter())   # use one of them
    async def echo_handler(message: Message) -> None:
        try:
            print("echo_handler:", message.text)
            await message.send_copy(chat_id=message.chat.id)
        except TypeError:
            await message.answer("Nice try!")
            
    
    async def main() -> None:
        bot = Bot(token=TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
        await dp.start_polling(bot)
    
    
    if __name__ == "__main__":
        logging.basicConfig(level=logging.INFO, stream=sys.stdout)
        asyncio.run(main())