tessl/pypi-aiogram

Modern and fully asynchronous framework for Telegram Bot API

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Filters and Handlers

Name: tessl/pypi-aiogram
Author: tessl

Powerful filtering system with magic filters, command matching, state filtering, and custom filters. Handler registration with decorators and functional approaches for organizing bot logic.

Capabilities

Base Filter Classes

Foundation classes for creating filters and handling updates.

class Filter:
    """Base class for all filters"""
    
    async def __call__(self, obj: TelegramObject) -> bool | dict[str, Any]:
        """
        Check if the filter matches the object.
        
        Parameters:
        - obj: Telegram object to filter
        
        Returns:
        - bool: True if filter matches
        - dict: Filter data to pass to handler if matches
        - False: Filter doesn't match
        """

class BaseFilter(Filter):
    """Alias for Filter class"""
    pass

Built-in Filters

Pre-built filters for common use cases.

class Command(Filter):
    """Filter for bot commands"""
    
    def __init__(
        self,
        *commands: str,
        prefix: str = "/",
        ignore_case: bool = False,
        ignore_mention: bool = False
    ):
        """
        Initialize command filter.
        
        Parameters:
        - commands: Command names (without prefix)
        - prefix: Command prefix (default: "/")
        - ignore_case: Case-insensitive matching
        - ignore_mention: Ignore bot mentions in commands
        """

class CommandStart(Command):
    """Specialized filter for /start command with deep linking support"""
    
    def __init__(
        self,
        deep_link: bool = False,
        deep_link_encoded: bool = False,
        ignore_case: bool = False,
        ignore_mention: bool = False
    ):
        """
        Initialize /start command filter.
        
        Parameters:
        - deep_link: Expect deep link parameter
        - deep_link_encoded: Deep link parameter is base64 encoded
        - ignore_case: Case-insensitive matching
        - ignore_mention: Ignore bot mentions
        """

class StateFilter(Filter):
    """Filter for FSM states"""
    
    def __init__(self, *states: State | str | None):
        """
        Initialize state filter.
        
        Parameters:
        - states: States to match (None matches any state)
        """

class ChatMemberUpdatedFilter(Filter):
    """Filter for chat member status changes"""
    
    def __init__(
        self,
        member_status_changed: bool = True,
        is_member: bool | None = None,
        join_transition: bool = False,
        leave_transition: bool = False
    ):
        """
        Initialize chat member updated filter.
        
        Parameters:
        - member_status_changed: Filter member status changes
        - is_member: Filter by membership status
        - join_transition: Filter join transitions
        - leave_transition: Filter leave transitions
        """

class ExceptionTypeFilter(Filter):
    """Filter exceptions by type"""
    
    def __init__(self, *exception_types: type[Exception]):
        """
        Initialize exception type filter.
        
        Parameters:
        - exception_types: Exception types to match
        """

class ExceptionMessageFilter(Filter):
    """Filter exceptions by message content"""
    
    def __init__(self, *messages: str, ignore_case: bool = True):
        """
        Initialize exception message filter.
        
        Parameters:
        - messages: Exception messages to match
        - ignore_case: Case-insensitive matching
        """

Magic Filters

Advanced filtering system with chained conditions and dynamic property access.

class MagicFilter:
    """Magic filter for complex conditions"""
    
    def __getattr__(self, name: str) -> MagicFilter:
        """Access object attributes dynamically"""
    
    def __call__(self, *args, **kwargs) -> MagicFilter:
        """Call the filtered object as function"""
    
    def __eq__(self, other: Any) -> MagicFilter:
        """Equality comparison"""
    
    def __ne__(self, other: Any) -> MagicFilter:
        """Inequality comparison"""
    
    def __lt__(self, other: Any) -> MagicFilter:
        """Less than comparison"""
    
    def __le__(self, other: Any) -> MagicFilter:
        """Less than or equal comparison"""
    
    def __gt__(self, other: Any) -> MagicFilter:
        """Greater than comparison"""
    
    def __ge__(self, other: Any) -> MagicFilter:
        """Greater than or equal comparison"""
    
    def __and__(self, other: MagicFilter) -> MagicFilter:
        """Logical AND operation"""
    
    def __or__(self, other: MagicFilter) -> MagicFilter:
        """Logical OR operation"""
    
    def __invert__(self) -> MagicFilter:
        """Logical NOT operation"""
    
    def in_(self, container: Any) -> MagicFilter:
        """Check if value is in container"""
    
    def contains(self, item: Any) -> MagicFilter:
        """Check if container contains item"""
    
    def startswith(self, prefix: str) -> MagicFilter:
        """Check if string starts with prefix"""
    
    def endswith(self, suffix: str) -> MagicFilter:
        """Check if string ends with suffix"""
    
    def regexp(self, pattern: str) -> MagicFilter:
        """Match against regular expression"""
    
    def func(self, function: Callable[[Any], bool]) -> MagicFilter:
        """Apply custom function"""
    
    def as_(self, name: str) -> MagicFilter:
        """Capture result with given name"""

# Global magic filter instance
F: MagicFilter

Callback Data Filters

Structured callback data handling with automatic serialization.

class CallbackData:
    """Callback data factory for structured inline keyboard data"""
    
    def __init__(self, *parts: str, sep: str = ":"):
        """
        Initialize callback data factory.
        
        Parameters:
        - parts: Data part names
        - sep: Separator between parts
        """
    
    def new(self, **values: str | int) -> str:
        """Create callback data string"""
    
    def filter(self, **values: str | int | None) -> CallbackDataFilter:
        """Create filter for this callback data"""
    
    def parse(self, callback_data: str) -> dict[str, str]:
        """Parse callback data string"""

class CallbackDataFilter(Filter):
    """Filter for structured callback data"""
    pass

Logic Operations

Combine filters with logical operations.

def and_f(*filters: Filter) -> Filter:
    """Combine filters with logical AND"""

def or_f(*filters: Filter) -> Filter:
    """Combine filters with logical OR"""

def invert_f(filter: Filter) -> Filter:
    """Invert filter with logical NOT"""

Handler Registration

Handler registration through decorators and programmatic methods.

# Decorator usage on router observers
@router.message(Command("start"))
async def start_handler(message: Message): ...

@router.callback_query(F.data == "button_clicked")
async def button_handler(callback_query: CallbackQuery): ...

@router.message(F.text.contains("hello"))
async def text_filter_handler(message: Message): ...

# Programmatic registration
router.message.register(start_handler, Command("start"))
router.callback_query.register(button_handler, F.data == "button_clicked")

FSM Integration

Finite State Machine integration with filters.

class State:
    """Represents a single state in FSM"""
    
    def __init__(self, state: str, group_name: str | None = None):
        """
        Initialize state.
        
        Parameters:
        - state: State name
        - group_name: Group name (auto-detected from StatesGroup)
        """

class StatesGroup:
    """Base class for grouping related states"""
    pass

# Example states group
class Form(StatesGroup):
    name = State()
    age = State()
    email = State()

# Handler with state filter
@router.message(StateFilter(Form.name), F.text)
async def process_name(message: Message, state: FSMContext):
    await state.set_data({"name": message.text})
    await state.set_state(Form.age)
    await message.answer("What's your age?")

Usage Examples

Basic Filter Usage

from aiogram import Router, F
from aiogram.types import Message, CallbackQuery
from aiogram.filters import Command, CommandStart, StateFilter

router = Router()

# Command filter
@router.message(Command("help", "info"))
async def help_handler(message: Message):
    await message.answer("Help information")

# Start command with deep linking
@router.message(CommandStart(deep_link=True))
async def start_with_link(message: Message, command: CommandObject):
    link_param = command.args
    await message.answer(f"Started with parameter: {link_param}")

# Magic filter examples
@router.message(F.text == "hello")
async def exact_text(message: Message):
    await message.answer("Hello to you too!")

@router.message(F.text.contains("python"))
async def contains_python(message: Message):
    await message.answer("I love Python too!")

@router.message(F.text.startswith("/"))
async def starts_with_slash(message: Message):
    await message.answer("That looks like a command!")

@router.message(F.from_user.is_bot == False)
async def from_human(message: Message):
    await message.answer("Hello, human!")

Advanced Magic Filter Usage

# Complex conditions
@router.message(
    (F.text.contains("hello") | F.text.contains("hi")) 
    & (F.from_user.id != 123456)
)
async def greeting_not_from_specific_user(message: Message):
    await message.answer("Hello!")

# Multiple property access
@router.message(F.chat.type == "private")
async def private_chat_only(message: Message):
    await message.answer("This is a private chat")

@router.message(F.photo[0].file_size > 1000000)  # Photo larger than 1MB
async def large_photo(message: Message):
    await message.answer("That's a large photo!")

# Content type filtering
@router.message(F.content_type.in_({"photo", "video"}))
async def media_handler(message: Message):
    await message.answer("Nice media!")

# Regular expressions
@router.message(F.text.regexp(r"^\d+$"))
async def numbers_only(message: Message):
    await message.answer("That's a number!")

# Custom function filter
def is_weekend(message):
    return datetime.now().weekday() >= 5

@router.message(F.func(is_weekend))
async def weekend_handler(message: Message):
    await message.answer("Happy weekend!")

Callback Data Usage

from aiogram.utils.keyboard import InlineKeyboardBuilder
from aiogram.filters.callback_data import CallbackData

# Define callback data structure
class ProductCallbackData(CallbackData, prefix="product"):
    action: str
    product_id: int
    category: str

# Create keyboard with structured callback data
def create_product_keyboard(product_id: int, category: str):
    builder = InlineKeyboardBuilder()
    builder.button(
        text="Buy",
        callback_data=ProductCallbackData(
            action="buy",
            product_id=product_id,
            category=category
        ).pack()
    )
    builder.button(
        text="Details",
        callback_data=ProductCallbackData(
            action="details",
            product_id=product_id,
            category=category
        ).pack()
    )
    return builder.as_markup()

# Handle callback with filter
@router.callback_query(ProductCallbackData.filter(action="buy"))
async def buy_product(callback: CallbackQuery, callback_data: ProductCallbackData):
    product_id = callback_data.product_id
    category = callback_data.category
    await callback.message.answer(f"Buying product {product_id} from {category}")
    await callback.answer()

# Filter by specific values
@router.callback_query(ProductCallbackData.filter(category="electronics"))
async def electronics_handler(callback: CallbackQuery, callback_data: ProductCallbackData):
    await callback.answer("Electronics selected!")

State Filtering

from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup

class Registration(StatesGroup):
    waiting_name = State()
    waiting_age = State()
    waiting_email = State()

# Start registration
@router.message(Command("register"))
async def start_registration(message: Message, state: FSMContext):
    await state.set_state(Registration.waiting_name)
    await message.answer("What's your name?")

# Handle name input
@router.message(StateFilter(Registration.waiting_name), F.text)
async def process_name(message: Message, state: FSMContext):
    await state.update_data(name=message.text)
    await state.set_state(Registration.waiting_age)
    await message.answer("What's your age?")

# Handle age input with validation
@router.message(StateFilter(Registration.waiting_age), F.text.regexp(r"^\d+$"))
async def process_valid_age(message: Message, state: FSMContext):
    age = int(message.text)
    if 13 <= age <= 120:
        await state.update_data(age=age)
        await state.set_state(Registration.waiting_email)
        await message.answer("What's your email?")
    else:
        await message.answer("Please enter a valid age (13-120)")

# Handle invalid age
@router.message(StateFilter(Registration.waiting_age))
async def process_invalid_age(message: Message):
    await message.answer("Please enter your age as a number")

# Complete registration
@router.message(StateFilter(Registration.waiting_email), F.text.regexp(r".+@.+\..+"))
async def process_email(message: Message, state: FSMContext):
    await state.update_data(email=message.text)
    data = await state.get_data()
    await state.clear()
    
    await message.answer(
        f"Registration complete!\n"
        f"Name: {data['name']}\n"
        f"Age: {data['age']}\n"
        f"Email: {data['email']}"
    )

Custom Filters

class AdminFilter(Filter):
    """Custom filter for admin users"""
    
    def __init__(self, admin_ids: list[int]):
        self.admin_ids = admin_ids
    
    async def __call__(self, message: Message) -> bool:
        return message.from_user and message.from_user.id in self.admin_ids

class WorkingHoursFilter(Filter):
    """Filter for working hours"""
    
    def __init__(self, start_hour: int = 9, end_hour: int = 17):
        self.start_hour = start_hour
        self.end_hour = end_hour
    
    async def __call__(self, obj: TelegramObject) -> bool:
        current_hour = datetime.now().hour
        return self.start_hour <= current_hour < self.end_hour

# Usage
admin_filter = AdminFilter([123456789, 987654321])
working_hours = WorkingHoursFilter()

@router.message(admin_filter, Command("admin"))
async def admin_command(message: Message):
    await message.answer("Admin command executed!")

@router.message(working_hours, F.text)
async def working_hours_handler(message: Message):
    await message.answer("Message received during working hours")

Error Handling Filters

@router.error(ExceptionTypeFilter(TelegramBadRequest))
async def bad_request_handler(error_event: ErrorEvent):
    print(f"Bad request: {error_event.exception}")

@router.error(ExceptionMessageFilter("Forbidden"))
async def forbidden_handler(error_event: ErrorEvent):
    print("Bot was blocked or lacks permissions")

Types

Filter Results

class CommandObject:
    """Command filter result object"""
    prefix: str
    command: str
    mention: str | None
    args: str | None

class MagicData:
    """Magic filter result data container"""
    pass

Install with Tessl CLI