Skip to main content

Aiogram-style Router for pyTeleBot โ€” simple, async-ready, and powerful routing system.

Project description

Telebot routers ๐Ÿค–

Powerful Router System for Sync and Async Telegram Bots

๐Ÿš€ Features

  • โœ… Sync & Async Support - TeleBot and AsyncTeleBot
  • โœ… Global Routers - Same handlers for multiple bots
  • โœ… Simple Routers - Dedicated handlers for single bot
  • โœ… Advanced Filters - Command, Text, IsAdmin and more
  • โœ… Modular Design - Clean file structure
  • โœ… Full Documentation - Examples for every method

๐Ÿ“ฆ Installation

pip install telebot-router

Or from source:

git clone https://github.com/Ergashev2006/telebot-router.git
cd telebot-router
pip install - e .

๐ŸŽฏ Quick Start

Sync Bot (TeleBot)

from telebot import TeleBot
from telebot_router import Router, Command

bot = TeleBot("YOUR_BOT_TOKEN")
router = Router("my_bot")

@router.message_handler(commands=['start'])
def start_handler(message, bot):
    bot.reply_to(message, "๐Ÿ‘‹ Hello! Sync bot started!")

@router.message_handler(commands=['help'])
def help_handler(message, bot):
    bot.reply_to(message, "๐Ÿ“– Need help?")

router.register(bot)
bot.polling()

Async Bot (AsyncTeleBot)

from telebot.async_telebot import AsyncTeleBot
from telebot_router import AsyncRouter, Command
import asyncio

bot = AsyncTeleBot("YOUR_BOT_TOKEN")
router = AsyncRouter("async_bot")

@router.message_handler(commands=['start'])
async def start_handler(message, bot):
    await bot.reply_to(message, "โšก Hello! Async bot started!")

router.register(bot)

async def main():
    await bot.polling()

asyncio.run(main())

๐ŸŒ Global Handlers

Multiple Sync Bots

from telebot import TeleBot
from telebot_router import message_handler, register_sync_bot

# Global handler - works for all sync bots
@message_handler(commands=['start'])
def global_start(message, bot):
    bot.reply_to(message, "๐ŸŒ This is global handler! Works in all bots!")

# Bot 1
bot1 = TeleBot("TOKEN1")
register_sync_bot(bot1)

# Bot 2
bot2 = TeleBot("TOKEN2")
register_sync_bot(bot2)

# Start only one bot
bot1.polling()

Multiple Async Bots

from telebot.async_telebot import AsyncTeleBot
from telebot_router import async_message_handler, register_async_bot
import asyncio

# Global async handler
@async_message_handler(commands=['start'])
async def global_async_start(message, bot):
    await bot.reply_to(message, "๐ŸŒโšก This is global async handler!")

# Async bots
bot1 = AsyncTeleBot("TOKEN1")
bot2 = AsyncTeleBot("TOKEN2")

register_async_bot(bot1)
register_async_bot(bot2)

async def main():
    await bot1.polling()

asyncio.run(main())

๐Ÿ” Filters

Command Filter

from telebot_router import Command

@router.message_handler(commands=['start', 'help'])
def command_handler(message, bot):
    bot.reply_to(message, "Command received!")

Text Filter

from telebot_router import Text

@router.message_handler(func=Text('hello'))
def text_handler(message, bot):
    bot.reply_to(message, "Hello! How are you?")

TextContains Filter

from telebot_router import TextContains

@router.message_handler(func=TextContains(['help', 'support']))
def contains_handler(message, bot):
    bot.reply_to(message, "Need help?")

IsAdmin Filter

from telebot_router import IsAdmin

ADMIN_IDS = [123456789, 987654321]

@router.message_handler(func=IsAdmin(ADMIN_IDS), commands=['admin'])
def admin_handler(message, bot):
    bot.reply_to(message, "๐Ÿ‘‘ Welcome to admin panel!")

CallbackData Filter

from telebot_router import CallbackData
from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton

@router.callback_query_handler(func=CallbackData('button1'))
def callback_handler(call, bot):
    bot.answer_callback_query(call.id, "Button clicked!")

๐Ÿ—๏ธ Modular Project Structure

handlers/start.py

from telebot_router import Router, Command

router = Router("start_handlers")

@router.message_handler(commands=['start'])
def start_handler(message, bot):
    bot.reply_to(message, "Start handler activated!")

handlers/admin.py

from telebot_router import Router, Command, IsAdmin

router = Router("admin_handlers")
ADMIN_IDS = [123456789]

@router.message_handler(func=IsAdmin(ADMIN_IDS), commands=['secret'])
def secret_handler(message, bot):
    bot.reply_to(message, "Secret admin command!")

main.py

from telebot import TeleBot
from telebot_router import Router
from handlers.start import router as start_router
from handlers.admin import router as admin_router

bot = TeleBot("YOUR_BOT_TOKEN")
main_router = Router("main")

# Combine all routers
main_router.include_router(start_router)
main_router.include_router(admin_router)

main_router.register(bot)
bot.polling()

๐Ÿ“Š Statistics

Handler Count

from telebot_router import get_sync_global_stats, get_async_global_stats

# Sync global statistics
sync_stats = get_sync_global_stats()
print(f"Sync bots: {sync_stats['total_bots']}")
print(f"Sync handlers: {sync_stats['handlers']}")

# Async global statistics  
async_stats = get_async_global_stats()
print(f"Async bots: {async_stats['total_bots']}")
print(f"Async handlers: {async_stats['handlers']}")

Simple Router Statistics

router = Router("test")
stats = router.get_handler_count()
print(f"Message handlers: {stats['message_handlers']}")
print(f"Callback handlers: {stats['callback_handlers']}")

๐ŸŽฏ Additional Features

Include Routers

from telebot_router import include_sync_router, include_async_router

# Include sync router
include_sync_router(my_router)

# Include async router
include_async_router(my_async_router)

Clear Handlers

router.clear_handlers()  # Clear all handlers

๐Ÿค Contributing

Want to contribute?

  1. Fork the repository
  2. Create a new branch (git checkout - b feature/awesome- feature)
  3. Commit your changes (git commit - am 'Add awesome feature')
  4. Push to the branch (git push origin feature/awesome- feature)
  5. Create a Pull Request

๐Ÿ“ License

This project is licensed under the MIT License. See LICENSE file for details.

๐Ÿ“ž Contact

If you have questions or need help:

๐Ÿ™ Acknowledgments

  • TeleBot - Main Telegram Bot library
  • All contributors and testers

โญ If you like this project, give it a star!


๐Ÿš€ Quick Reference

Imports

# Sync
from telebot_router import Router, GlobalRouter, message_handler, callback_handler

# Async  
from telebot_router import AsyncRouter, AsyncGlobalRouter, async_message_handler, async_callback_handler

# Filters
from telebot_router import Command, Text, TextContains, IsAdmin, CallbackData

# Functions
from telebot_router import register_sync_bot, register_async_bot, get_sync_global_stats, get_async_global_stats

Main Methods

  • router.message_handler() - Add message handler
  • router.callback_query_handler() - Add callback handler
  • router.include_router() - Include another router
  • router.register() - Register to bot
  • router.get_handler_count() - Get statistics
  • router.clear_handlers() - Clear handlers

Happy Coding! ๐ŸŽ‰


Telegram Bot Test Code#

๐Ÿงช Complete Test Bot with All Features

  1. test_bot.py - Main Test Bot
from telebot import TeleBot
from telebot_router import Router, Command, Text, TextContains, IsAdmin, CallbackData
from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton
import logging
import time

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

def create_test_bot():
    """Create and configure test bot"""
    
    # Bot configuration
    BOT_TOKEN = "YOUR_BOT_TOKEN_HERE"  # โš ๏ธ Replace with your token
    ADMIN_IDS = [123456789]  # โš ๏ธ Replace with your user ID
    
    if BOT_TOKEN == "YOUR_BOT_TOKEN_HERE":
        print("โŒ ERROR: Please set your bot token!")
        print("๐Ÿ’ก Get token from @BotFather in Telegram")
        return None
    
    # Create bot and router
    bot = TeleBot(BOT_TOKEN)
    router = Router("test_bot")
    
    # ==================== TEST HANDLERS ====================
    
    # 1. Basic command test
    @router.message_handler(commands=['start'])
    def start_test(message, bot):
        """Test basic command handling"""
        user = message.from_user
        welcome_text = f"""
๐Ÿค– **TEST BOT ACTIVATED**

๐Ÿ‘ค User: {user.first_name}
๐Ÿ†” ID: `{user.id}`
๐Ÿ”ง Status: **WORKING**

๐Ÿ“‹ Available Tests:
/start - Show this message
/ping - Test response time  
/echo [text] - Echo test
/stats - Handler statistics
/menu - Show test buttons
/admin - Admin test (if admin)
        """
        bot.reply_to(message, welcome_text, parse_mode='Markdown')
        logger.info(f"Start test: {user.username}")
    
    # 2. Ping test
    @router.message_handler(commands=['ping'])
    def ping_test(message, bot):
        """Test server response time"""
        start_time = time.time()
        msg = bot.reply_to(message, "๐Ÿ“ Pinging...")
        end_time = time.time()
        
        response_time = round((end_time - start_time) * 1000, 2)
        
        bot.edit_message_text(
            f"๐Ÿ“ **Pong!**\nโฑ Response: {response_time}ms\nโœ… Router working!",
            message.chat.id,
            msg.message_id,
            parse_mode='Markdown'
        )
        logger.info(f"Ping test: {response_time}ms")
    
    # 3. Echo test
    @router.message_handler(commands=['echo'])
    def echo_test(message, bot):
        """Test echo functionality"""
        text = message.text[6:]  # Remove '/echo '
        if not text:
            bot.reply_to(message, "โŒ Usage: `/echo your text`", parse_mode='Markdown')
            return
        
        bot.reply_to(message, f"๐Ÿ” Echo: {text}")
        logger.info(f"Echo test: {text}")
    
    # 4. Statistics test
    @router.message_handler(commands=['stats'])
    def stats_test(message, bot):
        """Test router statistics"""
        stats = router.get_handler_count()
        
        stats_text = f"""
๐Ÿ“Š **ROUTER STATISTICS**

๐Ÿ“จ Message Handlers: {stats['message_handlers']}
๐Ÿ”„ Callback Handlers: {stats['callback_handlers']}
๐Ÿ”ง Total Handlers: {stats['message_handlers'] + stats['callback_handlers']}

โœ… Router is working correctly!
        """
        bot.reply_to(message, stats_text, parse_mode='Markdown')
        logger.info(f"Stats test: {stats}")
    
    # 5. Menu with buttons test
    @router.message_handler(commands=['menu'])
    def menu_test(message, bot):
        """Test inline keyboard buttons"""
        keyboard = InlineKeyboardMarkup(row_width=2)
        keyboard.add(
            InlineKeyboardButton("โœ… Test Button 1", callback_data="test_btn1"),
            InlineKeyboardButton("๐Ÿ”„ Test Button 2", callback_data="test_btn2"),
            InlineKeyboardButton("๐Ÿ“Š Get Stats", callback_data="get_stats"),
            InlineKeyboardButton("โŒ Delete Menu", callback_data="delete_menu")
        )
        
        bot.reply_to(
            message,
            "๐ŸŽฎ **TEST MENU**\n\nSelect a button to test callback handlers:",
            parse_mode='Markdown',
            reply_markup=keyboard
        )
    
    # 6. Text filter test
    @router.message_handler(func=Text('hello'))
    def hello_test(message, bot):
        """Test exact text matching"""
        bot.reply_to(message, "โœ… **Text Filter Test!** - 'hello' detected!", parse_mode='Markdown')
    
    # 7. Text contains filter test
    @router.message_handler(func=TextContains(['help', 'yordam']))
    def help_test(message, bot):
        """Test text contains matching"""
        bot.reply_to(message, "โœ… **TextContains Filter!** - Help word found!")
    
    # 8. Admin filter test
    @router.message_handler(func=IsAdmin(ADMIN_IDS), commands=['admin'])
    def admin_test(message, bot):
        """Test admin- only commands"""
        bot.reply_to(message, "๐Ÿ‘‘ **Admin Test Passed!** - You have admin access!", parse_mode='Markdown')
    
    # 9. Lambda filter test
    @router.message_handler(func=lambda m: m.text and len(m.text) > 20)
    def long_text_test(message, bot):
        """Test custom lambda filter"""
        bot.reply_to(message, f"๐Ÿ“ **Long Text Test!** - {len(message.text)} characters")
    
    # ==================== CALLBACK HANDLERS ====================
    
    @router.callback_query_handler(func=CallbackData('test_btn1'))
    def button1_test(call, bot):
        """Test callback button 1"""
        bot.answer_callback_query(call.id, "โœ… Button 1 clicked!")
        bot.send_message(call.message.chat.id, "๐ŸŽ‰ **Button 1 Test Passed!**")
    
    @router.callback_query_handler(func=CallbackData('test_btn2'))
    def button2_test(call, bot):
        """Test callback button 2"""
        bot.answer_callback_query(call.id, "๐Ÿ”„ Button 2 clicked!")
        bot.edit_message_text(
            "๐Ÿ”„ **Button 2 Test Passed!**\nMessage edited successfully!",
            call.message.chat.id,
            call.message.message_id,
            parse_mode='Markdown'
        )
    
    @router.callback_query_handler(func=CallbackData('get_stats'))
    def stats_callback_test(call, bot):
        """Test stats in callback"""
        stats = router.get_handler_count()
        stats_text = f"""
๐Ÿ“Š **CALLBACK STATS**

Message Handlers: {stats['message_handlers']}
Callback Handlers: {stats['callback_handlers']}
โœ… All systems working!
        """
        
        bot.answer_callback_query(call.id, "๐Ÿ“Š Stats updated!")
        bot.edit_message_text(
            stats_text,
            call.message.chat.id,
            call.message.message_id,
            parse_mode='Markdown'
        )
    
    @router.callback_query_handler(func=CallbackData('delete_menu'))
    def delete_menu_test(call, bot):
        """Test message deletion"""
        bot.answer_callback_query(call.id, "๐Ÿ—‘๏ธ Menu deleted!")
        bot.delete_message(call.message.chat.id, call.message.message_id)
    
    # ==================== ERROR HANDLER ====================
    
    @router.message_handler(func=lambda message: True)
    def unknown_message(message, bot):
        """Handle unknown commands"""
        bot.reply_to(
            message,
            "โ“ **Unknown Command**\n\n"
            "Try these test commands:\n"
            "โ€ข /start - Show help\n"
            "โ€ข /ping - Test speed\n"  
            "โ€ข /echo [text] - Echo test\n"
            "โ€ข /stats - Show statistics\n"
            "โ€ข /menu - Test buttons\n"
            "โ€ข 'hello' - Text filter test\n"
            "โ€ข 'help me' - Contains filter test",
            parse_mode='Markdown'
        )
    
    return bot, router

def main():
    """Main function to run the test bot"""
    
    print("๐Ÿš€ TELEGRAM BOT TEST SUITE")
    print("=" * 50)
    
    # Create test bot
    result = create_test_bot()
    if not result:
        return
    
    bot, router = result
    
    # Register router
    router.register(bot)
    
    # Display test information
    stats = router.get_handler_count()
    print("โœ… Test Bot Configured!")
    print(f"๐Ÿ“Š Handlers: {stats['message_handlers']} message, {stats['callback_handlers']} callback")
    print("\n๐ŸŽฏ **Available Tests:**")
    print("   /start    - Basic functionality")
    print("   /ping     - Response time") 
    print("   /echo     - Echo test")
    print("   /stats    - Router statistics")
    print("   /menu     - Button tests")
    print("   /admin    - Admin test (if admin)")
    print("   'hello'   - Text filter")
    print("   'help'    - Contains filter")
    print("   Long text - Lambda filter")
    print("\n๐Ÿ“ฑ Send /start to your bot to begin testing!")
    print("=" * 50)
    
    try:
        print("๐Ÿค– Starting bot polling...")
        bot.polling(non_stop=True, interval=0)
    except KeyboardInterrupt:
        print("\nโน๏ธ Bot stopped by user")
    except Exception as e:
        print(f"โŒ Error: {e}")

if __name__ == "__main__":
    main()
  1. test_async_bot.py - Async Test Bot
from telebot.async_telebot import AsyncTeleBot
from telebot_router import AsyncRouter, Command, CallbackData
import asyncio
import logging
import time

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def create_async_test_bot():
    """Create async test bot"""
    
    BOT_TOKEN = "YOUR_BOT_TOKEN_HERE"  # โš ๏ธ Replace with your token
    
    if BOT_TOKEN == "YOUR_BOT_TOKEN_HERE":
        print("โŒ ERROR: Please set your bot token!")
        return None
    
    bot = AsyncTeleBot(BOT_TOKEN)
    router = AsyncRouter("async_test_bot")
    
    # Async test handlers
    @router.message_handler(commands=['start'])
    async def async_start_test(message, bot):
        """Async start test"""
        await bot.reply_to(
            message,
            "โšก **ASYNC BOT TEST**\n\n"
            "โœ… Async router working!\n"
            "๐Ÿ”ง All features available\n"
            "๐Ÿš€ Fast and efficient!",
            parse_mode='Markdown'
        )
    
    @router.message_handler(commands=['async_ping'])
    async def async_ping_test(message, bot):
        """Async ping test"""
        start_time = time.time()
        msg = await bot.reply_to(message, "โšก Async pinging...")
        end_time = time.time()
        
        response_time = round((end_time - start_time) * 1000, 2)
        
        await bot.edit_message_text(
            f"โšก **Async Pong!**\nโฑ {response_time}ms\n๐ŸŽฏ Async router working!",
            message.chat.id,
            msg.message_id,
            parse_mode='Markdown'
        )
    
    @router.callback_query_handler(func=CallbackData('async_btn'))
    async def async_button_test(call, bot):
        """Async callback test"""
        await bot.answer_callback_query(call.id, "โšก Async button clicked!")
        await bot.send_message(call.message.chat.id, "โœ… **Async callback working!**")
    
    return bot, router

async def main():
    """Main async function"""
    print("โšก ASYNC BOT TEST SUITE")
    print("=" * 40)
    
    result = await create_async_test_bot()
    if not result:
        return
    
    bot, router = result
    router.register(bot)
    
    stats = router.get_handler_count()
    print(f"โœ… Async Bot Ready!")
    print(f"๐Ÿ“Š Async handlers: {stats}")
    print("\n๐ŸŽฏ Test commands:")
    print("   /start      - Async test")
    print("   /async_ping - Async speed test")
    print("=" * 40)
    
    try:
        print("๐Ÿค– Starting async polling...")
        await bot.polling(non_stop=True)
    except Exception as e:
        print(f"โŒ Async error: {e}")

if __name__ == "__main__":
    asyncio.run(main())
  1. test_filters.py - Filter Testing
from telebot_router import Command, Text, TextContains, IsAdmin, CallbackData
from telebot.types import Message, CallbackQuery, User
import logging

logging.basicConfig(level=logging.INFO)

def test_all_filters():
    """Test all filters"""
    
    print("๐Ÿ” FILTER TESTING")
    print("=" * 40)
    
    # Test data
    test_message = Message()
    test_message.from_user = User()
    test_message.from_user.id = 123456789
    
    test_callback = CallbackQuery()
    test_callback.data = "test_data"
    
    # 1. Command filter test
    print("\n1. Testing Command Filter...")
    cmd_filter = Command(['start', 'help'])
    test_message.text = "/start"
    result = cmd_filter(test_message)
    print(f"   Command '/start': {result} โœ…" if result else "   โŒ FAILED")
    
    # 2. Text filter test
    print("\n2. Testing Text Filter...")
    text_filter = Text('hello')
    test_message.text = "hello"
    result = text_filter(test_message)
    print(f"   Text 'hello': {result} โœ…" if result else "   โŒ FAILED")
    
    # 3. TextContains filter test
    print("\n3. Testing TextContains Filter...")
    contains_filter = TextContains(['help', 'support'])
    test_message.text = "I need help"
    result = contains_filter(test_message)
    print(f"   Contains 'help': {result} โœ…" if result else "   โŒ FAILED")
    
    # 4. IsAdmin filter test
    print("\n4. Testing IsAdmin Filter...")
    admin_filter = IsAdmin([123456789, 999999999])
    test_message.from_user.id = 123456789
    result = admin_filter(test_message)
    print(f"   Admin ID 123456789: {result} โœ…" if result else "   โŒ FAILED")
    
    # 5. CallbackData filter test
    print("\n5. Testing CallbackData Filter...")
    callback_filter = CallbackData('test_data')
    test_callback.data = "test_data"
    result = callback_filter(test_callback)
    print(f"   Callback data 'test_data': {result} โœ…" if result else "   โŒ FAILED")
    
    print("\n" + "=" * 40)
    print("๐ŸŽ‰ ALL FILTER TESTS COMPLETED!")

if __name__ == "__main__":
    test_all_filters()

๐Ÿš€ How to Use

  1. Setup:
# Install required packages
pip install telebot-router

# Get bot token from @BotFather in Telegram
# Replace "YOUR_BOT_TOKEN_HERE" with your actual token
  1. Run Tests:
# Run main test bot
python test_bot.py

# Run async test bot (in separate terminal)
python test_async_bot.py

# Run filter tests
python test_filters.py
  1. Test Commands:

Send these to your bot:

  • /start - Show help and bot info
  • /ping - Test response speed
  • /echo Hello - Echo test
  • /stats - Router statistics
  • /menu - Button tests
  • hello - Text filter test
  • help me - Contains filter test
  • /admin - Admin test (if you're admin)

โœ… Expected Results

  • โœ… All commands should work
  • โœ… Buttons should respond immediately
  • โœ… Statistics should show handler counts
  • โœ… Filters should match correct messages
  • โœ… Async bot should work smoothly

This test suite covers all Telebot routers features! ๐ŸŽ‰

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

telebot_router-0.3.0.tar.gz (13.8 kB view details)

Uploaded Source

Built Distribution

If you're not sure about the file name format, learn more about wheel file names.

telebot_router-0.3.0-py3-none-any.whl (8.3 kB view details)

Uploaded Python 3

File details

Details for the file telebot_router-0.3.0.tar.gz.

File metadata

  • Download URL: telebot_router-0.3.0.tar.gz
  • Upload date:
  • Size: 13.8 kB
  • Tags: Source
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.2

File hashes

Hashes for telebot_router-0.3.0.tar.gz
Algorithm Hash digest
SHA256 fa5abad2d67c7f753b6f06f6e66df0dff6bffdf585445548bad9e42b644fa56a
MD5 1beba17410f770aaa6c35fa5d67f07e2
BLAKE2b-256 27d52168491dfce71e6f5c5f7ad071688905c42fe258f68fb9387ca4aec47d7d

See more details on using hashes here.

File details

Details for the file telebot_router-0.3.0-py3-none-any.whl.

File metadata

  • Download URL: telebot_router-0.3.0-py3-none-any.whl
  • Upload date:
  • Size: 8.3 kB
  • Tags: Python 3
  • Uploaded using Trusted Publishing? No
  • Uploaded via: twine/6.2.0 CPython/3.13.2

File hashes

Hashes for telebot_router-0.3.0-py3-none-any.whl
Algorithm Hash digest
SHA256 257d612a8935b5c33182fdbfcfe42ef8acaf94647f3ab7a6cd7d944bd841444e
MD5 a089cbd688329b3a167311a7696bae2b
BLAKE2b-256 4f9cc85525371e5ccff31f7934abdb2e1fceaf45d0e2753aa242532ce84b5dda

See more details on using hashes here.

Supported by

AWS Cloud computing and Security Sponsor Datadog Monitoring Depot Continuous Integration Fastly CDN Google Download Analytics Pingdom Monitoring Sentry Error logging StatusPage Status page