tgbotcore 1.0.2

Core library for Telegram bots

tgbotcore

Базовое ядро для Telegram-ботов на aiogram 3.x. Устанавливается как pip-пакет, расширяется на уровне шаблона — ядро никогда не трогается.

Содержание

• Требования
• Установка
• Быстрый старт
• Архитектура
• Компоненты
  • Settings
  • Database
  • Models
  • Middleware
  • Keyboards
  • Admin
• Создание шаблона
• Правила расширения
• Версионирование

---

Требования

• Python 3.11+
• PostgreSQL 14+ или SQLite 3.x

---

Установка

Последняя версия:

pip install tgbotcore

Локальная разработка ядра — изменения применяются сразу без переустановки:

pip install -e /path/to/tgbotcore

---

Быстрый старт

Минимальный рабочий бот поверх ядра:

my-bot/
├── alembic/
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
├── db/
│   └── models.py
├── handlers/
│   └── start.py
├── alembic.ini
├── config.py
├── main.py
├── requirements.txt
└── .env

requirements.txt:

tgbotcore
aiogram>=3.7.0
psycopg2-binary

alembic.ini:

[alembic]
script_location = alembic

alembic/env.py:

import db.models
from config import settings
from tgbotcore.alembic_env import run
run(settings.DATABASE_URL)

alembic/script.py.mako:

"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}

"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}

revision: str = ${repr(up_revision)}
down_revision: Union[str, None] = ${repr(down_revision)}
branch_labels: Union[str, Sequence[str], None] = ${repr(branch_labels)}
depends_on: Union[str, Sequence[str], None] = ${repr(depends_on)}

def upgrade() -> None:
    ${upgrades if upgrades else "pass"}

def downgrade() -> None:
    ${downgrades if downgrades else "pass"}

config.py:

from tgbotcore import Settings

class BotSettings(Settings):
    pass

settings = BotSettings()

db/models.py:

from tgbotcore import Base, UserMixin

class User(UserMixin, Base):
    pass

handlers/start.py:

from aiogram import Router
from aiogram.filters import CommandStart
from aiogram.types import Message

router = Router()

@router.message(CommandStart())
async def cmd_start(message: Message, user) -> None:
    await message.answer(f"Привет, {user.full_name}!")

main.py:

import asyncio
import logging
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.fsm.storage.memory import MemoryStorage
from tgbotcore import (
    AntiSpamMiddleware, UserMiddleware,
    create_admin_router, init_db, run_migrations,
)
from config import settings
from db.models import User
from handlers.start import router as start_router

logging.basicConfig(level=logging.INFO)

async def main() -> None:
    bot = Bot(
        token=settings.BOT_TOKEN,
        default=DefaultBotProperties(parse_mode=ParseMode.HTML),
    )
    dp = Dispatcher(storage=MemoryStorage())

    dp.update.middleware(AntiSpamMiddleware(
        limit=settings.RATE_LIMIT,
        window=settings.RATE_LIMIT_WINDOW,
    ))
    dp.update.middleware(UserMiddleware(
        user_model=User,
        admin_ids=settings.ADMIN_IDS,
    ))

    dp.include_router(create_admin_router(user_model=User))
    dp.include_router(start_router)

    run_migrations()
    await init_db(
        database_url=settings.DATABASE_URL,
        create_tables=False,
        user_model=User,
        admin_ids=settings.ADMIN_IDS,
    )

    try:
        await dp.start_polling(
            bot,
            allowed_updates=dp.resolve_used_update_types(),
        )
    finally:
        await bot.session.close()

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

.env:

BOT_TOKEN=123456:ABC...
ADMIN_IDS=123456789
DATABASE_URL=sqlite+aiosqlite:///bot.db
DEBUG=true

Запуск:

pip install -r requirements.txt
python main.py

---

Архитектура

Принцип односторонней зависимости

ядро → шаблон:     ЗАПРЕЩЕНО   ядро ничего не знает о шаблонах
шаблон → ядро:     РАЗРЕШЕНО   шаблон импортирует из ядра
шаблон → шаблон:   ЗАПРЕЩЕНО   шаблоны независимы друг от друга

Публичный API

from tgbotcore import (
    # database
    Base, get_session, get_session_factory,
    init_db, ensure_admins, run_migrations,
    # models
    UserMixin, TimestampMixin,
    # config
    Settings,
    # middleware
    AntiSpamMiddleware, UserMiddleware,
    # keyboards
    paginate, confirm_cancel, back_button, main_menu, url_button,
    # admin
    create_admin_router,
)

Разделение ответственности

Фаза         Инструмент              Ответственность
---------    ----------------------  ----------------------------------
deploy       run_migrations()        создание и обновление схемы БД
runtime      init_db()               инициализация соединения с БД
runtime      UserMiddleware          авторизация и инжект пользователя
runtime      AntiSpamMiddleware      защита от спама
runtime      handler'ы шаблона      бизнес-логика

---

Компоненты

Settings

Базовый класс конфигурации на pydantic-settings. Читает переменные из .env файла и окружения.

Базовые поля:

Поле               Тип        Дефолт                      Описание
BOT_TOKEN          str        -                           Токен бота от @BotFather
ADMIN_IDS          Any        []                          Telegram ID администраторов
DATABASE_URL       str        sqlite+aiosqlite:///bot.db  URL подключения к БД
RATE_LIMIT         int        30                          Максимум сообщений за окно
RATE_LIMIT_WINDOW  int        60                          Размер окна в секундах
DEBUG              bool       False                       Режим отладки
TIMEZONE           str        Europe/Moscow               Временная зона
REFERRAL_BONUS     int        10                          Бонус за реферала

Расширение:

from tgbotcore import Settings

class MySettings(Settings):
    OPENAI_API_KEY: str
    MAX_HISTORY: int = 20
    FEATURE_FLAG: bool = False

settings = MySettings()

Формат ADMIN_IDS в .env — обв варианта валидны:

ADMIN_IDS=123456789
ADMIN_IDS=123456789,987654321

---

Database

Асинхронный слой работы с БД на SQLAlchemy 2.x.

run_migrations():

Запускает alembic upgrade head при старте бота. Если папка versions/ пустая — автоматически генерирует начальную миграцию. Вызывать до init_db().

# main.py — обязательный порядок
run_migrations()
await init_db(database_url=settings.DATABASE_URL, ...)

init_db(database_url, create_tables, user_model, admin_ids):

# продакшн — схема управляется через alembic
await init_db(
    database_url=settings.DATABASE_URL,
    create_tables=False,
    user_model=User,
    admin_ids=settings.ADMIN_IDS,
)

# разработка — создаёт таблицы автоматически без alembic
await init_db(
    database_url=settings.DATABASE_URL,
    create_tables=True,
)

ensure_admins(user_model, admin_ids):

Назначает is_admin=True пользователям из ADMIN_IDS если они уже есть в БД. Вызывается автоматически внутри init_db() если переданы user_model и admin_ids.

get_session():

# автоматически через middleware (рекомендуется)
async def my_handler(message: Message, session: AsyncSession) -> None:
    result = await session.execute(select(User))

# вручную — только если нет доступа к middleware
from tgbotcore import get_session

async def some_service() -> None:
    async for session in get_session():
        result = await session.execute(select(User))

Поддерживаемые СУБД:

# SQLite — разработка
DATABASE_URL=sqlite+aiosqlite:///bot.db

# PostgreSQL — продакшн
DATABASE_URL=postgresql+asyncpg://user:password@localhost:5432/dbname

Alembic

Для работы run_migrations() шаблон должен содержать:

alembic/
├── env.py          — 3 строки, вызывает tgbotcore.alembic_env.run()
├── script.py.mako  — шаблон файлов миграций
└── versions/       — файлы миграций (может быть пустой при первом запуске)
alembic.ini         — минимальный, только script_location

alembic/env.py:

import db.models                        # регистрирует модели в Base.metadata
from config import settings
from tgbotcore.alembic_env import run
run(settings.DATABASE_URL)

alembic.ini:

[alembic]
script_location = alembic

Ручное управление миграциями:

alembic revision --autogenerate -m "add payments table"
alembic upgrade head
alembic downgrade -1

---

Models

UserMixin — базовые поля пользователя:

Поле         Тип           Описание
id           int           Первичный ключ
telegram_id  int           Уникальный Telegram ID
username     str | None    @username
full_name    str           Имя и фамилия
is_banned    bool          Заблокирован
is_admin     bool          Администратор
last_active  datetime      Последняя активность
referred_by  int | None    Telegram ID реферера
created_at   datetime      Дата регистрации
updated_at   datetime      Дата обновления

TimestampMixin — добавляет created_at и updated_at к любой модели.

Создание модели User:

from tgbotcore import Base, UserMixin

# минимально
class User(UserMixin, Base):
    pass

# с расширением
class User(UserMixin, Base):
    phone: Mapped[str | None]
    balance: Mapped[int] = mapped_column(default=0)

Важно — никогда не создавай второй Base:

# правильно
from tgbotcore import Base
class Product(Base): ...

# неправильно — сломает alembic и create_all
from sqlalchemy.orm import DeclarativeBase
class MyBase(DeclarativeBase): pass

---

Middleware

AntiSpamMiddleware(limit, window):

dp.update.middleware(AntiSpamMiddleware(limit=30, window=60))

UserMiddleware(user_model, admin_ids):

dp.update.middleware(UserMiddleware(
    user_model=User,
    admin_ids=settings.ADMIN_IDS,
))

Поведение:

• Создаёт пользователя при первом обращении

• При создании назначает is_admin=True если telegram_id есть в admin_ids

• Инжектит data["user"], data["session"], data["is_new_user"] в каждый handler

• Обрабатывает race condition при одновременных запросах через savepoint

  handler получает пользователя автоматически

  async def my_handler( message: Message, user: User, session: AsyncSession, is_new_user: bool, ) -> None: if user.is_banned: await message.answer("Вы заблокированы.")

Реферальная система — используй is_new_user и command.args:

# handlers/start.py
@router.message(CommandStart())
async def cmd_start(message, command, user, session, is_new_user):
    if is_new_user and command.args and command.args.isdigit():
        referrer_id = int(command.args)
        # логика начисления бонуса рефереру

Порядок регистрации важен:

# правильно — AntiSpam первым
dp.update.middleware(AntiSpamMiddleware(...))
dp.update.middleware(UserMiddleware(...))

# неправильно — запрос в БД для каждого спамера
dp.update.middleware(UserMiddleware(...))
dp.update.middleware(AntiSpamMiddleware(...))

---

Keyboards

paginate(items, page, callback_prefix, page_size=8):

products = [("Товар 1", "product:1"), ("Товар 2", "product:2")]
kb = paginate(products, page=0, callback_prefix="catalog")
await message.answer("Каталог:", reply_markup=kb)

confirm_cancel(...):

kb = confirm_cancel(
    confirm_text="✅ Оформить заказ",
    cancel_text="❌ Отмена",
    confirm_callback="checkout:confirm",
    cancel_callback="checkout:cancel",
)

back_button(callback, text="← Назад"):

kb = back_button(callback="catalog:page:0")

main_menu(buttons, resize=True, one_time=False):

kb = main_menu(["🛒 Каталог", "🛍 Корзина", "📦 Заказы"])

url_button(text, url):

kb = url_button("🌐 Открыть сайт", "https://example.com")

---

Admin

create_admin_router(user_model, extra_routers, stats_callback):

Базовые команды:

Команда       Описание
/stats        Статистика + вызывает stats_callback
/broadcast    Рассылка всем пользователям
/ban <id>     Заблокировать пользователя
/unban <id>   Разблокировать пользователя
/user <id>    Информация о пользователе

Минимальное подключение:

dp.include_router(create_admin_router(user_model=User))

Расширение статистики:

async def get_shop_stats() -> str:
    return f"📦 Заказов сегодня: <b>{await count_orders_today()}</b>"

dp.include_router(create_admin_router(
    user_model=User,
    stats_callback=get_shop_stats,
))

Добавление своих команд:

# my-bot/admin.py
router = Router()

@router.message(Command("orders"))
async def cmd_orders(message: Message, user, session) -> None:
    if not user.is_admin:
        return
    ...

# main.py
dp.include_router(create_admin_router(
    user_model=User,
    extra_routers=[router],
))

---

Создание шаблона

Структура проекта:

my-template/
├── alembic/
│   ├── env.py
│   ├── script.py.mako
│   └── versions/
├── db/
│   ├── __init__.py
│   └── models.py
├── handlers/
│   ├── __init__.py
│   └── start.py
├── admin.py
├── config.py
├── main.py
├── alembic.ini
├── requirements.txt
├── .env.example
└── .gitignore

Зависимости в requirements.txt:

tgbotcore
psycopg2-binary

---

Правила расширения

Можно:

# наследовать Settings
class MySettings(Settings): ...

# наследовать UserMixin
class User(UserMixin, Base): ...

# наследовать TimestampMixin
class MyModel(TimestampMixin, Base): ...

# передавать параметры в фабрики
create_admin_router(user_model=User, extra_routers=[my_router])

# создавать свои клавиатуры используя функции ядра
from tgbotcore import back_button

Нельзя:

# импортировать из внутренних модулей
from tgbotcore._session import ...

# создавать второй Base
from sqlalchemy.orm import DeclarativeBase
class MyBase(DeclarativeBase): pass

# вызывать init_db() больше одного раза

# менять порядок middleware (AntiSpam должен быть первым)

# вызывать init_db() до run_migrations()

---

Версионирование

Проект следует Semantic Versioning (https://semver.org):

MAJOR — breaking changes в публичном API
MINOR — новые возможности, обратно совместимые
PATCH — исправления багов

Рабочий процесс обновления ядра:

# 1. внёс изменения в код
# 2. описал в CHANGELOG.md под [Unreleased]
# 3. обновил версию
bump-my-version bump minor

# 4. пуш — PyPI обновляется автоматически через CI
git push origin main --tags

# 5. шаблоны обновляют зависимость
pip install --upgrade tgbotcore