- Add python-development-database skill: metaorm patterns for tables, repositories, filters, settings, transactions, pagination, and exceptions - Add python-development-telegram-bot skill: aiogram bot service with polling/webhook, router structure, middlewares, inline keyboards, and pagination - Add python-development-tests skill: test organization guidelines with pytest, fixtures, and dependency injection rules - Update python-development-stack: replace SQLModel/SQLAlchemy with metaorm; add pytest, pytest-asyncio, pytest-cov, faker - Update python-development-fastapi: add sections for dependencies and list endpoints with ListResponse schema - Update python-development-code-style: allow ctx as exception for typer.Context parameter - Update python-development index with links to all three new skills Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
13 KiB
13 KiB
name, description
| name | description |
|---|---|
| python-development-telegram-bot | Patterns for building a Telegram bot service using aiogram inside the facet service tree. Use when creating bot handlers, routers, callbacks, inline keyboards, pagination, or the telegram_bot service package. |
Rules
- The Telegram bot service lives in a sub-package named
telegram_bot/inside the root package. It is a regular facet service: it hasservice.py,settings.py, and__init__.py. - Use
aiogramas the Telegram Bot API framework. - The settings class inherits from
pydantic.BaseModel(notBaseSettings). It declares the minimum required:method(polling or webhook) andtoken. - Provide separate settings subclasses for each method:
TelegramBotPollingSettings— addstimeout.TelegramBotWebhookSettings— addsroot_url,root_path,port,secret_access_key.
- The base service class inherits from
facet.AsyncioServiceMixin. Its constructor accepts the settings instance declared in the same package'ssettings.py. Keep the constructor minimal; heavy initialization happens instart(). - The base service creates
aiogram.Botandaiogram.Dispatcherin__init__orstart(). - Extract router registration into a
get_router()function (orsetup_dispatcher()when the dispatcher itself needs external dependencies). Handlers live inhandlers.py, router wiring lives inrouter.py. - Register the router in the service using
dispatcher.include_router(get_router()). - Use
aiogram.fsm.state.StatesGroupfor multi-step conversations. - Use
aiogram.filters.callback_data.CallbackDatafor typed callback data on inline buttons. Every callback action gets its ownCallbackDatasubclass with a uniqueprefix.
Minimal complete example
telegram_bot/settings.py:
import enum
from pydantic import BaseModel
class TelegramBotMethodEnum(str, enum.Enum):
POLLING = "polling"
WEBHOOK = "webhook"
class TelegramBotSettings(BaseModel):
method: TelegramBotMethodEnum = TelegramBotMethodEnum.POLLING
token: str = "123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11"
telegram_bot/polling/settings.py:
from typing import Literal
from pydantic import PositiveInt
from project.telegram_bot.enums import TelegramBotMethodEnum
from project.telegram_bot.settings import TelegramBotSettings
class TelegramBotPollingSettings(TelegramBotSettings):
method: Literal[TelegramBotMethodEnum.POLLING] = TelegramBotMethodEnum.POLLING
timeout: PositiveInt = 10 # in seconds
telegram_bot/webhook/settings.py:
from typing import Literal
from pydantic import AnyHttpUrl, conint, constr
from project.telegram_bot.enums import TelegramBotMethodEnum
from project.telegram_bot.settings import TelegramBotSettings
class TelegramBotWebhookSettings(TelegramBotSettings):
method: Literal[TelegramBotMethodEnum.WEBHOOK] = TelegramBotMethodEnum.WEBHOOK
root_url: AnyHttpUrl
root_path: str = "/"
port: conint(ge=1, le=65535) = 8000
secret_access_key: constr(pattern=r"^\w{32}$") = "webhooksuperpupersecretaccesskey"
telegram_bot/service.py:
import aiogram
import facet
from .settings import TelegramBotSettings
class BaseTelegramBotService(facet.AsyncioServiceMixin):
def __init__(
self,
settings: TelegramBotSettings,
):
self._settings = settings
self._dispatcher = aiogram.Dispatcher()
self._bot = aiogram.Bot(token=settings.token)
async def start(self):
self.add_task(self.listen_events())
async def listen_events(self):
raise NotImplementedError
telegram_bot/polling/service.py:
from project.telegram_bot.base import BaseTelegramBotService
class TelegramBotPollingService(BaseTelegramBotService):
async def listen_events(self):
await self._bot.delete_webhook()
await self._dispatcher.start_polling(
self._bot,
handle_signals=False,
)
telegram_bot/webhook/service.py:
from typing import Any
import aiogram
import fastapi
import uvicorn
import yarl
from project.telegram_bot.base import BaseTelegramBotService
class UvicornServer(uvicorn.Server):
def install_signal_handlers(self):
pass
class TelegramBotWebhookService(BaseTelegramBotService):
async def listen_events(self):
webhook_url = yarl.URL(str(self._settings.root_url)) / self._settings.root_path.lstrip("/")
await self._bot.set_webhook(
url=str(webhook_url),
secret_token=self._settings.secret_access_key,
)
await self.get_server().serve()
def get_server(self) -> uvicorn.Server:
app = fastapi.FastAPI(
root_url=self._settings.root_url,
root_path=self._settings.root_path,
)
app.post("")(self.handler)
config = uvicorn.Config(app=app, host="0.0.0.0", port=self._settings.port)
return UvicornServer(config)
async def handler(
self,
update: dict[str, Any],
x_telegram_bot_api_secret_token: str | None = fastapi.Header(None),
):
if x_telegram_bot_api_secret_token != self._settings.secret_access_key:
raise fastapi.HTTPException(
status_code=fastapi.status.HTTP_403_FORBIDDEN,
detail="Forbidden.",
)
await self._dispatcher.feed_webhook_update(
bot=self._bot,
update=aiogram.types.Update(**update),
)
Router and handler structure
- Each domain or feature gets its own router package under
telegram_bot/ui/(ortelegram_bot/handlers/). - Every router sub-package contains at minimum:
router.py— definesget_router() -> aiogram.Routerand wires each route to a handler function.handlers.py— plain async functions that implement the bot logic.callback_data.py—CallbackDatasubclasses for inline buttons used by this feature.states.py—StatesGroupclasses for FSM flows used by this feature.__init__.py— exports onlyget_router.
Example telegram_bot/ui/router.py:
import aiogram
from aiogram.filters import Command
from . import handlers
from .callback_data import ItemCallbackData
from .states import FormState
def get_router() -> aiogram.Router:
router = aiogram.Router()
router.message.register(handlers.start_handler, Command("start"))
router.message.register(handlers.ask_name_handler, FormState.WAITING_FOR_NAME)
router.callback_query.register(
handlers.item_callback_handler,
ItemCallbackData.filter(),
)
return router
Example telegram_bot/ui/__init__.py:
from .router import get_router
__all__ = ("get_router",)
- Do not construct
aiogram.Routerinstances at import time outsideget_router(). The router is created only whenget_router()is called during service setup.
Dispatcher setup
- If the router needs external dependencies (repositories, services, etc.), expose a
setup_dispatcher(dispatcher, **deps)function that attaches middlewares and includes routers.
Example telegram_bot/ui/dispatcher.py:
import aiogram
from . import get_router
from .middlewares import AuthMiddleware, TypingMiddleware
def setup_dispatcher(
dispatcher: aiogram.Dispatcher,
users_repository: UsersRepository,
):
auth_middleware = AuthMiddleware(users_repository=users_repository)
typing_middleware = TypingMiddleware()
for middleware in (auth_middleware, typing_middleware):
dispatcher.update.outer_middleware(middleware)
router = get_router()
dispatcher.include_router(router)
Middlewares
- Keep middlewares stateless. If they need configuration, accept it as an argument and pass it when wiring in
setup_dispatcher().
Example telegram_bot/ui/middlewares/typing.py:
import asyncio
import contextlib
from collections.abc import Awaitable, Callable
import aiogram
class TypingMiddleware(aiogram.BaseMiddleware):
async def __call__(
self,
handler: Callable[[aiogram.types.TelegramObject, dict[str, Any]], Awaitable[Any]],
event: aiogram.types.TelegramObject,
data: dict[str, Any],
):
if not isinstance(event, aiogram.types.Message):
return await handler(event, data)
bot = data.get("bot")
chat_id = event.chat.id
async with send_typing_action(bot=bot, chat_id=chat_id):
return await handler(event, data)
@contextlib.asynccontextmanager
async def send_typing_action(bot: aiogram.Bot, chat_id: int, delay: float = 3):
async def background_task():
while True:
await bot.send_chat_action(chat_id=chat_id, action="typing")
await asyncio.sleep(delay)
task = asyncio.create_task(background_task())
try:
yield
finally:
task.cancel()
try:
await task
except asyncio.CancelledError:
pass
Callbacks and Inline Keyboards
- Use
aiogram.filters.callback_data.CallbackDatasubclasses for every inline button action. Theprefixmust be unique across the whole bot. - For actions that share a paginated list, create a base
PaginatorCallbackDatawith apagefield and inherit from it.
Example telegram_bot/ui/callback_data.py:
import enum
import uuid
from aiogram.filters.callback_data import CallbackData
class PaginatorCallbackData(CallbackData, prefix=""):
page: int = 1
class MenuCallbackData(CallbackData, prefix="menu"):
pass
class ItemsCallbackData(PaginatorCallbackData, prefix="items"):
pass
class ItemActionEnum(str, enum.Enum):
SHOW = "show"
DELETE = "delete"
class ItemCallbackData(CallbackData, prefix="item"):
item_id: uuid.UUID
action: ItemActionEnum = ItemActionEnum.SHOW
- Build inline keyboards with
aiogram.utils.keyboard.InlineKeyboardBuilder. - Pack callback data with
.pack(). Unpack inside handlers withMyCallbackData.unpack(callback_query.data).
Example inline keyboard builder in a handler:
from aiogram import types
from aiogram.utils.keyboard import InlineKeyboardBuilder
from .callback_data import ItemCallbackData, ItemsCallbackData
async def show_items_handler(message: types.Message):
keyboard = InlineKeyboardBuilder()
for item in items:
keyboard.row(types.InlineKeyboardButton(
text=item.name,
callback_data=ItemCallbackData(item_id=item.id).pack(),
))
keyboard.row(types.InlineKeyboardButton(
text="Next page",
callback_data=ItemsCallbackData(page=2).pack(),
))
await message.reply(text="Items:", reply_markup=keyboard.as_markup())
Lists and Pagination
- Use a shared
PaginatorCallbackDatabase for every paginated list. - The maximum number of list items per page is 5.
- In the handler or logic layer, unpack the callback to read
page, computetotal,pages_count, and render a single navigation row. - The navigation row contains up to 3 buttons in this order:
- "◀️" — shown only when
page > 1. - "{current_page}/{pages_count}" — shown only when
pages_count > 1. - "▶️" — shown only when
page < pages_count.
- "◀️" — shown only when
- Navigation buttons use the same
CallbackDatasubclass with an incremented or decrementedpage.
Example pagination logic inside a handler or logic.py:
from aiogram import types
from aiogram.utils.keyboard import InlineKeyboardBuilder
from .callback_data import ItemsCallbackData, ItemCallbackData
PER_PAGE = 5
async def render_item_list(
message: types.Message | None,
callback_query: types.CallbackQuery | None,
page: int,
total: int,
items: list[Item],
):
pages_count = total // PER_PAGE + int(bool(total % PER_PAGE))
keyboard = InlineKeyboardBuilder()
for item in items[:PER_PAGE]:
keyboard.row(types.InlineKeyboardButton(
text=item.name,
callback_data=ItemCallbackData(item_id=item.id).pack(),
))
navigation_row = []
if page > 1:
navigation_row.append(types.InlineKeyboardButton(
text="◀️",
callback_data=ItemsCallbackData(page=page - 1).pack(),
))
if pages_count > 1:
navigation_row.append(types.InlineKeyboardButton(
text=f"{page}/{pages_count}",
callback_data="null",
))
if page < pages_count:
navigation_row.append(types.InlineKeyboardButton(
text="▶️",
callback_data=ItemsCallbackData(page=page + 1).pack(),
))
if navigation_row:
keyboard.row(*navigation_row)
reply_markup = keyboard.as_markup()
if message is not None:
await message.reply(text="Items:", reply_markup=reply_markup)
elif callback_query is not None:
await callback_query.message.edit_text(text="Items:")
await callback_query.message.edit_reply_markup(reply_markup=reply_markup)
- Keep the handler thin: extract keyboard building and pagination math into a
logic.pymodule or a dedicated builder function.