aiogram-broadcaster 0.4.7

A lightweight aiogram-based library for broadcasting Telegram messages.

aiogram_broadcaster

aiogram_broadcaster is lightweight aiogram-based library for broadcasting Telegram messages.

Features

• Event system

• Placeholders (variables in texts)

• Flexible contents (LazyContent, KeyBasedContent)

• Storages

• Statistics

• Statuses

• Supports multiple mailings

• Supports multibot

Installation

• From PyPI

pip install -U aiogram-broadcaster

• From GitHub (Development build)

pip install https://github.com/loRes228/aiogram_broadcaster/archive/refs/heads/dev.zip

Creating a mailer and running broadcasting

How to create a mailer and initiate broadcasting.

Usage:

import logging
import sys
from typing import Any

from aiogram import Bot, Dispatcher, Router
from aiogram.types import Message

from aiogram_broadcaster import Broadcaster
from aiogram_broadcaster.contents import MessageSendContent
from aiogram_broadcaster.storages.file import FileMailerStorage

TOKEN = "1234:Abc"
USER_IDS = {78238238, 78378343, 98765431, 12345678}

router = Router(name=__name__)


@router.message()
async def process_any_message(message: Message, broadcaster: Broadcaster) -> Any:
    # Creating content based on the Message
    content = MessageSendContent(message=message)

    mailer = await broadcaster.create_mailer(
        content=content,
        chats=USER_IDS,
        interval=1,
        preserve=True,
        destroy_on_complete=True,
    )

    # The mailer launch method starts mailing to chats as an asyncio task.
    mailer.start()

    await message.reply(text="Run broadcasting...")


def main() -> None:
    logging.basicConfig(level=logging.INFO, stream=sys.stdout)

    bot = Bot(token=TOKEN)
    dispatcher = Dispatcher()
    dispatcher.include_router(router)

    storage = FileMailerStorage()
    broadcaster = Broadcaster(bot, storage=storage)
    broadcaster.setup(dispatcher=dispatcher)

    dispatcher.run_polling(bot)


if __name__ == "__main__":
    main()

Mailer

The Mailer class facilitates the broadcasting of messages to multiple chats in Telegram. It manages the lifecycle of the broadcast process, including starting, stopping, and destroying the broadcast.

Properties

• id: Unique identifier for the mailer.

• status: Current mailer status of the mailer (e.g., STARTED, STOPPED, COMPLETED).

• settings: Configuration settings for the mailer.

• statistic: Statistic instance containing statistics about the mailer's performance.

• content: Content to be broadcast.

• context: Additional context data used during the broadcasting process.

• bot: aiogram Bot instance used for interacting with the Telegram API.

Methods

• send(chat_id: int) -> Any: Sends the content to a specific chat identified by chat_id.

• add_chats(chats: Iterable[int]) -> Set[int]: Adds new chats to the mailer's registry.

• reset_chats() -> bool: Resets the state of all chats.

• destroy() -> None: Destroys the mailer instance and cleans up resources.

• stop() -> None: Stops the broadcasting process.

• run() -> bool: Initiates the broadcasting process.

• start() -> None: Starts the broadcasting process in background.

• wait() -> None: Waits for the broadcasting process to complete.

Usage:

mailer = await broadcaster.create_mailer(content=..., chats=...)
try:
    logging.info("Mailer starting...")
    await mailer.run()
finally:
    logging.info("Mailer shutdown...")
    await mailer.destroy()

Multibot

When using a multibot, it may be necessary to launch many mailings in several bots. For this case, there is a MailerGroup object that stores several mailers and can manage them.

Usage:

from aiogram import Bot

from aiogram_broadcaster import Broadcaster

# List of bots
bots = [Bot(token="1234:Abc"), Bot(token="5678:Vbn")]

broadcaster = Broadcaster()

# Creating a group of mailers based on several bots
mailer_group = await broadcaster.create_mailers(
    *bots,
    content=...,
    chats=...,
)

# Run all mailers in the mailer group
await mailer_group.run()

Event system

The event system empowers you to effectively manage events throughout the broadcast process.

[!NOTE] EventRegistry supports chained nesting, similar to aiogram Router.

Usage:

from aigoram_broadcaster import EventRegistry

event = EventRegistry(name=__name__)


# Define event handlers


@event.started()
async def mailer_started() -> None:
    """Triggered when the mailer begins its operations."""


@event.stopped()
async def mailer_stopped() -> None:
    """Triggered when the mailer stops its operations."""


@event.completed()
async def mailer_completed() -> None:
    """Triggered when the mailer successfully completes its operations."""


@event.before_sent()
async def mail_before_sent() -> None:
    """
    Triggered before sending content.

    Exclusive parameters for this type of event.
        chat_id (int): ID of the chat.
    """


@event.failed_sent()
async def mail_failed_sent() -> None:
    """
    Triggered when a content fails to send.

    Exclusive parameters for this type of event.
        chat_id (int): ID of the chat.
        error (Exception): Exception raised during sending.
    """


@event.success_sent()
async def mail_successful_sent() -> None:
    """
    Triggered when a mail is successfully sent.

    Exclusive parameters for this type of event:
        chat_id (int): ID of the chat.
        response (Any): Response from the sent mail.
    """


# Include the event instance in the broadcaster
broadcaster.event.bind(event)

Placeholders

Placeholders facilitate the insertion of dynamic content within texts, this feature allows for personalized messaging.

[!NOTE] PlaceholderRegistry supports chained nesting, similar to aiogram Router.

Usage:

• Function-based

from aiogram_broadcaster import PlaceholderRegistry

placeholder = PlaceholderRegistry(name=__name__)


@placeholder(key="name")
async def get_username(chat_id: int, bot: Bot) -> str:
    """Retrieves the username using the Telegram Bot API."""
    member = await bot.get_chat_member(chat_id=chat_id, user_id=chat_id)
    return member.user.first_name


broadcaster.placeholder.bind(placeholder)

• Class-based

from aiogram_broadcaster import PlaceholderItem


class NamePlaceholder(PlaceholderItem, key="name"):
    async def __call__(self, chat_id: int, bot: Bot) -> str:
        member = await bot.get_chat_member(chat_id=chat_id, user_id=chat_id)
        return member.user.first_name


broadcaster.placeholder.register(NamePlaceholder())

• Other registration methods

placeholder["name"] = function
placeholder.add({"key": "value"}, name=function)

And then

text_content = TextContent(text="Hello, $name!")
photo_content = PhotoContent(photo=..., caption="Photo especially for $name!")

Key-based content

This module provides utilities to create personalized content targeted to specific users or groups based on their language preferences or geographical location, etc.

[!NOTE] If the default key is not specified, an error will be given if the key is not found.

Usage:

from aiogram.exceptions import TelegramBadRequest

from aiogram_broadcaster.contents import KeyBasedContent, TextContent


class LanguageBasedContent(KeyBasedContent):
    """Content based on the user's language."""

    async def __call__(self, chat_id: int, bot: Bot) -> Optional[str]:
        try:
            member = await bot.get_chat_member(chat_id=chat_id, user_id=chat_id)
        except TelegramBadRequest:
            return None
        else:
            return member.user.language_code


content = LanguageBasedContent(
    # default=TextContent(text="Hello!"),
    uk=TextContent(text="Привіт!"),
    ru=TextContent(text="Привет!"),
)

Lazy content

Allows content to be generated dynamically at the time the message is sent.

Usage:

from secrets import choice
from typing import List

from pydantic import SerializeAsAny

from aiogram_broadcaster.contents import BaseContent, LazyContent, TextContent


class RandomizedContent(LazyContent):
    contents: List[SerializeAsAny[BaseContent]]

    async def __call__(self) -> BaseContent:
        return choice(self.contents)


content = RandomizedContent(
    contents=[
        TextContent(text="Hello!"),
        TextContent(text="Hi!"),
    ],
)
await broadcaster.create_mailer(content=content, chats=...)

Dependency injection

It is used for comprehensive dependency management, used in event system, key-based/lazy content, placeholders and so on.

Usage:

• Main contextual data

from aiogram_broadcaster import Broadcaster

broadcaster = Broadcaster(key="value")

• Fetching the dispatcher contextual data

from aiogram import Dispatcher

from aiogram_broadcaster import Broadcaster

dispatcher = Dispatcher()
dispatcher["key"] = "value"

broadcaster = Broadcaster()
broadcaster.setup(dispatcher, fetch_dispatcher_context=True)

• Contextual data only for mailer

await broadcaster.create_mailer(content=..., chats=..., key=value)

• Stored contextual data only for mailer

await broadcaster.create_mailer(content=..., chats=..., stored_context={"key": "value"})

• Event-to-event

@event.completed()
async def transfer_content() -> Dict[str, Any]:
    return {"my_data": 1}


@event.completed()
async def mailer_completed(my_data: 1) -> None:
    print(my_data)

Storages

Storage allow you to save mailer states to external storage.

• BaseMailerStorage Abstract class of storage.

• FileMailerStorage Saves the mailers to a file.

• MongoDBMailerStorage Saves the mailers to a MongoDB.

• RedisMailerStorage Saves the mailers to a Redis.

• SQLAlchemyMailerStorage Saves the mailers using SQLAlchemy.

Usage:

from aiogram_broadcaster import Broadcaster
from aiogram_broadcaster.storages.redis import RedisMailerStorage

# from aiogram_broadcaster.storages.file import FileMailerStorage
# from aiogram_broadcaster.storages.mongodb import MongoDBMailerStorage
# from aiogram_broadcaster.storages.sqlalchemy import SQLAlchemyMailerStorage

# storages = FileMailerStorage()
# storages = MongoDBMailerStorage.from_url(url="mongodb://localhost:27017")
# storages = SQLAlchemyMailerStorage.from_url(url="sqlite+aiosqlite:///database.db")

storage = RedisMailerStorage.from_url(url="redis://localhost:6379")
broadcaster = Broadcaster(storage=storage)

Default mailer settings

The DefaultMailerSettings class defines the default properties for mailers created within the broadcaster. It allows setting various parameters like interval, dynamic_interval, run_on_startup, handle_retry_after, destroy_on_complete, and preserve. These properties provide flexibility and control over the behavior of mailers.

Parameters:

• interval: The interval (in seconds) between successive message broadcasts. It defaults to 0, indicating immediate broadcasting.

• dynamic_interval: A boolean flag indicating whether the interval should be adjusted dynamically based on the number of chats. If set to True, the interval will be divided equally among the chats.

• run_on_startup: A boolean flag indicating whether the mailer should start broadcasting messages automatically on bot startup.

• handle_retry_after: A boolean flag indicating whether the mailer should handle the TelegramAPIError error automatically.

• destroy_on_complete: A boolean flag indicating whether the mailer should be destroyed automatically upon completing its operations.

• preserve: A boolean flag indicating whether the mailer's state should be preserved even after completion. If set to True, the mailer's state will be stored in the specified storage.

Usage:

from aiogram_broadcaster import Broadcaster
from aiogram_broadcaster.mailer import DefaultMailerSettings

default = DefaultMailerSettings(
    interval=60_000,
    dynamic_interval=True,
    run_on_startup=True,
    handle_retry_after=True,
    destroy_on_complete=True,
    preserve=True,
)
broadcaster = Broadcaster(default=default)