Filter rules

In order for the handler to catch only the necessary messages / other events, filters are needed, there are filters in the framework right out of the box.

To access rules out of the box, you can do different things:

1. Import them from waio.rules and use them, initializing them directly in the decorator or in any other part of the code:

from waio.rules import MessageCommandsRule


@dp.message_handler(MessageCommandsRule(commands=["start", "echo"]))
async def commands_rule_without_labeler(event: Event):
    await event.answer(f"Filter used: [MessageCommandsRule], msg: {event.text}")

1. Use the pre-built automatic rule unpackers that are specified as an argument to the decorator. but in this case, some secondary arguments for the rule cannot be passed, if the rule contains more than one argument, for example: TextRule, MessageCommandsRule

   @dp.message_handler(text_equals=["foo", "bar"])
   async def start_text_equals(event: Event):
       await event.answer(f"Filter used: [text_equals], msg: {event.text}")
   
   
   @dp.message_handler(text_contains=["ru", "com"])
   async def start_text_contains(event: Event):
       await event.answer(f"Filter used: [text_contains], msg: {event.text}")
   

Combine filters!

You can set any number of named rules as well as unnamed ones.

from waio.rules import TextRule, ContentType
from waio.types import Event


@dp.message_handler(
    TextRule(startswith=["1111", "2222"], endswith=["x", "y", "z"]),
    content_type=[ContentType.TEXT],
)
async def text_start_switch_without_labeler(event: Event):
    await event.answer(f"Filter used: [TextRule], msg: {event.text}")

Default named rules:

{
    "commands": MessageCommandsRule,
    "state": StateRule,
    "regex": RegexRule,
    "text_equals": TextRuleEquals,
    "text_contains": TextRuleContains,
    "text_startswith": TextRuleStartswith,
    "text_endswith": TextRuleEndswith,
    "content_type": ContentTypeRule,
}

Creating your own filter rules

A rule is a class corresponding to the ABCRule interface, which must implement only one asynchronous check method that accepts an event and returns False if the check was not passed and True or a dictionary with arguments that will be unpacked into the handler as non-positional arguments.

Unnamed filter rule

To create rules, we import the abstract ABCRule interface and implement the asynchronous check method, it is also worth importing Union from typing to type your code.

from waio.rules import ABCRule
from waio.types import Event


class StaticLongMessageRule(ABCRule):
    async def check(self, event: Event) -> bool:
        return len(event.text) > 200

Our rule does not accept any arguments, it only checks if the message length is more than two hundred characters, it will return True, the filter will be triggered and the handler will be called, otherwise False will be returned and the filter will not be processed.

Now we can use our rule.

@dp.message_handler(StaticLongMessageRule())
async def foo(event: Event):
    ...

We have got a static rule that does not accept any arguments, the logic has a hard binding to a length of 200 characters, let's modernize our rule by adding the ability to specify the length ourselves in __init__.

from waio.rules import ABCRule
from waio.types import Event


class DynamicLongMessageRule(ABCRule):
    def __init__(self, len_message: int):
        self.len_message = len_message

    async def check(self, event: Event) -> bool:
        return len(event.text) > self.len_message

Then we use it in the handler

@dp.message_handler(DynamicLongMessageRule(len_message=120))
async def foo(event: Event):
    ...

Now we can set an identifier for our rule (the name of the rule) and then pass to the decorator not an object of the rule, but a named argument of the name of the rule.

Named filter rule

Please note that you need to declare the rules Now we can set an identifier for our rule (the name of the rule) and then pass to the decorator not an object of the rule, but a named argument of the name of the rule.before the handlers!

dp.labeler.bind_rule("len_more", DynamicLongMessageRule)

Usage:

@dp.message_handler(len_more=12)
async def text_len(event: Event):
    await event.answer(f"msg len: {len(event.text)}")

Full code:

from aiohttp import web

from waio import Bot, Dispatcher
from waio.rules.abc import ABCRule
from waio.types import Event
from waio.logs import loguru_filter

loguru_filter.set_level("DEBUG")

bot = Bot(apikey="API_KEY", src_name="SRC_NAME", phone_number=79281112233)

dp = Dispatcher(bot=bot)


class DynamicLongMessageRule(ABCRule):
    def __init__(self, len_message: int):
        self.len_message = len_message

    async def check(self, event: Event) -> bool:
        return len(event.text) > self.len_message


dp.labeler.bind_rule("len_more", DynamicLongMessageRule)


@dp.message_handler(len_more=20)
async def text_len(event: Event):
    await event.answer(f"msg len: {len(event.text)}")


async def handler_gupshup(request):
    event = await request.json()
    await dp.handle_event(event)
    return web.Response(status=200)


if __name__ == "__main__":
    webhook = web.Application()
    webhook.add_routes([web.post("/api/v1/gupshup/hook", handler_gupshup)])
    web.run_app(webhook, port=8017)

Unpacking data into a handler from a filter

Return custom data from filter? Of course!

The check method can return not only bool type, but also a dict that will be unpacked into a handler, in which you can access everything that the filter returned.

from typing import Dict, Union, Tuple
from phonenumbers import timezone, parse, geocoder

from waio.rules import ABCRule
from waio.types import Event

G_T = Dict[str, Union[int, str, Tuple[str]]]


class RussianNumberRule(ABCRule):
    async def check(self, event: Event) -> Union[bool, Dict[str, G_T]]:
        phone_number_data = self.get_phone_number_data(event.sender_number)
        if phone_number_data["country"] == "Russia":
            return {"number_data": phone_number_data}
        return False

    @staticmethod
    def get_phone_number_data(number: str) -> G_T:
        number_and_plus = f"+{number}"
        phone_number = parse(number_and_plus)
        country_name = geocoder.country_name_for_number(phone_number, "en")
        time_zones_number = timezone.time_zones_for_number(phone_number)

        return {
            "country_code": phone_number.country_code,
            "national_number": phone_number.national_number,
            "country": country_name,
            "time_zone": time_zones_number,
        }

Let's write a handler:

@dp.message_handler(RussianNumberRule(), commands=["check_number"])
async def register_rule_check_number(event: Event, number_data: G_T):
    await event.answer(f"You are from Russia! Number data:\n" f"```{number_data}```")

Let's analyze what is written above: In the decorator, we will set our filterRussianNumberRule(), if the user who sent the message to the bot has a Russian telecom operator - the filter will be processed. We will also add the number_data argument to the handler, which will contain our dictionary with information about this phone number, which was created in the filter.