Фильтрация событий

Фильтры нужны для маршрутизации обновлений к конкретному обработчику. Поиск обработчика всегда останавливается на первом совпадении набора фильтров.

litegram имеет несколько встроенных полезных фильтров.

Встроенные фильтры

Вот список встроенных фильтров:

Написание собственных фильтров

Фильтры могут быть:

  • Асинхронной функцией (async def my_filter(*args, **kwargs): pass)

  • Синхронной функцией (def my_filter(*args, **kwargs): pass)

  • Анонимной функцией (lambda event: True)

  • Любым ожидаемым объектом (awaitable object)

  • Подклассом litegram.filters.base.Filter

  • Экземпляром MagicFilter

и должен возвращать bool или dict. Если в качестве результата фильтра передан словарь — полученные данные будут переданы следующим фильтрам и обработчику в качестве именованных аргументов (keywords arguments).

Базовый класс для собственных фильтров

class litegram.filters.base.Filter[исходный код]

Если вы хотите зарегистрировать собственные фильтры, как встроенные фильтры, вам нужно будет написать подкласс этого класса с переопределением метода __call__ и добавлением атрибутов фильтра.

abstractmethod async __call__(*args: Any, **kwargs: Any) bool | dict[str, Any][исходный код]

Этот метод следует переопределить.

Принимает входящее событие и должен вернуть логическое (bool) значение или dict.

Результат:

bool or Dict[str, Any]

update_handler_flags(flags: dict[str, Any]) None[исходный код]

Кроме того, если вы хотите расширить флаги обработчика (handler) с помощью этого фильтра, вам следует реализовать этот метод

Параметры:

flags – существующие флаги, можно обновить напрямую

Пример собственного фильтра

Например, если вам нужно создать простой текстовый фильтр:

from __future__ import annotations

from typing import TYPE_CHECKING

from litegram import Router
from litegram.filters import Filter

if TYPE_CHECKING:
    from litegram.types import Message

router = Router()


class MyFilter(Filter):
    def __init__(self, my_text: str) -> None:
        self.my_text = my_text

    async def __call__(self, message: Message) -> bool:
        return message.text == self.my_text


@router.message(MyFilter("hello"))
async def my_handler(message: Message) -> None: ...

Комбинирование фильтров

В целом, все фильтры можно комбинировать двумя способами

Другой возможный способ

Альтернативным способом является объединение с помощью специальных функций (and_f(), or_f(), invert_f() из модуля litegram.filters):

and_f(F.text.startswith("show"), F.text.endswith("example"))
or_f(F.text(text="hi"), CommandStart())
invert_f(IsAdmin())
and_f(<A>, or_f(<B>, <C>))