This tutorial is designed for classroom use with students learning to build Telegram bots using aiogram 3.
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.
File Manager Bot with Aiogram 3 (by ExploitX)
A telegram bot implementation that shows basics of Aiogram 3 in action.
General Plan
This bot lets users upload, categorize, and retrieve files directly through Telegram messages. We aim to use as much aiogram features as we can in a practical way. Let’s go through the whole process step by step. The overall plan is as follows:
- User registration and management
- File upload with automatic categorization
- Browse files by category (Photos, Videos, Documents)
- Download previously uploaded files
- Clean inline keyboard interface
Implementation
Tech Stack
We use the following technologies to achieve our goal:
- Aiogram 3.x - Modern async Telegram bot framework
- PostgreSQL - Reliable database for file metadata
- SQLAlchemy - Async ORM for database operations
- Alembic - Database migrations
Project Structure
Here’s how I organized the project:
app/
├── __init__.py
├── database.py # Database connection
├── models.py # SQLAlchemy models
├── middleware.py # Database session middleware
└── routers/
├── start.py # /start command
├── menu.py # Navigation menus
├── upload.py # File upload logic
└── files.py # File retrieval
main.py # Bot entry point
requirements.txt # Dependencies
.env # Environment variables
Project Setup
This project uses PIP (Python package installer and resolver) for dependency management.
# Initialize new project
# Create virtual environment
python3 -m venv .venv
# Activate virtual environment
source .venv/bin/activate # Linux/Mac
# Install dependencies
pip install aiogram
Create a .env file with your credentials:
DB_NAME=your_database_name
DB_USER=postgres
DB_PASSWORD=your_password
DB_HOST=localhost
DB_PORT=5432
BOT_TOKEN=your_bot_token_from_botfather
Database Setup
As we are using aiogram which is asynchronous, we need an async postgres driver. So we used asyncpg over psycopg2. Here is the implementation:
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
import os
from dotenv import load_dotenv
load_dotenv()
DB_NAME = os.getenv("DB_NAME")
DB_USER = os.getenv("DB_USER")
DB_PASSWORD = os.getenv("DB_PASSWORD")
DB_HOST = os.getenv("DB_HOST")
DB_PORT = os.getenv("DB_PORT")
DATABASE_URL = f"postgresql+asyncpg://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
engine = create_async_engine(DATABASE_URL, echo=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase):
pass
async def get_session():
async with SessionLocal() as session:
yield session
Here, we have a asynchronous generator function get_session that yields a session from the database, which we'll use as a dependency. Even though in Aiogram we don't have dependency injection, we'll soon achieve the behaviour using Middlewares.
Database Models
We kept the models simple but effective. Two main entities: User and File with a one-to-many relationship:
from typing import List, Optional
from sqlalchemy import String, ForeignKey, BigInteger, Enum
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
import enum
class FileCategory(str, enum.Enum):
document = "Document"
image = "Image"
video = "Video"
other = "Other"
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, index=True, autoincrement=True)
telegram_id: Mapped[int] = mapped_column(BigInteger, unique=True, nullable=False, index=True)
username: Mapped[Optional[str]] = mapped_column(String, nullable=True)
fullname: Mapped[Optional[str]] = mapped_column(String, nullable=True)
phone: Mapped[Optional[str]] = mapped_column(String, nullable=True)
files: Mapped[List["File"]] = relationship(back_populates="user", cascade="all, delete-orphan")
class File(Base):
__tablename__ = "files"
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, index=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("users.id", ondelete="CASCADE"))
file_id: Mapped[str] = mapped_column(String, nullable=False)
file_name: Mapped[Optional[str]] = mapped_column(String, nullable=True)
category: Mapped[FileCategory] = mapped_column(Enum(FileCategory), default=FileCategory.other)
user: Mapped["User"] = relationship(back_populates="files")
Database Migrations with Alembic
We ran into an issue during migration setup. The error was:
The reason is we are using asyncpg driver for postgres, and that requires us to override the built-in alembic/env.py file. We should include a directive to show how to run async migrations:
def do_run_migrations(connection: Connection) -> None:
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
async def run_async_migrations() -> None:
connectable = async_engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await connectable.dispose()
Remember to install asyncpg: pip install asyncpg
Database Session Middleware
As we said, Aiogram does not natively support dependency injection, which FastAPI does. So, we learn from FastAPI and mimic that behaviour using middlewares:
from aiogram import BaseMiddleware
from typing import Callable, Dict, Any, Awaitable
from app.database import SessionLocal
class DbSessionMiddleware(BaseMiddleware):
async def __call__(
self,
handler: Callable[[Dict[str, Any]], Awaitable[Any]],
event: Any,
data: Dict[str, Any]
) -> Any:
session = SessionLocal()
data["db"] = session
try:
result = await handler(event, data)
finally:
await session.close()
return result
This middlewares ensures that in every request, our router will have an access (session) to the database and freely can work with it.
User Registration Handler
The start command handles user registration elegantly:
from aiogram import Router
from aiogram.types import Message
from aiogram.filters import CommandStart
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models import User
from app.routers.menu import main_menu
router = Router()
@router.message(CommandStart())
async def cmd_start(message: Message, db: AsyncSession):
user = message.from_user
result = await db.execute(
select(User).where(User.telegram_id == user.id)
)
existing_user = result.scalar_one_or_none()
if not existing_user:
new_user = User(
telegram_id=user.id,
username=user.username,
fullname=user.full_name,
)
db.add(new_user)
await db.commit()
await message.answer(
f"Hey! What's up, {user.full_name or user.username}!\nChoose from the menu below:",
reply_markup=main_menu()
)
Navigation Menus
I created reusable keyboard functions for clean navigation:
from aiogram import Router, F
from aiogram.types import CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton
router = Router()
def main_menu() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[
[InlineKeyboardButton(text="📤 Upload File", callback_data="upload_file")],
[InlineKeyboardButton(text="📂 My Files", callback_data="my_files")],
]
)
def category_menu() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(
inline_keyboard=[
[InlineKeyboardButton(text="🖼 Photos", callback_data="category_image")],
[InlineKeyboardButton(text="🎥 Videos", callback_data="category_video")],
[InlineKeyboardButton(text="📄 Documents", callback_data="category_document")],
[InlineKeyboardButton(text="📂 All Files", callback_data="category_all")],
]
)
@router.callback_query(F.data == "my_files")
async def show_category_menu(callback: CallbackQuery):
await callback.message.answer("📂 Choose the category", reply_markup=category_menu())
await callback.answer()
File Upload Logic
The upload handler uses FSM (Finite State Machine) for step-by-step file uploading:
from aiogram import Router, F
from aiogram.types import CallbackQuery, Message, InlineKeyboardMarkup, InlineKeyboardButton
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
from app.models import File, FileCategory, User
from sqlalchemy import select
router = Router()
class UploadStates(StatesGroup):
waiting_for_category = State()
waiting_for_file = State()
@router.callback_query(F.data == "upload_file")
async def upload_file_request(callback: CallbackQuery, state: FSMContext):
kb = InlineKeyboardMarkup(
inline_keyboard=[
[InlineKeyboardButton(text="📷 Photos", callback_data="choose_category:image")],
[InlineKeyboardButton(text="📄 Documents", callback_data="choose_category:document")],
[InlineKeyboardButton(text="🎥 Videos", callback_data="choose_category:video")],
]
)
await callback.message.answer("📂 Choose the category", reply_markup=kb)
await state.set_state(UploadStates.waiting_for_category)
await callback.answer()
@router.message(F.document | F.photo | F.video)
async def receive_file(message: Message, state: FSMContext, db):
# Smart file type detection
if message.photo:
file_category = FileCategory.image
file_id = message.photo[-1].file_id
filename = f"Photo_{file_id}.jpg"
elif message.video:
file_category = FileCategory.video
file_id = message.video.file_id
file_name = message.video.file_name or f"Video_{file_id}.mp4"
elif message.document:
if message.document.mime_type.startswith("image/"):
file_category = FileCategory.image
elif message.document.mime_type.startswith("video/"):
file_category = FileCategory.video
else:
file_category = FileCategory.document
file_id = message.document.file_id
file_name = message.document.file_name
# Save to database
user_id = message.from_user.id
result = await db.execute(select(User).where(User.telegram_id == user_id))
user = result.scalars().first()
new_file = File(
user_id=user.id,
file_id=file_id,
file_name=file_name,
category=file_category
)
db.add(new_file)
await db.commit()
await state.clear()
await message.answer(f"✅ File saved successfully ({file_category.name})")
File Retrieval System
Users can browse and download their files by category:
from aiogram import Router, F
from aiogram.types import CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton
from app.models import File, FileCategory, User
from sqlalchemy import select
router = Router()
@router.callback_query(F.data.startswith("category_"))
async def list_files_by_category(callback: CallbackQuery, db):
user_id = callback.from_user.id
cat = callback.data.split("_")[1]
query = select(File).join(User).where(User.telegram_id == user_id)
if cat != "all":
query = query.where(File.category == getattr(FileCategory, cat))
result = await db.execute(query)
files = result.scalars().all()
if not files:
await callback.message.answer("❌ No files found in this category.")
else:
kb = InlineKeyboardMarkup(
inline_keyboard=[
[InlineKeyboardButton(text=f.file_name, callback_data=f"get_file:{f.id}")]
for f in files
]
)
await callback.message.answer(f"📂 {cat.capitalize()} files:", reply_markup=kb)
await callback.answer()
@router.callback_query(F.data.startswith("get_file:"))
async def send_file_handler(callback: CallbackQuery, db):
file_id = int(callback.data.split(":")[1])
file = await db.get(File, file_id)
if not file:
await callback.message.answer("❌ File not found.")
return
# Send appropriate file type
if file.category == FileCategory.image:
await callback.message.answer_photo(file.file_id, caption=file.file_name)
elif file.category == FileCategory.document:
await callback.message.answer_document(file.file_id, caption=file.file_name)
elif file.category == FileCategory.video:
await callback.message.answer_video(file.file_id, caption=file.file_name)
await callback.answer()
Main Application
Finally, everything comes together in the main file:
import asyncio
import logging
import os
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from app.routers import start, menu, files, upload
from app.middleware import DbSessionMiddleware
from dotenv import load_dotenv
load_dotenv()
BOT_TOKEN = os.getenv("BOT_TOKEN")
logging.basicConfig(level=logging.INFO)
bot = Bot(
token=BOT_TOKEN,
default=DefaultBotProperties(parse_mode="HTML")
)
dp = Dispatcher()
# Register middleware
dp.message.middleware(DbSessionMiddleware())
dp.callback_query.middleware(DbSessionMiddleware())
# Register routers
dp.include_router(start.router)
dp.include_router(menu.router)
dp.include_router(upload.router)
dp.include_router(files.router)
async def main():
logging.info("Bot is running successfully!")
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())
What We Learned
Building this bot taught us several valuable lessons:
- Aiogram 3.x is much cleaner than version 2.x with better async support. The framework is going towards the FastAPI philosophy.
- FSM states make complex user interactions manageable.
- SQLAlchemy async requires careful session management.
- File type detection through MIME types is more reliable than extensions.
- Middleware is perfect for cross-cutting concerns like database sessions.
Potential Improvements
If we were to extend this bot, we'd add: - File size limits and validation - User storage quotas - File sharing between users - Search functionality
Running the Project
- Clone the repository
- Install dependencies:
pip install -r requirements.txt - Set up PostgreSQL database
- Configure
.envfile - Run migrations:
alembic upgrade head - Start the bot:
python main.py
That's it! You now have a fully functional file manager bot. The code is clean, scalable, and ready for test! Here is the working repository: GitHub
Happy coding!