aiogram-broadcast 0.2.0

Broadcast/newsletter library for aiogram 3.x

aiogram-broadcast

Broadcast/newsletter library for Telegram bots built with aiogram 3.x.

Features

• Automatic subscriber registration via middleware
• Rate-limited broadcasting to avoid Telegram API limits
• Scheduled broadcasts with APScheduler
• Pluggable subscriber storage: Redis or PostgreSQL
• Progress callbacks for monitoring
• Interactive UI menu with multi-language support (EN/RU)

Installation

Pick a storage backend (the core package no longer pulls in a driver by default):

pip install aiogram-broadcast[redis]      # Redis storage
pip install aiogram-broadcast[postgres]   # PostgreSQL storage (asyncpg)

Everything (both backends, scheduler and UI):

pip install aiogram-broadcast[all]

Quick Start

from aiogram_broadcast import BroadcastMiddleware, BroadcastService, RedisBroadcastStorage

redis = Redis(host="localhost")
storage = RedisBroadcastStorage(redis)
service = BroadcastService(bot, storage)

# Auto-register subscribers
dp.update.outer_middleware.register(BroadcastMiddleware(storage))

# Broadcast to all
result = await service.broadcast_text("Hello everyone!", parse_mode="HTML")
print(f"{result.successful}/{result.total} sent, {result.success_rate:.0f}%")

See examples/basic.py for a complete runnable bot.

Storage

Both backends implement the same BaseBroadcastStorage interface, so the service, middleware, scheduler and UI work identically regardless of which one you pick.

Redis

from redis.asyncio import Redis
from aiogram_broadcast import RedisBroadcastStorage

redis = Redis(host="localhost")
storage = RedisBroadcastStorage(redis)

PostgreSQL

import asyncpg
from aiogram_broadcast import PostgresBroadcastStorage

pool = await asyncpg.create_pool("postgresql://user:pass@localhost:5432/db")
storage = PostgresBroadcastStorage(pool)
await storage.create_schema()  # create the table and index once on startup

# or build the pool for you from a DSN:
storage = await PostgresBroadcastStorage.from_dsn("postgresql://user:pass@localhost/db")
await storage.create_schema()

See examples/postgres.py for a complete runnable bot.

Usage

Broadcasting

# Text, photo, video, document, or copy any message
await service.broadcast_text("Hello!", parse_mode="HTML")
await service.broadcast_photo(photo_file_id, caption="Check this out")
await service.broadcast_copy(from_chat_id=admin_id, message_id=msg_id)

# Custom sender for any message type
await service.broadcast_custom(my_sender_func)

Scheduled Broadcasts

from aiogram_broadcast import BroadcastScheduler

scheduler = BroadcastScheduler(service, AsyncIOScheduler())
task_id = await scheduler.schedule_text("Reminder!", run_date=some_datetime)
await scheduler.cancel(task_id)

See examples/scheduled.py for full setup.

Progress Tracking

async def on_progress(current, total, result):
    print(f"{current}/{total} — {result.successful} sent, {result.failed} failed")

await service.broadcast_text("msg", progress_callback=on_progress)

See examples/progress.py.

Interactive UI Menu

A complete FSM-based wizard for composing and scheduling broadcasts from Telegram — supports text/photo/video/document messages, inline URL buttons, and scheduled delivery.

See examples/ui_menu.py.

License

MIT