Телеграм-бот для бронирования столов на вебхуках: FastAPI, Aiogram Dialog, FastStream и RabbitMQ в единой экосистеме

Друзья, приветствую.

На Хабре я уже описал достаточно много материала, но гештальт с Aiogram dialog и взаимодействие с брокером сообщений RabbitMQ так и не закрыл. Сегодня хочу исправить это.

В рамках статьи, которую вы сейчас читаете, мы реализуем функционального телеграм-бота. Как вы поняли из названия, для реализации этой задачи мы будем использовать брокер сообщений RabbitMQ и ряд мощных Python-инструментов.

Постараюсь вас сильно не грузить теорией, но, к сожалению, без неё никуда. Поэтому буду придерживаться баланса между теорией и написанием кода по ходу повествования далее.

В рамках теоретического блока мы коротко пройдемся по таким темам, как брокеры сообщений и зачем они нужны, поговорим про вебхуки в мире разработки телеграм-ботов и разберемся, зачем в этот проект я решил подтянуть Aiogram dialog, FastStream, SQLAlchemy и прочие технологии.

Функционал, который будут видеть пользователи

Теперь давайте отдельно поговорим о том, что за бота мы будем писать или, другими словами, давайте разберемся, что наш бот будет уметь делать.

Напишем мы телеграм-бота для бронирования столиков в выдуманном ресторане «Binary Bites». Под бронированием я подразумеваю следующие этапы:

1. Выбор количества мест

2. Выбор конкретного стола

3. Выбор даты, на когда бронь

4. Выбор времени, на когда бронь

5. Подтверждение

Кроме того, пользователь сможет просматривать свои брони и, при необходимости, отменять их. Также он получит доступ к информации о ресторане.

Добавим простенькую админку, чтобы пользователь с уровнем доступа администратора мог просматривать все существующие брони и мог получать информацию по всем пользователям.

Функционал «под капотом»

Как вы понимаете, вся основная «магия» у нас будет происходить «под капотом». Сейчас я не буду говорить про логику взаимодействия с базой данных, думаю, очевидно, что нам нужно будет как-то сохранить пользователя, добавить столы с параметрами, проверить, свободен ли стол, и прочее — это очевидно. Сейчас же я хочу сосредоточиться на более интересной логике.

Во-первых, мы сделаем так, чтоб наш бот работал не через лонг поллинг, который идет «из коробки» при разработке ботов, а через технологию вебхуков. О том, зачем я выбрал такой подход и какой профит он нам даст, подробнее поговорим далее.

В видео «Вебхуки для начинающих: от основ до Telegram-бота за 60 минут» я подробно рассказал о том, что такое вебхуки и как они работают в контексте Telegram-ботов. Если вам интересен аудио-видео формат, приглашаю вас к просмотру. Я постарался максимально раскрыть тему (видео доступно и на Rutube).

Во-вторых, наш бот будет взаимодействовать с брокером сообщений RabbitMQ. Это говорит о том, что наша система, кроме данного брокера, будет подразумевать наличие продюсера (того, кто ставит задачу) и слушателя (того, кто эту задачу будет исполнять). О том, что это значит, поговорим в отдельном блоке. Сейчас скажу, что эту задачу мы будем закрывать через FastStream.

В-третьих, наш бот будет уметь работать с отложенными и задачами по расписанию. К сожалению, в FastStream нет метода «из коробки» для работы с отложенными и задачами по расписанию, поэтому мы дополнительно поставим APScheduler. Но в статье я дополнительно покажу, как использовать связку FastStream + APScheduler.

В общих чертах определились, теперь давайте более детально рассмотрим стек используемых технологий в данном проекте.

Стек технологий

Как вы поняли, писать код мы будем на языке Python, а в качестве брокера будем использовать RabbitMQ.

Технологии Python, которые сегодня будут использованы:

• Aiogram 3: лучший фреймворк для разработки телеграм-ботов на Python

• Aiogram dialog 2.3: фреймворк, который очень сильно упрощает работу с FSM в Aiogram 3

• SQLAlchemy 2: фреймворк для работы с табличными базами данных (демонстрацию покажу на примере SQLite, но можно будет легко переключить на любую другую табличную базу данных)

• FastStream 0.5.3: python-фреймворк для работы с брокерами сообщений (многие нарекают его «убийцей Celery»)

• APScheduler 3.1: фреймворк, который позволит удобно работать с отложенными задачами и задачами по расписанию

• FastAPI: фреймворк, который позволит всех этих ребят: телеграм-бота, APScheduler, FastStream и так далее объединить в рамках одного приложения (одной экосистемы)

• Alembic: для миграций базы данных

• Uvicorn: для запуска веб-сервера

Дополнительно используем aiosqlite, pydantic, loguru и прочие библиотеки.

Как видите, стек достаточно большой.

План на статью

Начнем мы с того, что просто разберемся теоретически с брокерами, продюсерами, слушателями, фоновыми задачами и прочими сущностями, которых будем сегодня касаться. Условно, начнем мы с теории.

Далее приступим к разработке. Тут можно будет выделить следующие этапы:

1. Подготовка. В рамках этапа мы поднимем RabbitMQ на локальной машине, подготовим структуру проекта, установим зависимости и прочее.

2. База данных. На этом этапе мы опишем модели таблиц, превратим их в реальные таблицы базы данных (выполним миграции) и напишем все необходимые методы взаимодействия с базой данных. Говорю сразу, блок этот сегодня досканально, как в прошлых статьях своих, разбирать не буду, так как процесс тогда снова может занять пол статьи.

3. Разработаем базовую пользовательскую часть бота и админку.

4. Подробно разберем блок с диалогом бронирования (Aiogram dialog).

5. Разберемся с блоком взаимодействия с APScheduler и FastStream.

6. Выполним деплой проекта (удаленный запуск).

Что касается блока с удаленным запуском, хочу остановиться подробнее. Дело тут в том, что, несмотря на то, что мы будем запускать сразу несколько разных сервисов через FastAPI, у нас в любом случае будет необходимость где-то поднимать отдельно RabbitMQ. Считаю, что удобно, когда все в одном месте, поэтому для деплоя проекта я выбрал сервис Amvera Cloud.

Почему Amvera?

Первая и основная причина — это простота деплоя. Нам буквально достаточно будет закинуть файлы проекта (наш код) на сервис Amvera. Затем, прямо на сервисе, можно будет собрать конфигурационный файл, и проект запустится.

Доставить файлы можно или через интерфейс, или через стандартные GIT-команды. Я пример покажу через загрузку через интерфейс на сайте.

Бонусом Amvera дает HTTPS-доменное имя, которого будет более чем достаточно для полноценной работы бота в формате вебхуков.

Кроме того, новые пользователи получают 111 рублей на основной баланс.

Вторая причина — это возможность прямо на Амвере поднять RabbitMQ. В этом случае вам даже не нужно будет никакие файлы добавлять. Буквально пару кликов мышкой, и бот взаимодействует с RabbitMQ на серверах Амверы. После останется боту подключаться уже к новому RabbitMQ. На деле все ещё проще, чем кажется.

Зачем выбран FastApi

В архитектуре нашего проекта важное место занимают микросервисы, каждый из которых может функционировать как самостоятельная единица. FastStream и APScheduler — яркие примеры таких компонентов, обладающих собственным жизненным циклом и требующих отдельной инициализации и запуска.

Когда мы говорим о запуске этих сервисов, речь идёт не просто о вызове функций, а о полноценном запуске долгоживущих процессов, которые после активации начинают выполнять свою логику (хотя с FastStream ситуация несколько сложнее, об этом поговорим позже).

Ключевой особенностью нашего бота является использование технологии вебхуков вместо традиционного лонг-поллинга. Эта архитектурная концепция полностью меняет подход к получению обновлений: вместо того, чтобы наш бот периодически опрашивал серверы Telegram, серверы сами уведомляют нас о новых событиях.

Для тех, кто хочет глубже понять принципы работы вебхуков, особенно в контексте разработки Telegram-ботов, рекомендую посмотреть моё видео «Вебхуки для начинающих: от основ до Telegram-бота за 60 минут» (https://www.youtube.com/watch?v=g4yZDnoVD3g&t=2146s).

Если объяснить кратко: при любом действии пользователя в Telegram (отправка сообщения, нажатие кнопки) генерируется обновление. При использовании лонг-поллинга бот регулярно подключается к серверам Telegram и проверяет наличие новых обновлений. В случае с вебхуками серверы Telegram сами инициируют отправку обновлений на указанный нами эндпоинт.

Для реализации такого подхода требуется веб-сервер со специальным маршрутом, способным обрабатывать поступающие обновления. И здесь на сцену выходит FastAPI — высокопроизводительный фреймворк, который идеально справляется с задачей быстрой обработки запросов от серверов Telegram. Это позволяет нам создать масштабируемого и отзывчивого бота, способного эффективно обрабатывать значительные нагрузки.

Помимо превосходной производительности, FastAPI предлагает гибкую систему управления жизненным циклом приложения. Это даёт нам возможность элегантно организовать запуск всех компонентов — Telegram-бота, FastStream, APScheduler и других микросервисов — при старте основного приложения.

Таким образом, FastAPI выступает не просто как веб-сервер, а как объединяющая платформа, позволяющая гармонично интегрировать все элементы нашей системы в рамках единого, хорошо структурированного приложения.

Архитектура брокеров сообщений: основы и применение в проекте

В сердце нашего проекта лежит архитектура, основанная на брокерах сообщений — элегантное решение для организации асинхронного взаимодействия между компонентами системы. Эта архитектура строится на трех фундаментальных компонентах, каждый из которых выполняет свою уникальную роль.

Ключевые компоненты системы

1. Продюсеры (Producers) — источники сообщений, стоящие в начале информационного потока. Они формируют задачи, упаковывают их в структурированные сообщения и направляют их в специализированные каналы (очереди) брокера. Продюсеры инициируют процесс коммуникации, не заботясь о том, кто и когда обработает их сообщения.

2. Слушатели (Consumers) — исполнители, постоянно прослушивающие определенные каналы в ожидании новых сообщений. Как только сообщение появляется в очереди, слушатель мгновенно реагирует, извлекает его и приступает к выполнению соответствующей бизнес-логики. Эти компоненты отвечают за фактическую реализацию задач, порожденных продюсерами.

3. Брокер сообщений (Message Broker) — центральное звено системы, представленное в нашем случае RabbitMQ. Брокер выполняет роль высоконадежного посредника, который:

   • Принимает и сохраняет сообщения от продюсеров

   • Организует маршрутизацию сообщений к соответствующим слушателям

   • Гарантирует доставку, сохраняя сообщения до подтверждения их успешной обработки

   • Обеспечивает балансировку нагрузки при наличии нескольких слушателей

Информационный поток в системе

Взаимодействие компонентов образует элегантный цикл: продюсер генерирует сообщение и отправляет его брокеру, который временно хранит это сообщение и затем маршрутизирует его к подходящему слушателю. Слушатель обрабатывает сообщение и, при необходимости, отправляет подтверждение об успешном выполнении. Такая архитектура обеспечивает:

• Асинхронность выполнения задач

• Высокую масштабируемость системы

• Отказоустойчивость при сбоях отдельных компонентов

• Гарантированную доставку сообщений

Технологический стек реализации

Для воплощения этой архитектуры я выбрал FastStream — современный и высокопроизводительный фреймворк, предлагающий не только удобный API для работы с брокерами сообщений, но и специализированные интеграции с FastAPI, которые мы будем активно использовать в этом проекте.

Настройка компонентов

Реализация системы требует конфигурации трех основных элементов:

• Брокер: Настройка RabbitMQ выполняется по стандартной процедуре, применимой для любой системы обмена сообщениями. Необходимо развернуть экземпляр брокера и обеспечить к нему необходимый уровень доступа.

• Продюсер и слушатель: Их конфигурация требует более тонкой настройки, которую мы детально рассмотрим в ходе разработки. Важно отметить, что FastStream существенно упрощает этот процесс благодаря своему декларативному API.

Типы задач в нашей системе

В рамках проекта мы будем работать с двумя категориями задач:

1. Немедленные (фоновые) задачи — требующие моментального выполнения. Их обработку берет на себя FastStream, обеспечивая быстрое асинхронное выполнение.

2. Отложенные и периодические задачи — требующие выполнения в определенное время или с заданной периодичностью. Для них мы задействуем APScheduler — мощный планировщик задач, о котором я подробно рассказывал в предыдущей серии публикаций.

Комбинированный подход

Особый интерес представляют сценарии, где мы комбинируем возможности обеих систем. Например, для циклических задач (таких как автоматическое закрытие просроченных бронирований) мы настроим отдельное расписание, а для более сложных кейсов реализуем многоступенчатую схему:

1. Формирование задачи продюсером

2. Передача через брокер к слушателю

3. Регистрация в планировщике APScheduler для отложенного или периодического выполнения

Такой комбинированный подход позволяет нам максимально гибко управлять жизненным циклом каждой задачи и оптимально использовать возможности обеих технологий.

База данных: архитектура и взаимодействие с данными

Концепция бота для бронирования столиков в ресторане "Binary Bites" естественным образом определяет ключевые требования к нашей системе хранения и обработки данных:

• Необходимость систематизированного хранения информации о пользователях, доступных столиках и текущих бронированиях

• Потребность в эффективных механизмах манипулирования этими данными: создание новых записей, обновление существующих, удаление устаревших, а также проведение различных проверок и валидаций

• Требование к надежному и масштабируемому хранилищу всей этой информации

Для решения этих задач мы обратимся к мощному инструментарию, предоставляемому SQLAlchemy 2 – современному и гибкому ORM-фреймворку для Python. Этот инструмент, особенно в сочетании с системой миграций Alembic, обеспечит не только скорость разработки, но и высокий уровень безопасности при работе с данными.

SQLAlchemy 2 позволяет нам элегантно абстрагироваться от низкоуровневых деталей работы с базой данных, предоставляя интуитивно понятный интерфейс для:

• Декларативного описания структуры таблиц и связей между ними

• Построения сложных запросов с использованием выразительного API

• Управления транзакциями и обеспечения целостности данных

• Оптимизации производительности при работе с большими объемами информации

Для тех, кто хочет углубиться в возможности этого фреймворка, я подготовил целую серию публикаций на Хабре, посвященную именно SQLAlchemy 2, где детально рассматриваются как базовые концепции, так и продвинутые техники работы с данным инструментом.

Важно отметить, что SQLAlchemy сам по себе предоставляет только средства для описания моделей и логики взаимодействия с базой данных, но не трансформирует эти абстрактные описания в реальные структуры таблиц. Именно здесь на сцену выходит Alembic – специализированный инструмент для управления миграциями, который позволяет:

• Преобразовывать декларативные модели в физические таблицы

• Отслеживать изменения в структуре базы данных

• Безопасно применять эти изменения на рабочей системе

• Откатывать изменения при необходимости

Такой тандем технологий обеспечивает нам необходимую гибкость в развитии схемы данных вместе с эволюцией функциональности нашего бота, сохраняя при этом высокий уровень надежности и производительности системы в целом.

Aiogram и Aiogram Dialog: удобство работы с машиной состояний

Aiogram заслуженно считается одним из лучших инструментов для создания Telegram-ботов на Python, а возможно, и в мире программирования в целом. Однако у него есть один значительный недостаток — сложность работы с машиной состояний.

Машина состояний в Telegram-ботах

Машина состояний — это модель поведения системы, позволяющая управлять её переходами между различными состояниями в зависимости от условий. В контексте Telegram-ботов это означает возможность вести пользователя по заранее определённому сценарию: бот задаёт вопросы, а пользователь отвечает с помощью текста, кнопок или отправки файлов. Основная цель — корректный сбор и сохранение данных в базе.

Проблемы работы с машиной состояний в Aiogram

Несмотря на преимущества Aiogram, управление состояниями в нём оставляет желать лучшего:

1. Отсутствие поддержки одновременного нахождения в нескольких сценариях. Пользователь может участвовать только в одном диалоге. Чтобы вернуться к предыдущему, необходимо завершить текущий.

2. Обилие однотипного кода. Для реализации базовых элементов интерфейса (например, пагинации в клавиатуре, календаря, перемещения между состояниями) приходится писать много кода вручную.

Решение: Aiogram Dialog

Здесь на помощь приходит Aiogram Dialog — расширение Aiogram, которое со временем эволюционировало в самостоятельный фреймворк. Он решает главные проблемы работы с состояниями, предлагая:

1. Разделение получения данных и отображения сообщений.

2. Объединение кнопок и их обработчиков в единую структуру.

3. Упрощённую маршрутизацию состояний.

4. Гибкие виджеты для удобного построения интерфейса.

Основные концепции Aiogram Dialog

В центре архитектуры Aiogram Dialog — окна (Window). Каждое окно представляет собой сообщение, отправляемое пользователю, и содержит:

• Виджеты (текстовые сообщения, кнопки, клавиатуры);

• Функции обратного вызова, отвечающие за обработку пользовательских действий и передачу данных.

Окна объединяются в диалоги (Dialog), что позволяет легко переключаться между этапами взаимодействия с пользователем. Более того, Aiogram Dialog поддерживает одновременную работу с несколькими диалогами, передачу данных между ними и автоматическое возвращение после завершения вложенного сценария.

На этом, думаю, теоретический блок можно завершать. Приступаем к практике!

Поднимаем RabbitMQ на локальном компьютере

Начнём с развертывания брокера сообщений RabbitMQ на вашем локальном компьютере. Один из самых простых и быстрых способов сделать это — использовать Docker.

Установка Docker Desktop

Если вы новичок в работе с Docker, первым шагом будет установка Docker Desktop. Этот инструмент доступен для Windows, macOS и Linux, обеспечивая удобную среду для работы с контейнерами. Скачайте и установите Docker Desktop с официального сайта, затем запустите его.

После успешной установки вы сможете использовать командную строку для управления контейнерами Docker.

Запуск RabbitMQ через Docker

Чтобы поднять RabbitMQ, выполните следующую команду:

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 \
-e RABBITMQ_DEFAULT_USER=admin \
-e RABBITMQ_DEFAULT_PASS=password \
-e RABBITMQ_DEFAULT_VHOST=myapp_vhost \
rabbitmq:3-management

Эта команда создаст контейнер с RabbitMQ, настроит учетные данные администратора и развернет веб-интерфейс для управления очередями сообщений.

Доступ к веб-интерфейсу RabbitMQ

После успешного запуска контейнера веб-интерфейс RabbitMQ будет доступен по адресу:
http://localhost:15672/

Для входа используйте логин admin и пароль password (рекомендуется заменить их на собственные данные для безопасности).

Получаем токен бота

Для взаимодействия с Telegram-ботом потребуется токен, который можно получить в BotFather — официальном Telegram-боте для создания и управления другими ботами.

1. Откройте Telegram и найдите BotFather.

2. Отправьте команду /newbot.

3. Следуйте инструкциям, укажите имя и юзернейм бота.

4. В конце процесса вы получите токен — сохраните его, так как он понадобится для настройки бота.

Поднимаем туннель для локального доступа

Перед началом разработки важно обеспечить доступ к локальному серверу из внешней сети. Это особенно необходимо, если ваш бот работает на вебхуках. Для этого можно использовать ngrok:

1. Скачайте и установите ngrok с официального сайта.

2. Выполните команду для создания туннеля:

ngrok http 8000

3. Ngrok создаст внешний URL, который можно использовать для настройки вебхуков.

Теперь ваш локальный сервер будет доступен извне, и вы сможете тестировать взаимодействие бота с Telegram API.

Создание проекта

Настроим среду для разработки, чтобы подготовить основу для будущего проекта. Откройте свою любимую среду разработки, например PyCharm, и создайте новый проект.

Установка зависимостей

В корневой директории проекта создайте файл requirements.txt и добавьте в него следующие зависимости:

aiogram==3.17.0
aiogram_dialog==2.3.1
sqlalchemy==2.0.38
pydantic==2.10.6
pydantic_settings==2.7.1
aio-pika==9.5.4
faststream==0.5.34
loguru==0.7.3
aiosqlite==0.21.0
alembic==1.14.1
pytz==2025.1
apscheduler==3.11.0
fastapi==0.115.8
uvicorn==0.34.0

Ранее мы уже разобрали основные технологии, которые будем использовать, поэтому не будем углубляться в их описание. Однако обратите внимание на две ключевые библиотеки:

• aio-pika – это основной движок для FastStream, обеспечивающий корректное взаимодействие с брокером сообщений RabbitMQ.

• aiosqlite – асинхронный драйвер для работы с базой данных, используемый в связке с SQLAlchemy.

Для установки зависимостей выполните команду:

pip install -r requirements.txt

Создание файла конфигурации

Следующим шагом создайте в корневой директории проекта файл .env и заполните его следующим содержимым:

BOT_TOKEN=bot_token
ADMIN_IDS=[admin_id1, admin_id2]
INIT_DB=0
BASE_URL=https://ngrok_url
RABBITMQ_USERNAME=admin
RABBITMQ_PASSWORD=password
RABBITMQ_HOST=127.0.0.1
RABBITMQ_PORT=5672
VHOST=myapp_vhost

Разбор переменных окружения

• BOT_TOKEN – токен вашего Telegram-бота.

• ADMIN_IDS – список ID администраторов, имеющих доступ к управлению ботом.

• INIT_DB – флаг для инициализации базы данных (далее подробнее разберем зачем он нужен).

• BASE_URL – URL, предоставленный ngrok, для работы вебхуков (может быть любой другой туннель, предварительно не забудьте запустить).

• RABBITMQ_USERNAME / RABBITMQ_PASSWORD – учетные данные для подключения к RabbitMQ.

• RABBITMQ_HOST / RABBITMQ_PORT – параметры подключения к брокеру сообщений.

• VHOST – виртуальный хост, используемый в RabbitMQ для изоляции задач.

После этих шагов среда разработки полностью подготовлена, и можно переходить к следующему этапу — написанию кода. Начнем мы с разработки логики для взаимодействия с базой данных.

Настройка файла конфигурации

Чтобы упростить работу с основными переменными проекта, создадим централизованный файл конфигурации.

Создание структуры проекта

1. В корневой директории проекта создайте папку app.

2. Внутри нее создайте файл config.py и заполните следующим содержимым:

import os
from typing import List
from urllib.parse import quote
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from faststream.rabbit import RabbitBroker
from loguru import logger
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    BOT_TOKEN: str
    ADMIN_IDS: List[int]
    INIT_DB: bool
    FORMAT_LOG: str = "{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}"
    LOG_ROTATION: str = "10 MB"
    DB_URL: str = 'sqlite+aiosqlite:///data/db.sqlite3'
    STORE_URL: str = 'sqlite:///data/jobs.sqlite'
    TABLES_JSON: str = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dao", "tables.json")
    SLOTS_JSON: str = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dao", "slots.json")

    BASE_URL: str
    RABBITMQ_USERNAME: str
    RABBITMQ_PASSWORD: str
    RABBITMQ_HOST: str
    RABBITMQ_PORT: int
    VHOST: str

    @property
    def rabbitmq_url(self) -> str:
        return (
            f"amqp://{self.RABBITMQ_USERNAME}:{quote(self.RABBITMQ_PASSWORD)}@"
            f"{self.RABBITMQ_HOST}:{self.RABBITMQ_PORT}/{self.VHOST}"
        )

    @property
    def hook_url(self) -> str:
        """Возвращает URL вебхука"""
        return f"{self.BASE_URL}/webhook"

    model_config = SettingsConfigDict(
        env_file=os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", ".env")
    )


# Инициализация конфигурации
settings = Settings()

# Настройка логирования
log_file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "log.txt")
logger.add(log_file_path, format=settings.FORMAT_LOG, level="INFO", rotation=settings.LOG_ROTATION)

# Создание брокера сообщений RabbitMQ
broker = RabbitBroker(url=settings.rabbitmq_url)

# Создание планировщика задач
scheduler = AsyncIOScheduler(jobstores={'default': SQLAlchemyJobStore(url=settings.STORE_URL)})

Разбор конфигурации

Использование Pydantic Settings

В этом коде основная логика завязана на Pydantic Settings, что позволяет удобно управлять переменными окружения.

Класс Settings отвечает за хранение всех конфигурационных параметров проекта, обеспечивая:

• Гибкость – возможность менять настройки через .env файл.

• Безопасность – пароли и ключи API не хранятся в коде.

• Читаемость – все параметры собраны в одном месте.

Дополнительно определены два вспомогательных свойства:

• rabbitmq_url – формирует строку подключения к RabbitMQ.

• hook_url – возвращает URL вебхука.

Инициализация логирования

Файл логов log.txt создается автоматически, а Loguru позволяет удобно управлять логами, включая их ротацию (удаление старых записей, когда размер превышает 10MB).

Подключение к RabbitMQ

Для взаимодействия с брокером сообщений создается объект RabbitBroker, использующий данные из rabbitmq_url. В нашем приложении данный объект будет отвечать за отправку отправку сообщений в очередь в асинхронном режиме.

Планировщик задач AsyncIOScheduler

Используем AsyncIOScheduler из APScheduler для управления фоновыми задачами.

К сожалению, APScheduler не умеет взаимодействовать с брокером сообщений напрямую, поэтому под него будет использоваться свое хранилище. Я для удобства выбрал в качестве хранилища заданий SQLite, но его можно заменить на PostgreSQL или другой источник.

База данных под APSCheduler будет создана автоматически.

Вас могли заинтересовать эти строки:

TABLES_JSON: str = os.path.join(os.path.dirname(os.path.abspath(**file**)), "dao", "tables.json")
SLOTS_JSON: str = os.path.join(os.path.dirname(os.path.abspath(**file**)), "dao", "slots.json")

Я решил показать вам, как можно сделать миграцию базы данных через JSON файлы.

Нельзя сказать, что это единственный подход, но в некоторых случаях, особенно когда речь заходит за быструю установку в базе данных статических файлов выручает.

Как вы поняли, это пути к файлам с JSON с определенными данными. Вместе с ними, кстати, работает переменная INIT_DB. Если вы установите в файле .env в значении для этого параметра 1, то при запуске FastApi приложения произойдет загрузка данных из этих JSON в базу данных. Сама база данных должна будет предварительно существовать

Логика работы с базой данных

Друзья, практика показывает, что в каждой статье я подробно разбираю взаимодействие с базой данных, и из раза в раз этот блок повторяется. Это связано с тем, что я использую базовый класс с универсальными методами для работы с SQLAlchemy, что значительно упрощает процесс взаимодействия с данными.

Чтобы не дублировать одну и ту же информацию, в этот раз я не буду останавливаться на деталях миграций, особенностях связей между таблицами и универсальных методах работы с данными. Вместо этого сосредоточимся только на тех элементах, которые имеют непосредственное отношение к текущему проекту.

Если вы хотите глубже разобраться в подходах, которые я применяю, рекомендую ознакомиться с моими предыдущими статьями или видео, где я подробно объясняю, как описывать модели таблиц и работать с методами SQLAlchemy.

Кроме того, важно отметить, что в этой статье мы активно будем использовать Pydantic — он понадобится как для работы с базой данных, так и в целом для корректного функционирования FastAPI. Если вы еще не знакомы с Pydantic, рекомендую изучить мою "Pydantic 2: Полное руководство для Python-разработчиков — от основ до продвинутых техник" статью, где я подробно разобрал его возможности.

Подготовка конфигурации SQLAlchemy

Начнем с создания файла конфигурации для SQLAlchemy. Он будет находиться в папке app/dao, которую необходимо предварительно создать. Сам файл назовем database.py и заполним следующим содержимым:

import uuid
from datetime import datetime
from decimal import Decimal
from sqlalchemy import inspect, TIMESTAMP, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncAttrs, async_sessionmaker, create_async_engine, AsyncSession
from app.config import settings

engine = create_async_engine(url=settings.DB_URL)
async_session_maker = async_sessionmaker(engine, class_=AsyncSession)


class Base(AsyncAttrs, DeclarativeBase):
    __abstract__ = True

    created_at: Mapped[datetime] = mapped_column(
        TIMESTAMP,
        server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        TIMESTAMP,
        server_default=func.now(),
        onupdate=func.now()
    )

    def to_dict(self, exclude_none: bool = False):
        """
        Преобразует объект модели в словарь.

        Args:
            exclude_none (bool): Исключать ли None значения из результата

        Returns:
            dict: Словарь с данными объекта
        """
        result = {}
        for column in inspect(self.__class__).columns:
            value = getattr(self, column.key)

            # Преобразование специальных типов данных
            if isinstance(value, datetime):
                value = value.isoformat()
            elif isinstance(value, Decimal):
                value = float(value)
            elif isinstance(value, uuid.UUID):
                value = str(value)

            # Добавляем значение в результат
            if not exclude_none or value is not None:
                result[column.key] = value

        return result

Этот файл выполняет несколько важных задач:

• Создает движок для взаимодействия с базой данных.

• Настраивает асинхронную фабрику сессий (async_session_maker).

• Определяет базовый класс моделей, от которого будут наследоваться все таблицы.

• Добавляет две универсальные колонки: created_at и updated_at, которые автоматически фиксируют время создания и обновления записей.

• Реализует метод to_dict(), который позволяет конвертировать объекты SQLAlchemy в обычные Python-словари.

Если вы уже знакомы с моими предыдущими статьями, эта структура не должна вызывать вопросов.

Описание моделей базы данных

Теперь мы готовы описать основные модели таблиц. Для этого создадим файл app/dao/models.py и добавим в него следующий код:

from datetime import datetime
from sqlalchemy import BigInteger, String
from app.dao.database import Base
from sqlalchemy import Integer, Date, ForeignKey
from sqlalchemy.orm import relationship, Mapped, mapped_column


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
    username: Mapped[str | None]
    first_name: Mapped[str | None]
    last_name: Mapped[str | None]
    bookings: Mapped[list["Booking"]] = relationship("Booking", back_populates="user")


class Table(Base):
    __tablename__ = "tables"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    capacity: Mapped[int]
    description: Mapped[str | None]
    bookings: Mapped[list["Booking"]] = relationship("Booking", back_populates="table")


class TimeSlot(Base):
    __tablename__ = "time_slots"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    # Используем String для времени, так как SQLite не имеет встроенного типа Time
    start_time: Mapped[str] = mapped_column(String(5), nullable=False)  # формат: HH:MM
    end_time: Mapped[str] = mapped_column(String(5), nullable=False)  # формат: HH:MM

    bookings: Mapped[list["Booking"]] = relationship(
        "Booking",
        back_populates="time_slot",
        cascade="all, delete-orphan"
    )

    def __repr__(self) -> str:
        return f"TimeSlot(id={self.id}, {self.start_time}-{self.end_time})"


class Booking(Base):
    __tablename__ = "bookings"

    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    user_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("users.id"))
    table_id: Mapped[int] = mapped_column(Integer, ForeignKey("tables.id"))
    time_slot_id: Mapped[int] = mapped_column(Integer, ForeignKey("time_slots.id"))
    date: Mapped[datetime] = mapped_column(Date)
    status: Mapped[str]
    user: Mapped["User"] = relationship("User", back_populates="bookings")
    table: Mapped["Table"] = relationship("Table", back_populates="bookings")
    time_slot: Mapped["TimeSlot"] = relationship("TimeSlot", back_populates="bookings")

Разбираем структуру данных

В проекте у нас есть несколько ключевых сущностей:

1. Пользователи (User) – информация о клиентах, которые совершают бронирования.

2. Столы (Table) – описание столиков, доступных для бронирования.

3. Временные слоты (TimeSlot) – промежутки времени, в которые можно забронировать стол.

4. Бронирования (Booking) – связывают пользователей, столы и временные слоты.

Некоторые важные моменты:

• Столы и временные слоты – статические данные.
  

  Например, описание стола может выглядеть так:

  { "id": 1, "capacity": 2, "description": "Уютный столик для двоих у окна" }

  Айди будет соответствовать номеру стола в заведении.

• Временные слоты фиксированы.

  { "id": 1, "start_time": "06:00", "end_time": "08:00" }

  В системе будет всего 9 таких записей, так как ресторан работает до полуночи, а каждый слот рассчитан на 2 часа.

• Связывание данных
  В таблице Booking мы объединяем пользователей, столы и временные интервалы, что позволяет эффективно управлять процессом бронирования.

Теперь, когда у нас есть описанные модели, мы можем переходить к следующему этапу — настройке работы с данными через SQLAlchemy.

Настройка миграций с Alembic

На данный момент мы только описали модели таблиц, но сами таблицы, как и база данных, еще не созданы. Давайте исправим это с помощью Alembic — инструмента для управления миграциями.

Инициализация Alembic и создание первой миграции

Сначала переходим в директорию app. В терминале выполняем команду:

cd app

Теперь инициализируем Alembic с поддержкой асинхронной работы с базой данных:

alembic init -t async migration

После выполнения команды в проекте появится папка migration и файл alembic.ini. Для удобства работы переместите alembic.ini в корневую директорию проекта.

Настройка файла alembic.ini

Открываем alembic.ini и изменяем путь к папке миграций:

Было:

script_location = migration

Стало:

script_location = app/migration

Это позволит запускать миграции из корневой директории проекта без лишних путей.

Подключение к базе данных в env.py

Теперь необходимо обновить файл app/migration/env.py, чтобы Alembic корректно взаимодействовал с нашей базой данных.

Открываем env.py и заменяем его содержимое следующим образом:

Исходный код (было):

import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context

config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = None

Обновленный код (стало):

import asyncio
from logging.config import fileConfig
from sqlalchemy import pool
from sqlalchemy.engine import Connection
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context
from app.config import settings
from app.dao.database import Base
from app.dao.models import User, Table, TimeSlot, Booking

config = context.config
config.set_main_option("sqlalchemy.url", settings.DB_URL)

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata

Мы добавили:

• Подключение к настройкам проекта (settings.DB_URL), чтобы Alembic использовал актуальный URL базы данных.

• Импорт моделей (User, Table, TimeSlot, Booking) для корректного отслеживания изменений.

• Передачу Base.metadata в target_metadata, что позволит Alembic автоматически определять структуру таблиц.

Остальная часть файла остается без изменений.

Создание первой миграции

Теперь можно создать и применить первую миграцию.

1. Возвращаемся в корневую директорию проекта:

cd ../

• Генерируем файл миграции:

alembic revision --autogenerate -m "Initial revision"

• Создаем папку для хранения базы данных:

В корне проекта (на уровень выше app) создаем папку data:

mkdir data

• Применяем миграции и создаем таблицы в базе данных:

alembic upgrade head

После успешного выполнения этой команды в проекте появится файл базы данных с созданными таблицами.

Теперь наша база данных полностью готова к работе, а мы готовы к описанию методов взаимодействия с базой данных.

Универсальный класс BaseDAO для работы с базой данных

Как я уже упоминал, в проекте используется универсальный базовый класс для работы с базой данных через SQLAlchemy. Этот класс позволяет удобно выполнять основные операции с таблицами.

Полный код находится в файле app/dao/base.py, и его общий вид следующий:

from typing import List, TypeVar, Generic, Type
from pydantic import BaseModel
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.future import select
from sqlalchemy import update as sqlalchemy_update, delete as sqlalchemy_delete, func
from loguru import logger
from sqlalchemy.ext.asyncio import AsyncSession

from app.dao.database import Base

T = TypeVar("T", bound=Base)


class BaseDAO(Generic[T]):
    model: Type[T] = None

    def __init__(self, session: AsyncSession):
        self._session = session
        if self.model is None:
            raise ValueError("Модель должна быть указана в дочернем классе")

    async def find_one_or_none_by_id(self, data_id: int):
        try:
            query = select(self.model).filter_by(id=data_id)
            result = await self._session.execute(query)
            record = result.scalar_one_or_none()
            log_message = f"Запись {self.model.__name__} с ID {data_id} {'найдена' if record else 'не найдена'}."
            logger.info(log_message)
            return record
        except SQLAlchemyError as e:
            logger.error(f"Ошибка при поиске записи с ID {data_id}: {e}")
            raise

Полный код этого файла и всего проекта можно найти в моём телеграм-канале «Лёгкий путь в Python». В настоящее время сообщество насчитывает более 2600 участников и продолжает активно расти. Присоединяйтесь — всё совершенно бесплатно!

Ключевые особенности

1. Вынос сессии на глобальный уровень
   Теперь сессию не нужно передавать в каждый класс отдельно, что упрощает работу с базой данных.

2. Гибкость и масштабируемость

   • Базовый класс BaseDAO является обобщенным (Generic), что позволяет использовать его с любыми моделями.

   • Для каждой таблицы создается отдельный дочерний DAO-класс, который наследуется от BaseDAO.

   • Все базовые методы для работы с базой уже реализованы, а если нужно добавить специфичную логику, это можно сделать в дочернем классе.

3. Отсутствие commit в методах
   Коммиты вынесены в middleware AIOGram и зависимости FastAPI, что позволяет централизованно управлять транзакциями.

Как использовать BaseDAO в проекте

Для каждой модели создается отдельный DAO-класс, который наследуется от BaseDAO. Например, DAO для работы с пользователями может выглядеть так:

from app.dao.base import BaseDAO
from app.dao.models import User

class UserDAO(BaseDAO[User]):
    model = User

Теперь UserDAO автоматически получает все базовые методы, такие как find_one_or_none_by_id, и может дополняться уникальными методами, если это необходимо.

Дочерние DAO классы с новыми методами

Теперь подготовим дочерние классы. Опишем их в файле app/dao/dao.py.

Специфические методы нужны будут только под таблицу с бронями столиков, а остальным будет достаточно унаследовать от BaseDao:

from datetime import date, datetime
from typing import Dict
from loguru import logger
from sqlalchemy import select, update, delete, func
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import joinedload
from app.dao.base import BaseDAO
from app.dao.models import User, TimeSlot, Table, Booking


class UserDAO(BaseDAO[User]):
    model = User


class TimeSlotUserDAO(BaseDAO[TimeSlot]):
    model = TimeSlot


class TableDAO(BaseDAO[Table]):
    model = Table

Теперь отдельно рассмотрим дочерний класс модели Booking (таблица с бронями столиков). Там кода будет чуть больше. Сначала покажу полный код, а после некоторые его аспекты разберем более детально.

class BookingDAO(BaseDAO[Booking]):
    model = Booking

    async def check_available_bookings(self,
                                       table_id: int,
                                       booking_date: date,
                                       time_slot_id: int):
        """Проверяет наличие существующих броней для стола на указанную дату и временной слот."""
        try:
            query = select(self.model).filter_by(
                table_id=table_id,
                date=booking_date,
                time_slot_id=time_slot_id
            )
            result = await self._session.execute(query)

            # Если результатов нет, стол свободен
            if not result.scalars().all():
                return True

            # Проверяем статус существующих бронирований
            for booking in result.scalars().all():
                if booking.status == "booked":
                    return False  # Стол занят

                # Для других статусов считаем стол свободным
                continue
            # Если все брони имеют неактивные статусы
            return True

        except SQLAlchemyError as e:
            logger.error(f"Ошибка при проверке доступности брони: {e}")

    async def get_available_time_slots(self, table_id: int, booking_date: date):
        """Получает список доступных временных слотов для стола на указанную дату."""
        try:
            # Получаем все брони для данного стола и даты
            bookings_query = select(self.model).filter_by(
                table_id=table_id,
                date=booking_date
            )
            bookings_result = await self._session.execute(bookings_query)
            # Составляем набор занятых слотов (только с активными бронями)
            booked_slots = {booking.time_slot_id for booking in bookings_result.scalars().all() if
                            booking.status == "booked"}
            # Получаем все доступные слоты, исключая занятые
            available_slots_query = select(TimeSlot).filter(
                ~TimeSlot.id.in_(booked_slots)
            )
            available_slots_result = await self._session.execute(available_slots_query)
            return available_slots_result.scalars().all()
        except SQLAlchemyError as e:
            logger.error(f"Ошибка при получении доступных временных слотов: {e}")

    async def get_bookings_with_details(self, user_id: int):
        """
        Получает список всех бронирований пользователя с полной информацией о столике и временном слоте.

        :param user_id: ID пользователя, брони которого нужно получить.
        :return: Список объектов Booking с загруженными данными о столе и времени.
        """
        try:
            query = select(self.model).options(
                joinedload(self.model.table),
                joinedload(self.model.time_slot)
            ).filter_by(user_id=user_id)
            result = await self._session.execute(query)
            return result.scalars().all()
        except SQLAlchemyError as e:
            logger.error(f"Ошибка при получении бронирований с деталями: {e}")
            return []

    async def complete_past_bookings(self):
        """
        Обновляет статус бронирований на 'completed', если дата и время бронирования уже прошли.
        """
        try:
            # Получаем текущее время
            now = datetime.now()
            subquery = select(TimeSlot.start_time).where(TimeSlot.id == self.model.time_slot_id).scalar_subquery()
            query = select(Booking.id).where(
                Booking.date < now.date(),
                self.model.status == "booked"
            ).union_all(
                select(Booking.id).where(
                    self.model.date == now.date(),
                    subquery < now.time(),
                    self.model.status == "booked"
                )
            )

            # Выполняем запрос и получаем id бронирований, которые нужно обновить
            result = await self._session.execute(query)
            booking_ids_to_update = result.scalars().all()

            if booking_ids_to_update:
                # Формируем запрос на обновление статуса бронирований
                update_query = update(Booking).where(
                    Booking.id.in_(booking_ids_to_update)
                ).values(status="completed")

                # Выполняем запрос на обновление
                await self._session.execute(update_query)

                # Подтверждаем изменения
                await self._session.commit()

                logger.info(f"Обновлен статус для {len(booking_ids_to_update)} бронирований на 'completed'")
            else:
                logger.info("Нет бронирований для обновления статуса.")

        except SQLAlchemyError as e:
            logger.error(f"Ошибка при обновлении статуса бронирований: {e}")
            await self._session.rollback()

    async def cancel_book(self, book_id: int):
        try:
            query = (
                update(self.model)
                .filter_by(id=book_id)
                .values(status="canceled")
                .execution_options(synchronize_session="fetch")
            )
            result = await self._session.execute(query)
            await self._session.flush()
            return result.rowcount
        except SQLAlchemyError as e:
            logger.error(f"Ошибка при отмене книги с ID {book_id}: {e}")
            await self._session.rollback()
            raise

    async def delete_book(self, book_id: int):
        try:
            query = delete(self.model).filter_by(id=book_id)
            result = await self._session.execute(query)
            logger.info(f"Удалено {result.rowcount} записей.")
            await self._session.flush()
            return result.rowcount
        except SQLAlchemyError as e:
            logger.error(f"Ошибка при удалении записей: {e}")
            raise

    async def book_count(self) -> Dict[str, int]:
        """
        Подсчитывает количество заявок по каждому статусу (booked, completed, canceled).
        """
        try:
            status_counts = {}
            statuses = ["booked", "completed", "canceled"]

            for status in statuses:
                query = select(func.count(self.model.id)).where(self.model.status == status)
                result = await self._session.execute(query)
                count = result.scalar()
                status_counts[status] = count
                logger.info(f"Найдено {count} заявок со статусом '{status}'.")

            total_query = select(func.count(self.model.id))
            total_result = await self._session.execute(total_query)
            total_count = total_result.scalar()
            status_counts['total'] = total_count
            logger.info(f"Всего найдено {total_count} заявок.")

            return status_counts
        except SQLAlchemyError as e:
            logger.error(f"Ошибка при подсчете заявок по статусам: {e}")
            raise

Описание методов BookingDAO

• check_available_bookings(table_id, booking_date, time_slot_id)

Проверяет, свободен ли столик на указанную дату и временной слот.

• get_available_time_slots(table_id, booking_date)

Возвращает список доступных временных слотов для бронирования конкретного столика на выбранную дату.

• get_bookings_with_details(user_id)

Получает список всех бронирований пользователя с информацией о столике и времени бронирования.

• complete_past_bookings()

Обновляет статус всех прошедших бронирований на "completed", если их время уже истекло.

• cancel_book(book_id)

Меняет статус бронирования на "canceled".

• delete_book(book_id)

Удаляет запись о бронировании из базы данных.

• book_count()

Подсчитывает количество бронирований по статусам ("booked", "completed", "canceled") и общее число записей.

Так-же в полном исходнике кода вы можете найти логику по автоматическому заполнению таблиц базы данных информацией из JSON файлов. Сейчас на это время тратить не будем.

Теперь мы уверенно можем сказать, что логика взаимодействия с базой данных у нас полностью закрыта и мы можем переходить к тому из-за чего большинство из вас тут собралось – описание логики бронирования через Aiogram dialog.

Микросервис бронирования на Aiogram Dialog

Сейчас мы готовы к тому, чтобы описать логику бронирования столиков с использованием Aiogram Dialog.

Для начала давайте создадим пакет app/bot. В этом пакете (папка с файлом init.py) мы будем описывать логику, которая будет иметь прямое отношение к нашему телеграм-боту.

Напоминаю, что Aiogram Dialog имеет следующие основные сущности:

1. Функции обратного вызова

2. Функции-геттеры

3. Виджеты

4. Окна

5. Сам диалог

И работать это всё может только в рамках состояний (state).

Далее выстраивается следующая логика: мы собираем из виджетов и функций окно, используя принцип вложенного конструктора. После этого мы объединяем их в окно, а окно уже является кирпичиком для создания конкретного диалога.

Создадим в папке app/bot/ папку с именем booking и там уже сосредоточимся на структуре. Для удобства предлагаю использовать следующую структуру файлов:

booking/
  - dialog.py: тут мы объединим все окна в один диалог
  - state.py: тут мы опишем класс с нашими состояниями
  - schemas.py: Pydantic-схемы
  - windows.py: функции для создания окон (каждая функция будет включать в себя разный набор виджетов, геттеров и т.д.)
  - getters.py: функции-геттеры
  - handlers.py: функции обратного вызова (обработчики)

Такая структура позволит не только удобно организовать код, но и даст общее понимание философии Aiogram Dialog на практике.

Сразу опишем класс с состояниями в booking/state.py:

from aiogram.fsm.state import StatesGroup, State


class BookingState(StatesGroup):
    count = State()
    table = State()
    booking_date = State()
    booking_time = State()
    confirmation = State()
    success = State()

Тут используется стандартный синтаксис Aiogram 3.

Затем опишем схемы Pydantic в файле booking/schemas.py:

from datetime import date
from pydantic import BaseModel


class SCapacity(BaseModel):
    capacity: int


class SNewBooking(BaseModel):
    user_id: int
    table_id: int
    time_slot_id: int
    date: date
    status: str

Они нам понадобятся для взаимодействия с базовыми методами из BaseDAO.

Теперь опишем геттеры. Это специальные функции, которые нужны для отображения информации в окнах нашего диалога. Опишем их в файле booking/getters.py:

from aiogram_dialog import DialogManager


async def get_all_tables(dialog_manager: DialogManager, **kwargs):
    """Получение списка столов с учетом выбранной вместимости."""
    tables = dialog_manager.dialog_data['tables']
    capacity = dialog_manager.dialog_data['capacity']
    return {"tables": [table.to_dict() for table in tables],
            "text_table": f'Всего для {capacity} человек найдено {len(tables)} столов. Выберите нужный по описанию'}


async def get_all_available_slots(dialog_manager: DialogManager, **kwargs):
    """Получение списка доступных временных слотов для выбранного стола и даты."""
    selected_table = dialog_manager.dialog_data["selected_table"]
    slots = dialog_manager.dialog_data["slots"]
    text_slots = (
        f'Для стола №{selected_table.id} найдено {len(slots)} '
        f'{"свободных слотов" if len(slots) != 1 else "свободный слот"}. '
        'Выберите удобное время'
    )
    return {"slots": [slot.to_dict() for slot in slots], "text_slots": text_slots}


async def get_confirmed_data(dialog_manager: DialogManager, **kwargs):
    """Получение данных для подтверждения бронирования."""
    selected_table = dialog_manager.dialog_data['selected_table']
    booking_date = dialog_manager.dialog_data['booking_date']
    selected_slot = dialog_manager.dialog_data['selected_slot']

    confirmed_text = (
        "<b>