Полный гайд по разработке бота на Aiogram

kirilljs

Так ну что ж, давайте без какого либо вступления сразу в бой
Будем создавать простого бота на Aiogram, а также хранить пользователей в Sqlite для сбора id.

Нам понадобятся следующие библиотеки:

aiogram (документация тут - https://aiogram.dev)
dotenv
os
aiosqlite
asyncio

И так архитектура у нас будет такая:
bot.py
config.py
commands.py
sqlite_db.py
.env

Начнем с файла bot.py:

Ах, да не забудьте установить все библиотеки через pip install aiogram dotenv aiosqlite asyncio, os - уже встроен в python.

Если кому в падлу читать тут, я выложил готовый код на GitHub:

GitHub - KirillEvo/defaultbot: Default telegram bot + db Sqlite

Default telegram bot + db Sqlite. Contribute to KirillEvo/defaultbot development by creating an account on GitHub.

GitHub (github.com)

Файл: bot.py

import asyncio
import logging

from aiogram import Bot, Dispatcher, Router, types
from aiogram.filters import Command

from sqlite_db import SqliteUserDataManager

# Создаем экземпляр базы данных
data = SqliteUserDataManager()

# Тут импортируем все команды, нам понадобится только старт

from commands import (
    start_command,
    allusers_command
)

# Также добавим логирование что бы видеть ошибки
# Настройка логирования
logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO
)
logger = logging.getLogger(__name__)

# Импорт конфигурации
from config import TOKEN

# Создаем основной роутер
main_router = Router()


# Регистрируем роутер для вызова команды /start
@main_router.message(Command("start"))
async def handle_start(message: types.Message):
    await start_command(message, data)

@main_router.message(Command("users"))
async def handle_users(message: types.Message):
    await allusers_command(message, data)


async def main():
    # Инициализация бота
    global bot
    bot = Bot(TOKEN)
    dp = Dispatcher()

    # Регистрируем роутер
    dp.include_router(main_router)

    # Запуск бота
    logger.info("🤖 Бот запущен")
    await bot.delete_webhook(drop_pending_updates=True)
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

Я оставил комментарии в коде, надеюсь всем будет понятно что куда и к чему.

После того как мы заполним bot.py приступаем к написанию кода для хранения базы данных:
Файл: sqlite_db.py

import asyncio
import aiosqlite
from pathlib import Path

# Создаем класс нашей БД с методами
class SqliteUserDataManager:
    def __init__(self, db_file="users.db"):
        self.db_file = Path(db_file)
        self.lock = asyncio.Lock()
        self.initialized = False


    # Этот код представляет собой асинхронный метод _ensure_db_exists, который проверяет и при необходимости создает базу данных SQLite с таблицей users.
    async def _ensure_db_exists(self):
        if self.initialized:
            return
            
        # Использование асинхронного блокировщика для предотвращения race condition (состояния гонки), когда несколько потоков/корутин могут попытаться инициализировать базу одновременно
        async with self.lock:
            if self.initialized:  # Дважды проверьте блокировку
                return
            
            # Подключение к SQLite базе данных с использованием асинхронной библиотеки aiosqlite.
            async with aiosqlite.connect(self.db_file) as db:
                await db.execute('''
                    CREATE TABLE IF NOT EXISTS users (
                        user_id INTEGER PRIMARY KEY,
                        username TEXT,
                        subscribed BOOLEAN DEFAULT 1,
                        created_at REAL
                    )
                ''')
                await db.commit()
            self.initialized = True
        
    # Этот код представляет собой асинхронный метод add_user, который добавляет пользователя в базу данных SQLite.
    async def add_user(self, user_id, username=None):
        await self._ensure_db_exists()
        async with self.lock:
            async with aiosqlite.connect(self.db_file) as db:
                await db.execute(
                    '''
                    INSERT OR IGNORE INTO users (user_id, username, created_at)
                    VALUES (?, ?, ?)
                    ''',
                    (user_id, username, asyncio.get_event_loop().time())
                )
                # Обновляем имя пользователя, если оно изменилось
                if username is not None:
                    await db.execute(
                        "UPDATE users SET username = ? WHERE user_id = ?",
                        (username, user_id)
                    )
                await db.commit()
    
    # Этот код представляет собой асинхронный метод get_user, который получает информацию о пользователе из базы данных SQLite по его ID.
    async def get_all_users(self):
        await self._ensure_db_exists()
        async with self.lock:
            async with aiosqlite.connect(self.db_file) as db:
                cursor = await db.execute("SELECT * FROM users")
                rows = await cursor.fetchall()
                return {
                    row[0]: {
                        "username": row[1],
                        "subscribed": bool(row[2]),
                        "created_at": row[3]
                    } for row in rows
                }

После написание методов для Sqlite переходим к commands.py:

# Импортируем настройки ID админа из нашего файла config.py
from config import ADMIN_ID

# Простая команда которая отдает приветствие при выполнении /start в самом боте
async def start_command(message, data):
    user = message.from_user
    await data.add_user(user.id, user.username)
    await message.answer("Прииииивеееееет!",)

# Команда для получения списка всех пользователей, но только для админа
async def allusers_command(message, data):
    user_id = message.from_user.id
    admin_id = int(ADMIN_ID)

    if user_id != admin_id:
        await message.answer("❌ У вас нет прав для выполнения этой команды.")
        return

    try:
        users = await data.get_all_users()
        count = len(users)
        await message.answer(f"Количество пользователей: {count}")
    except Exception as e:
        await message.answer("Ошибка при получении списка пользователей.")
        print(f"[ERROR] {e}")

Ну и конечно же не забываем о файле: config.py

import os
from dotenv import load_dotenv
load_dotenv()

TOKEN = os.getenv("TOKEN")
ADMIN_ID = os.getenv("ADMIN_ID")

На этом все! Запускайте своих ботов и тестируйте.
Если будут вопросы пишите в комментариях на форуме.

Кстати не забудь подписаться на мой ТГ канал и еще оставлю ссылку на своего бота для скачивания клипов из ТикТока

Kirilljs

Обо всем и не о чем сразу 👨🏻‍💻

Telegram (t.me)

Сам бот:

Скачать видео ВК Клипы | ТикТок

🤖 Умею скачивать видео из VK Clips, TikTok, Yappy.

Telegram (telegram.me)