Предисловие
Продолжаем рассказывать о разработке нашего Open Source проекта Taigram.
В этой статье мы поделимся тем, как организовали конфигурацию приложения, решили проблемы с запуском в разных окружениях и как справлялись с потоком уведомлений.
Также затронем тему бета-тестирования и расскажем, как вы можете помочь нам сделать Taigram лучше.
Класс конфигурации
Хорошим тоном является создание файла конфигурации, в котором прописываются все значения, которые могут быть изменены без "хардкода". Это позволяет разработчику и пользователю контролировать конфигурацию приложения, не прибегая к поиску прописанных в коде значений.
Для этого мы создали класс Configuration, основанный на Singleton, о котором мы писали в прошлой статье "Taigram: Архитектура приложения".
В этом классе у нас пять полей:
settings- объект классаDynaconf;logger- объект классаLoggerUtils;strings- словарь (dict) с текстом, необходимым в приложении, генерируемый в функцииgenerate_strings_dict();bot- объект классаBotиз библиотекиaiogram;dispatcher- Объект классаDispatcherиз библиотекиaiogram.
class Configuration(Singleton):
settings = Dynaconf(envvar_prefix=False, environments=True, settings_files=["config/settings.yaml"])
logger = LoggerUtils(settings=settings)
strings = generate_strings_dict(path=settings.YAML_FILE_PATH)
bot = Bot(token=settings.TELEGRAM_BOT_TOKEN, default=DefaultBotProperties(parse_mode="HTML"))
dispatcher = Dispatcher()Про всё это было рассказано в прошлой статье, так, что повторяться нет смысла.
Для того, чтобы обратиться к переменной из файла конфигураций, необходимо вызывать Configuration.settings.API_TOKEN, а для строк, например, Configuration.strings.get("...").get("..."). Согласитесь, выходит весьма "громоздко".
Для получения конфигурации, логгера и текста, были написаны три get-функции:
def get_settings() -> Dynaconf:
return Configuration.settings
def get_strings() -> dict:
return Configuration.strings
def get_logger(**kwargs) -> Logger:
if not kwargs.get("name"):
kwargs["name"] = __name__
return Configuration.logger.get_logger(**kwargs)Их суть сводится к более удобному доступу до часто используемых полей класса.
Однако, "к чему весь этот геморрой", спросите вы? Ведь можно сделать так:
settings = Dynaconf(envvar_prefix=False, environments=True, settings_files=["config/settings.yaml"])
logger = LoggerUtils(settings=settings)
strings = generate_strings_dict(path=settings.YAML_FILE_PATH)
bot = Bot(token=settings.TELEGRAM_BOT_TOKEN, default=DefaultBotProperties(parse_mode="HTML"))
dispatcher = Dispatcher()И окажетесь правы, так можно сделать и никто вас за это не побьёт, но для себя мы решили, что в нашем проекте не будет "глобальных переменных", совсем, если, конечно, не считать вызов логгера в начале модуля где он применяется.
Именно по этой причине Singleton пришёлся к месту. Класс инициализируется всего один раз:
Читается конфигурационный файл;
Инициализируется логгер;
Формируется словарь строк;
Инициализируется объект бота и диспетчер.
Из любого места в коде мы можем быстро обратиться к Configuration и получить доступ или воспользоваться "сокращающими путь" функциями.
Мы считаем это удачным решением. Оно позволит расширять класс по необходимости и гарантировать, что новые разработчики, вместо десятка переменных будут знать, что нужные им данные точно есть в вызове Configuration.
Точка входа
Без точки входа не обходится ни одно приложение. Нужно же его как-то запускать =)
В нашем проекте их две:
runner.py- точка входа дляuv;app.py- основной запуск приложения.
Точка входа для uv
Также как и poetry, uv позволяет запускать приложение самостоятельно, что оказывается весьма удобно и универсально. Вместо команды python -m <путь_к_файлу>, достаточно просто выполнить uv run app.
Реализуется это достаточно просто, добавлением системы сборки и создание скрипта в pyproject.toml:
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src"]
[project.scripts]
app = "src.runner:run"В первом блоке указываем систему для сборки - Hatchling.
Во втором блоке указываем расположение исходного кода проекта. В нашем случае это директория src.
Последний блок самый интересный. В нём мы указываем название скрипта и расположение функции, которую необходимо вызвать. Нашем случае это:
app- название скрипта. Вызывается командойuv run app;src- директория с исходным кодом;runner- название Python-модуля, в нашем случае этоrunner.py;run- функция внутри указанного файла.
Сам файл runner.py достаточно прост:
def run():
logger.info("Starting...")
run_app()
logger.info("Stopping...")В нём вызывается функция run_app() из модуля app.py, о которой следующий блок.
Точка входа app.py
В файле app.py находится основная точка входа в приложение - run_app().
Поскольку проект использует одновременно FastAPI и aiogram, перед нами стояла непростая задача. Сложность заключалась не в том, чтобы запустить их вместе, а в том, чтобы разделить окружения.
Дело в том, что в production окружении, FastAPI предоставляет вебхук-маршруты, а aiogram передаёт на сервера Telegram URL-вебхука. При локальной разработке невозможно получать данные от Telegram используя вебхук, если у вас нет статичного IP или не настроено специальное ПО для перенаправления запросов.
Мы решили проблему следующим образом:
Функция run_app(), при помощи Dynaconf получает значение текущего окружения и в зависимости от него создаёт экзепляр FastAPI с жизненным циклом (lifespan) для dev и prod окружении.
Коротко про lifespan - это функция, реализующая асинхронный контекстный менеджер. В ней можно прописать различные действия выполняемые перед запуском и перед остановкой приложения, например, инициализация БД или как в нашем случае, запуск Telegram-бота.
def run_app():
match current_env := get_settings().current_env:
case EnvironmentEnum.PROD:
web_app = FastAPI(lifespan=prod_lifespan)
case EnvironmentEnum.DEV | EnvironmentEnum.TEST:
web_app = FastAPI(lifespan=dev_lifespan)
case _:
raise RuntimeError(f"Unknown environment {current_env}")
asyncio.run(register_bot_middlewares())
asyncio.run(register_bot_routers())
web_app.include_router(web_app_router)
web_app = asyncio.run(handling_exceptions(app=web_app))
uvicorn.run(web_app, host="0.0.0.0", port=8000, loop="asyncio", log_config=None)Для этого, в этом же файле созданы две lifespan-функции: prod_lifespan и dev_lifespan.
Не будем заострять внимание на содержимом кода, рассмотрим логику и ключевые моменты.
prod_lifespan
@asynccontextmanager
async def prod_lifespan(app: FastAPI):
settings = get_settings()
bot = Configuration.bot
url_webhook = f"{settings.WEBHOOK_DOMAIN}{settings.UPDATES_PATH}"
await bot.set_webhook(
url=url_webhook,
allowed_updates=Configuration.dispatcher.resolve_used_update_types(),
drop_pending_updates=True,
)
await start_bot()
yield
await stop_bot()
await bot.delete_webhook()
await bot.session.close()В этой функции при запуске проекта выполняется добавление вебхука к Telegram-боту и выполняется оповещение администратора о том, что бот запущен. После остановки приложения, выполняется оповещение и удаление установленного вебхука с закрытием сессии.
dev_lifespan
@asynccontextmanager
async def dev_lifespan(app: FastAPI):
async def _start_polling():
await Configuration.dispatcher.start_polling(Configuration.bot, handle_signals=False)
polling_task = asyncio.create_task(_start_polling())
yield
polling_task.cancel()
await Configuration.bot.session.close()Тут чуточку интереснее. При запуске проекта инициализируется Telegram-бот используя механизм long-polling. Интересно это тем, что вызвало проблему, а именно, при остановке приложения в терминале через CTRL+C, останавливался только бот, FastAPI продолжал работать. Связанно это с тем, что библиотеки независимо друг от друга обрабатывают команды KeyboardInterrupt и срабатывает "первая поймавшая", т.е. постоянно обновляющийся бот.
Решение этой проблемы оказалось невероятно простым! Достаточно в метод .start_polling() передать аргумент handle_signals=False, запрещающий боту обрабатывать системные сигналы.
После остановки проекта, всё более-менее такое же, как и в предыдущей функции. Останавливаем polling и закрываем сессию.
Таким образом мы решили проблему локальной разработки и запуска проекта на сервере.
Запустив проект локально, у разработчика будут рабочие маршруты, на которые он сможет самостоятельно отправлять данные для проверки работоспособности.
Разбор данных из Taiga
Предыстория
Не стоит забывать зачем мы начали разработку Taigram, а именно: парсить данные из вебхуков и отправлять их в Telegram.
Мы и тут допустили ошибки, но вовремя их исправили настолько, насколько это возможно без изменения исходного кода Taiga.
Стоит немного восстановить историю: поскольку мы учимся использовать проектные менеджеры для работы над проектами (в нашем случае это Taiga), то мы хотим настроить процесс взаимодействия с доской максимально эффективно и по-возможности, автоматизировано. Напомним, что нас не устраивали уведомления, которых приходили на почту и когда мы увидели, что для каждого проекта есть возможность добавить неограниченное количество веб-хуков, то у нас зародилась идея отправлять уведомления в Telegram (о том какое еще другое-плохое зло мы задумали благодаря "неограниченному количеству вебхуков" - мы расскажем в следующей статье).
Мы сразу обсудили, какой функционал нам нужен и что для этого потребуется, после чего приступили к разработке. Однако мы упустили важный шаг — не изучили данные, которые Taiga отправляет через вебхуки. Мы сделали это, но значительно позже.
Временное решение
До того, как мы настроили обработчики внутри бота, мы использовали временное решение: сервис Webhook.site, который формирует ссылки для получения данных от вебхуков. У этого сервиса были ограничения (например, 100 сообщений на одну ссылку, а также ограниченное время жизни ссылок), но на начальном этапе этого было достаточно.
Структура данных
Из-за того, что изначально мы смогли найти только документацию Taiga по API, но не по WebHook - мы подумали, что хорошей идеей будет получить уведомления по всем возможным событиям (создание, изменение, удаление) всех возможных типов событий (эпик, спринт, история, задача, запрос, вики).
Задача казалась несложной, так как структура уведомлений выглядела понятной и даже исчерпывающей (хотя иногда данные дублировались без явной необходимости).
Основные поля уведомления
Каждое уведомление содержит следующие обязательные поля:
action: Описывает тип произошедшего события (create,change,delete).type: Указывает тип сущности, к которой относится событие (epic,milestone,userstory,task,issue,wiki-page).by: Вложенная структура, содержащая информацию об инициаторе события.date: Время события в формате datetime.data: Вложенная структура с основными данными уведомления.
Сложности с полем data
Структура поля data оказалась более сложной, так как она зависит от типа события. Например, для задачи (task) в data могут быть указаны вложенные структуры, такие как project, milestone, userstory. При этом в структуре userstory отдельно дублируется информация о milestone.
Изначально обилие данных казалось полезным, но в процессе работы мы столкнулись с тем, что нам не хватало уникальных данных. Это стало одной из проблем, к которой мы вернемся позже.
Дополнительные материалы
Мы не будем подробно описывать специфику всех структур данных, так как с ними можно ознакомиться:
В нашем репозитории проекта.
В документации Taiga.
На нашей странице с документацией (в настоящее время в разработке).
Пример данных для создания задачи:
{
"action": "create",
"type": "task",
"by": {
"id": 6,
"permalink": "https://tasks.pressanybutton.ru/profile/whgaleon1",
"username": "whgaleon1",
"full_name": "Victor Vangeli",
},
"date": "2025-02-12T12:14:54.458Z",
"data": {
"custom_attributes_values": {},
"id": 181,
"ref": 103,
"created_date": "2025-02-12T12:14:54.360Z",
"modified_date": "2025-02-12T12:14:54.375Z",
"finished_date": null,
"due_date": null,
"due_date_reason": "",
"subject": "Добавить мультиязычность",
"is_iocaine": false,
"watchers": [],
"is_blocked": false,
"blocked_note": "",
"description": "",
"tags": [],
"permalink": "https://tasks.pressanybutton.ru/project/taiga-webhook-telegram-notifier/task/103",
"project": {
"id": 6,
"permalink": "https://tasks.pressanybutton.ru/project/taiga-webhook-telegram-notifier",
"name": "Taiga WebHook Telegram Notifier",
"logo_big_url": null
},
"assigned_to": null,
"status": {
"id": 26,
"name": "New",
"is_closed": false
},
"user_story": {
"custom_attributes_values": {},
"id": 47,
"ref": 67,
"is_closed": false,
"created_date": "2025-02-11T16:11:58.721Z",
"modified_date": "2025-02-12T09:27:57.850Z",
"finish_date": null,
"due_date": null,
"due_date_reason": "",
"subject": "Телеграм клавиатура",
"client_requirement": false,
"team_requirement": false,
"generated_from_issue": null,
"generated_from_task": null,
"from_task_ref": null,
"external_reference": null,
"watchers": [,
"is_blocked": false,
"blocked_note": "",
"description": "",
"tags": [],
"permalink": "https://tasks.pressanybutton.ru/project/taiga-webhook-telegram-notifier/us/67",
"owner": {
"id": 6,
"permalink": "https://tasks.pressanybutton.ru/profile/whgaleon1",
"username": "whgaleon1",
"full_name": "Victor Vangeli",
"photo": "https://tasks.pressanybutton.ru/media/user/1/3/4/4/372f7e07418b363d9b6759af3f5ea907f919d8fe330c0b6523daaab12adf/huu74pd-mda.jpeg.80x80_q85_crop.jpg",
"gravatar_id": "05adc03d5532a78d6695e6cafa1da0a9"
},
"assigned_to": {
"id": 6,
"permalink": "https://tasks.pressanybutton.ru/profile/whgaleon1",
"username": "whgaleon1",
"full_name": "Victor Vangeli",
"photo": "https://tasks.pressanybutton.ru/media/user/1/3/4/4/372f7e07418b363d9b6759af3f5ea907f919d8fe330c0b6523daaab12adf/huu74pd-mda.jpeg.80x80_q85_crop.jpg,
"gravatar_id": "05adc03d5532a78d6695e6cafa1da0a9"
},
"assigned_users": [
6
],
"points": [
{
"role": "Design",
"name": "?",
"value": null
},
{
"role": "Back",
"name": "5",
"value": 5.0
}
],
"status": {
"id": 32,
"name": "New",
"slug": "new",
"color": "#70728F",
"is_closed": false,
"is_archived": false
},
"milestone": {
"id": 12,
"name": "Базовый функционал",
"slug": "bazovyi-funktsional",
"estimated_start": "2025-02-11",
"estimated_finish": "2025-02-21",
"created_date": "2025-02-10T19:32:31.034Z",
"modified_date": "2025-02-10T19:32:31.043Z",
"closed": false,
"disponibility": 0.0,
"permalink": "https://tasks.pressanybutton.ru/project/taiga-webhook-telegram-notifier/taskboard/bazovyi-funktsional",
"project": {
"id": 6,
"permalink": "https://tasks.pressanybutton.ru/project/taiga-webhook-telegram-notifier",
"name": "Taiga WebHook Telegram Notifier",
"logo_big_url": null
},
"owner": {
"id": 6,
"permalink": "https://tasks.pressanybutton.ru/profile/whgaleon1",
"username": "whgaleon1",
"full_name": "Victor Vangeli",
"photo": "https://tasks.pressanybutton.ru/media/user/1/3/4/4/372f7e07418b363d9b6759af3f5ea907f919d8fe330c0b6523daaab12adf/huu74pd-mda.jpeg.80x80_q85_crop.jpg",
"gravatar_id": "05adc03d5532a78d6695e6cafa1da0a9"
}
}
},
"milestone": {
"id": 12,
"name": "Базовый функционал",
"slug": "bazovyi-funktsional",
"estimated_start": "2025-02-11",
"estimated_finish": "2025-02-21",
"created_date": "2025-02-10T19:32:31.034Z",
"modified_date": "2025-02-10T19:32:31.043Z",
"closed": false,
"disponibility": 0.0,
"permalink": "https://tasks.pressanybutton.ru/project/taiga-webhook-telegram-notifier/taskboard/bazovyi-funktsional",
"project": {
"id": 6,
"permalink": "https://tasks.pressanybutton.ru/project/taiga-webhook-telegram-notifier",
"name": "Taiga WebHook Telegram Notifier",
"logo_big_url": null
},
},
"promoted_to": []
}
}Проблемы и наше решение
1. Нам не нужен монолит
Мы столкнулись с необходимостью обработки данных, поступающих через вебхуки, и решили, что хорошей идеей будет использовать pydantic для создания схем данных. Это позволило бы нам аннотировать все возможные типы полей и гарантировать, что данные будут корректными и готовыми к отправке, отсекая всё лишнее.
Тогда мы создали пакет webhook_data с такой структурой:
- webhook_data
- __init__.py
- base_webhook_schemas.py
- diff_webhook_schemas.py
- nested_schemas.py
- webhook_payload_schemas.pyВ файле webhook_payload_schemas.py мы реализовали основную схему WebhookPayload, с которой работаем при подготовке уведомлений:
class WebhookPayload(BaseModel):
action: str
type: EventTypeEnum
by: User
date: datetime
data: Task | Milestone | UserStory | Epic | Wiki | Issue
change: Change | None = None
@field_validator("data", mode="before")
def validate_data(cls, value, values):
type_map = {
EventTypeEnum.TASK: Task,
EventTypeEnum.MILESTONE: Milestone,
EventTypeEnum.USERSTORY: UserStory,
EventTypeEnum.EPIC: Epic,
EventTypeEnum.WIKIPAGE: Wiki,
EventTypeEnum.ISSUE: Issue,
}
target_type = type_map.get(values.data.get("type"))
if not target_type:
raise ValueError(f"Неизвестный тип: {values['type']}")
if isinstance(value, dict):
return target_type(**value)
return valueМы разделили схемы на несколько модулей:
base_webhook_schemas.py: Содержит базовые схемы, от которых наследуются другие.nested_schemas.py: Включает схемы для вложенных структур.diff_webhook_schemas.py: Описывает изменения, которые приходят в уведомлениях.
Такое разделение позволило нам избежать создания монолитного модуля с огромным количеством сущностей. Мы стремились к инкапсуляции всего, что возможно, и, как показала практика, это было правильным решением.
2. Разработка шаблона уведомлений
После тщательного анализа данных, полученных через вебхуки, и создания первоначальных pydantic схем, мы приступили к следующим шагам:
Подготовка шаблонов для всех типов уведомлений;
Написание логики формирования этих уведомлений;
Тестирование всей системы.
С первым пунктом мы справились достаточно быстро, во многом благодаря тому, что к команде присоединился Роман. Он взял на себя разработку модулей для формирования уведомлений. Для начала работы ему потребовались инструкции и минимальный пример того, как должно выглядеть итоговое уведомление. Вот пример одного из таких уведомлений:
Интересные статьи
Интересные статьи
