aiotdlib

repository·main·Indexed 40 days ago

https://github.com/pylakey/aiotdlib

An asynchronous Python Telegram client built on top of TDLib. It features high-level asyncio wrappers, Pydantic-based type validation, and automatic API generation from the TDLib schema. The library provides a Client class for lifecycle management, a robust event handling system with specialized filters and middleware, and high-level helpers for sending various media types and iterating through chat history.

Tokens
9.2K
Snippets
37
Records
41
Agent score
49%

What's inside aiotdlib

  1. Use Middlewares for event processing

    main

    Middlewares allow you to execute logic before and after event handlers are called. They are executed for every event received.

    Important: Middlewares are called in the order they are added. Avoid long-running tasks in middlewares as they can significantly impact performance.

    Middleware signature: async def my_middleware(client: Client, event: BaseObject, /, *, call_next: HandlerCallable):

    You must call await call_next(client, event) to proceed to the next middleware or event handler.

    from aiotdlib import Client, ClientSettings, HandlerCallable
    from aiotdlib.api import BaseObject
    
    async def my_middleware(client: Client, event: BaseObject, /, *, call_next: HandlerCallable):
        # Pre-event logic
        print(f"Before event {event.ID}")
        
        try:
            await call_next(client, event)
        finally:
            # Post-event logic
            print(f"After event {event.ID}")
    
    async def main():
        client = Client(settings=ClientSettings(api_id=123, api_hash="", phone_number=""))
        client.add_middleware(my_middleware)
        async with client:
            await client.idle()
  2. Initialize and use the aiotdlib Client

    main

    The Client class is the main entry point. You can configure it using ClientSettings. You can also provide configuration via environment variables with the prefix AIOTDLIB_* (e.g., AIOTDLIB_API_ID, AIOTDLIB_API_HASH, AIOTDLIB_BOT_TOKEN).

    All TDLib functions are accessible through the client.api field.

    import asyncio
    from aiotdlib import Client, ClientSettings
    
    API_ID = 123456
    API_HASH = ""
    PHONE_NUMBER = ""
    
    async def main():
        client = Client(
            settings=ClientSettings(
                api_id=API_ID,
                api_hash=API_HASH,
                phone_number=PHONE_NUMBER
            )
        )
    
        async with client:
            # Access TDLib functions via client.api
            me = await client.api.get_me()
            print(f"Logged in as: {me.model_dump_json()}")
    
    if __name__ == '__main__':
        asyncio.run(main())
  3. Install aiotdlib

    main

    You can install aiotdlib using pip or poetry.

    Requirements:

    • Python 3.9+
    • Telegram api_id and api_hash (obtainable from Telegram docs).

    Note: This package includes prebuilt TDLib binaries for macOS (arm64) and Debian (amd64). To use a custom binary, pass the library_path argument to the Client constructor.

    # Using pip
    pip install aiotdlib
    
    # Using poetry
    poetry add aiotdlib
  4. How to use Filters to process Telegram updates

    main

    Filters are used to intercept and process incoming Telegram updates (like UpdateNewMessage). They can act as simple boolean checks or as 'data filters' that extract information and merge it into the update.EXTRA dictionary for later use in your handlers.

    Key Concepts

    • Boolean Filters: Return True if the update matches the criteria, otherwise False.
    • Data Filters: If a filter matches, it can return a dict. This dictionary is automatically merged into update.EXTRA. This is useful for extracting command arguments or regex matches.
    • Logical Composition: You can combine filters using bitwise operators:
      • & (AND): Both filters must match.
      • | (OR): Either filter must match.
      • ^ (XOR): Exactly one filter must match.
      • ~ (NOT): The inverse of the filter.
    from aiotdlib.filters import Filters
    
    # Example: Match only text messages that start with a specific command
    my_filter = Filters.text & Filters.bot_command(command="/start")
    
    # Example: Match text OR photos
    combined_filter = Filters.text | Filters.photo
  5. Configure a Proxy

    main

    You can route your Telegram connection through a proxy by configuring proxy_settings within ClientSettings. Supported types include ClientProxyType.SOCKS5.

    from aiotdlib import Client, ClientSettings, ClientProxySettings, ClientProxyType
    
    async def main():
        client = Client(
            settings=ClientSettings(
                api_id=123456,
                api_hash="",
                phone_number="",
                proxy_settings=ClientProxySettings(
                    host="10.0.0.1",
                    port=3333,
                    type=ClientProxyType.SOCKS5,
                    username="aiotdlib",
                    password="somepassword",
                )
            )
        )
        async with client:
            await client.idle()
  6. Initialize and manage the Client lifecycle

    main

    The Client class is the primary entry point for interacting with Telegram via TDLib. It can be used as an asynchronous context manager to ensure proper startup and shutdown.

    To start the client, use await client.start(). This handles TDLib parameter setup, proxy configuration, and the authorization process (phone number, bot token, etc.). To stop the client and clean up resources, use await client.stop() or exit the async with block.

    from aiotdlib.client import Client
    from aiotdlib.client_settings import ClientSettings
    
    async def main():
        settings = ClientSettings(api_id=123, api_hash='hash', bot_token='token')
        async with Client(settings=settings) as client:
            # Your code here
            pass
    
    import asyncio
    asyncio.run(main())
  7. Handle Telegram events with event handlers

    main

    You can register asynchronous handlers for specific event types using client.add_event_handler.

    • To listen for a specific event, use update_type=API.Types.UPDATE_NEW_MESSAGE (or other types from API.Types).
    • To listen for all events, use update_type=API.Types.ANY.
    • Use await client.idle() to keep the client running and listening for events.
    import asyncio
    from aiotdlib import Client, ClientSettings
    from aiotdlib.api import API, BaseObject, UpdateNewMessage
    
    async def on_update_new_message(client: Client, update: UpdateNewMessage):
        chat_id = update.message.chat_id
        chat = await client.api.get_chat(chat_id)
        print(f'Message received in chat {chat.title}')
    
    async def any_event_handler(client: Client, update: BaseObject):
        print(f'Event of type {update.ID} received')
    
    async def main():
        client = Client(settings=ClientSettings(api_id=123, api_hash="", phone_number=""))
    
        # Register specific handler
        client.add_event_handler(on_update_new_message, update_type=API.Types.UPDATE_NEW_MESSAGE)
        
        # Register catch-all handler
        client.add_event_handler(any_event_handler, update_type=API.Types.ANY)
    
        async with client:
            await client.idle()
    
    if __name__ == '__main__':
        asyncio.run(main())
  8. Implement Bot commands

    main

    For bot implementations, use the bot_command_handler method (which can be used as a decorator) to respond to slash commands.

    When a command is triggered, the handler receives an UpdateNewMessage object where the update.EXTRA field is populated with command metadata: {'bot_command': 'command_name', 'bot_command_args': ['arg1', 'arg2']}.

    from aiotdlib import Client, ClientSettings
    from aiotdlib.api import UpdateNewMessage
    
    BOT_TOKEN = ""
    
    bot = Client(
        settings=ClientSettings(api_id=123, api_hash="", bot_token=BOT_TOKEN)
    )
    
    @bot.bot_command_handler(command='help')
    async def on_help_command(client: Client, update: UpdateNewMessage):
        # update.EXTRA contains {'bot_command': 'help', 'bot_command_args': []}
        await client.send_text(update.message.chat_id, "I will help you!")
    
    # Or register via method call
    async def on_start_command(client: Client, update: UpdateNewMessage):
        print(update.EXTRA)
        await client.send_text(update.message.chat_id, "Have a good day!")
    
    bot.bot_command_handler(on_start_command, command='start')
    
    if __name__ == '__main__':
        bot.run()
  9. Use environment variables to configure ClientSettings

    main

    Since ClientSettings inherits from pydantic_settings.BaseSettings, you can configure it using environment variables. All variables must be prefixed with aiotdlib_.

    Example Environment Variables:

    • AIOTDLIB_API_ID
    • AIOTDLIB_API_HASH
    • AIOTDLIB_PHONE_NUMBER
    • AIOTDLIB_BOT_TOKEN

    Settings can also be loaded from a .env file.

  10. Use specialized message handlers

    main

    The Client provides convenience methods for common message types using predefined filters:

    • text_message_handler(function): Registers a handler for UPDATE_NEW_MESSAGE events that specifically contain text content.
    • bot_command_handler(function, command=None): Registers a handler for UPDATE_NEW_MESSAGE events where the text starts with / (and optionally matches a specific command).
    # As a decorator
    @client.text_message_handler
    async def on_text(client, update):
        print(update.message.content.text)
    
    # As a command handler
    @client.bot_command_handler(command="start")
    async def on_start(client, update):
        print("User sent /start")
  11. Send an animation with `send_animation()`

    main

    Sends an animation (e.g., GIF) to a target chat. The animation parameter can be a local file path or a remote file_id.

    await client.send_animation(
        chat_id=123456,
        animation="/path/to/animation.gif",
        caption="Cool animation"
    )
  12. Execute TDLib queries with `request` and `execute`

    main

    For low-level interaction with TDLib:

    • request(query, request_id=None, request_timeout=10): Sends a query and waits for the specific response associated with that request_id. It returns the BaseObject result of the request.
    • execute(query): Executes a query without waiting for a specific response object, returning the raw result of the execution. It raises AioTDLibError if the execution fails.
    from aiotdlib.api import GetCurrentState
    
    # Using request to get a specific object
    state = await client.request(GetCurrentState())
    
    # Using execute for raw commands
    await client.execute(SetLogVerbosityLevel(new_verbosity_level=1))