This tutorial is designed for classroom use with students learning to build Telegram bots using aiogram 3.
Aiogram 3.x — Developer's Guide
This tutorial is designed for classroom use with students learning to build Telegram bots using aiogram 3.x. Each step introduces a key feature with explanation, examples, and interactive exercises.
Step 0 — Setup & Tiny Echo Bot
Goal: Get everyone a working bot.
Example:
import asyncio
from aiogram import Bot, Dispatcher
from aiogram.filters import Command
from aiogram.types import Message
BOT_TOKEN = "YOUR_TOKEN"
async def main():
bot = Bot(BOT_TOKEN)
dp = Dispatcher()
@dp.message(Command("start"))
async def start(m: Message):
await m.answer("Hello! Send me anything, I’ll echo it.")
@dp.message()
async def echo(m: Message):
await m.answer(m.text or "Got your message!")
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
Exercise: Run the bot and confirm it echoes text.
Step 1 — Dispatcher, Routers, Modular Handlers
Why? Organize large projects by splitting features.
Example:
# routers/start.py
from aiogram import Router
from aiogram.filters import CommandStart
from aiogram.types import Message
router = Router()
@router.message(CommandStart())
async def start(m: Message):
await m.answer("Welcome!")
# main.py
from aiogram import Bot, Dispatcher
from routers import start
bot = Bot("TOKEN")
dp = Dispatcher()
dp.include_router(start.router)
await dp.start_polling(bot)
Exercise: Split the echo bot into routers/echo.py
.
Step 2 — Filters & F
Expressions
Why? Catch exactly the messages you want.
Example:
from aiogram import F
@router.message(F.photo | F.document)
async def media_handler(m: Message):
await m.answer("Nice media!")
Exercise: Add handler for text longer than 10 chars.
Step 3 — Keyboards (Reply & Inline)
Reply keyboards insert text; Inline keyboards attach actions.
Example:
from aiogram.utils.keyboard import InlineKeyboardBuilder
builder = InlineKeyboardBuilder()
builder.button(text="Work", callback_data="cat:work")
builder.button(text="Personal", callback_data="cat:personal")
builder.adjust(2)
await message.answer("Choose: ", reply_markup=builder.as_markup())
Exercise: Build a 3×2 inline keyboard.
Step 4 — CallbackData
Why? Safer structured callback payloads.
Example:
from aiogram.filters.callback_data import CallbackData
class FileCb(CallbackData, prefix="file"):
action: str
file_id: int
@router.callback_query(FileCb.filter())
async def on_file_action(cb: CallbackQuery, callback_data: FileCb):
await cb.answer(f"Action {callback_data.action} on {callback_data.file_id}")
Exercise: Create Delete/Rename buttons using FileCb
.
Step 5 — FSM (Finite State Machine)
Why? Handle multi-step flows.
Example:
from aiogram.fsm.state import State, StatesGroup
from aiogram.filters import StateFilter
class Upload(StatesGroup):
waiting_for_file = State()
waiting_for_category = State()
@router.message(Command("upload"))
async def start_upload(m: Message, state: FSMContext):
await m.answer("Send me a file.")
await state.set_state(Upload.waiting_for_file)
@router.message(StateFilter(Upload.waiting_for_file))
async def got_file(m: Message, state: FSMContext):
await state.update_data(file_id=m.document.file_id)
await m.answer("Choose category.")
await state.set_state(Upload.waiting_for_category)
Exercise: Create /survey
that asks two questions.
Step 6 — Middlewares & Dependency Injection
Why? Inject DB sessions, configs, etc.
Example middleware:
from aiogram import BaseMiddleware
class DBSessionMiddleware(BaseMiddleware):
async def __call__(self, handler, event, data):
async with async_sessionmaker() as session:
data["session"] = session
return await handler(event, data)
Exercise: Middleware that injects request_id
into handlers.
Step 7 — File Handling (Telegram specifics)
- Telegram gives file_id: store & resend.
- To get best photo:
m.photo[-1].file_id
- To send back:
await m.answer_document(file_id)
Exercise: Receive a file, store file_id
in memory, resend with /get
.
Step 8 — Async DB Integration
Pattern (SQLAlchemy async):
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
engine = create_async_engine("sqlite+aiosqlite:///db.sqlite3")
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
Use middleware to inject session: AsyncSession
into handlers.
Exercise: Create users
table, save new user on /start
.
Step 9 — Deployment: Polling → Webhooks
- Polling: good for dev.
- Webhooks: required for production, faster & efficient.
Pattern:
Exercise: Demo FastAPI app forwarding updates to dp.feed_update(...)
.
Step 10 — Testing & Pitfalls
Tips:
- Always
await session.commit()
. - Always
await state.clear()
when FSM flow finishes. - Debug callback data by logging
callback_query.data
.
Homework: Build mini file manager bot: upload, list, delete.
Final Project Milestones
- Midterm: Upload flow storing file_id + category in SQLite.
- Final: File Keeper Bot with CRUD (rename, delete, paginate) and webhook deployment.