Привет, друзья!

Сегодня я продолжу делиться примерами создания приложений с использованием MiniApp в Telegram, и на этот раз мы создадим настоящую классику — головоломку 2048, полностью интегрированную в Telegram MiniApp.

Что мы будем делать?

В этой статье шаг за шагом разработаем проект, где FastAPI возьмет на себя все основные задачи:

• Обслуживание статики (JavaScript, стили);

• Рендеринг HTML-страниц;

• Настройка вебхука для бота;

• Создание API для взаимодействия с игрой.

Если вы следите за моими публикациями на Хабре, то, возможно, видели другие похожие проекты:

• Создание Telegram Web App с FastAPI: Генерация и сканирование QR-кодов за 5 минут.
  В этом проекте я показал, как использовать камеру устройства (смартфона или вебкамеру) для сканирования QR-кодов в MiniApp.

• Telegram Web App, FastAPI и вебхуки в одном приложении: Бот для приема заявок.
  Здесь мы создали пример для парикмахерской, обрабатывающий заявки от пользователей через MiniApp.

Сегодняшний проект: целостный подход с FastAPI

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

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

• SQLAlchemy для работы с базой данных

• Alembic для миграций и работы со структурой таблиц

• AioSQLite для асинхронного взаимодействия с SQLite (в качестве основного движка SQLAlchemy)

• Aiogram 3 для работы бота,

• FastAPI для всех вышеперечисленных задач.

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

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

• Подготовка: создаем токен для бота, настраиваем домен для вебхуков и запускаем MiniApp.

• Настройка проекта: создаем виртуальное окружение, устанавливаем зависимости и подготавливаем структуру проекта.

• Подготовка игры: берем готовую версию головоломки 2048, созданную 10 лет назад, и дорабатываем её для интеграции с FastAPI. Этот шаг закладывает прочную основу для написания кода и методов.

• Настройка SQLAlchemy: настраиваем SQLAlchemy для работы с базой данных.

• Миграции с Alembic: создаем и применяем миграции для базы данных с помощью Alembic.

• Методы для работы с базой данных: пишем функции для взаимодействия с базой данных через SQLAlchemy.

• Создание Telegram-бота: разрабатываем чистый Telegram-бот на Aiogram 3.

• Интеграция бота с FastAPI: соединяем Telegram-бота и FastAPI в одном проекте.

• Дополнительные функции и страницы для игры: пишем новые методы и создаем страницы для улучшения игрового процесса. В частности, мы создадим страницу со списком рекордсменов (топ-20), добавим функционал для очистки лучшего результата и прочее.

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

• Связывание компонентов: интегрируем все созданные элементы в единую систему.

• Деплой на Amvera Cloud: размещаем и настраиваем проект на Amvera Cloud для стабильной работы в сети.

Почему Amvera Cloud?

Для финального деплоя я выбрал Amvera Cloud — платформу, которая позволяет быстро развернуть проект без лишних настроек. Преимущества:

• Бесплатный HTTPS-домен, что упрощает настройку вебхуков для Telegram;

• Простая развертка — достаточно указать версию Python и команду запуска в конфигурации;

• Гибкие способы загрузки — через веб-интерфейс или Git.

Работы предстоит немало, так что давайте начнем!

Подготовка проекта

Для разработки Telegram-бота с WebApp и вебхуками необходимо обеспечить приложению доступ к глобальной сети. Сделать это можно с помощью туннелей, например, с использованием Ngrok. Мы рассмотрим настройку туннеля на Windows, хотя также подойдут и другие сервисы, такие как LocalTunnel, Xtunnel или Tuna.

Как работает туннель?

Принцип туннелирования прост: сначала запускаем наше FastAPI-приложение на локальном порте (например, 8000), затем открываем туннель к этому порту, чтобы получить временный HTTPS-домен. Этот домен будет основным URL-адресом для взаимодействия бота с вебхуками.

Шаги для настройки Ngrok

1. Регистрация на сайте Ngrok:
   Зайдите на официальный сайт Ngrok, зарегистрируйтесь и войдите в свой аккаунт.

2. Загрузка и установка Ngrok:
   Скачайте подходящую версию Ngrok для вашей операционной системы и распакуйте файл.

3. Добавление токена авторизации:
   Настройте Ngrok для вашего аккаунта, выполнив команду с токеном авторизации, который можно найти в личном кабинете Ngrok:

   ngrok config add-authtoken ваш_токен

4. Запуск туннеля:
   Укажите порт, на котором работает ваше FastAPI-приложение (например, 8000), и запустите туннель:

   ngrok http 8000

   Если всё настроено верно, в окне терминала отобразится временный HTTPS-домен. Этот адрес будет использоваться для настройки вебхуков и подключения MiniApp в Telegram.

Настройка Telegram-бота с поддержкой MiniApp

Чтобы привязать созданный туннель и FastAPI-приложение к Telegram-боту, выполните несколько шагов.

1. Создание бота через BotFather

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

2. Отправьте команду /newbot, чтобы создать нового бота.

3. Укажите имя бота (можно на русском).

4. Задайте уникальный логин (на латинице, должен оканчиваться на bot, BOT или Bot).

2. Подключение MiniApp

1. В интерфейсе Telegram перейдите в настройки созданного бота.

2. Выберите опцию Configure MiniApp.

3. Включите MiniApp, нажав Enable MiniApp.

4. Введите сгенерированную ссылку Ngrok в поле URL. После завершения разработки и деплоя на Amvera эту ссылку можно будет заменить на постоянный адрес.

3. Добавление MiniApp в меню команд (опционально)

1. В настройках бота найдите раздел Menu Button.

2. Укажите текст кнопки для быстрого доступа к MiniApp (например, «2048»).

3. Сохраните настройки — теперь пользователи смогут одним нажатием открыть MiniApp из меню.

Теперь бот готов к работе, а туннель Ngrok настроен. Используйте полученный HTTPS-домен как временный URL для подключения MiniApp и настройки вебхуков. Затем, на этапе деплоя в Amvera Cloud, мы заменим эту ссылку на бесплатный домен, который нам подарит Amvera.

Настройка проекта

Для начала разработки откройте свою среду разработки (например, PyCharm) и создайте новый проект. В корне проекта создайте следующую структуру файлов и папок:

проект/
│
├── .env                   # файл для хранения переменных окружения
├── requirements.txt       # файл со списком зависимостей
│
└── app/                   # основная папка с кодом приложения
    ├── bot/               # папка для логики и файлов, связанных с Telegram-ботом
    │
    ├── game/              # папка для логики и файлов, связанных с головоломкой 2048
    │
    ├── static/            # папка для статических файлов (CSS, JavaScript, изображения и др.)
    │
    ├── templates/         # папка для HTML-шаблонов
    ├── config.py          # файл для конфигурации приложения (настройки базы данных и т.д.)
    ├── main.py            # точка входа для FastAPI-приложения
    └── database.py        # файл для работы с базой данных (подключение, модели, настройки)
    
└── data/                  # папка для хранения файла базы данных SQLITE

Папки будут постепенно заполняться файлами, поэтому сейчас сосредоточимся на ключевых элементах: файле .env, requirements.txt и файле конфигурации app/config.py.

Файл .env

BOT_TOKEN=bot_token
ADMIN_IDS=[TelegramIDAmin1, TelegramIDAmin2]
BASE_SITE=https://ngrok_url.ng

Здесь хранятся:

• токен бота,

• список Telegram ID администраторов (если вы хотите, чтобы администратором были только вы, добавьте только свой ID в список),

• URL-адрес, сгенерированный Ngrok.

Для получения Telegram ID любого человека, группы или канала можно воспользоваться ботом IDBot Finder Pro.

Файл requirements.txt

aiogram==3.13.1
fastapi==0.115.0
pydantic==2.9.2
uvicorn==0.31.0
jinja2==3.1.4
pydantic_settings==2.5.2
aiosqlite==0.20.0
alembic==1.13.3
SQLAlchemy==2.0.35

Описание зависимостей

1. aiogram==3.13.1 — библиотека для разработки Telegram‑бота. В проекте она будет использоваться для обработки запросов от Telegram API и взаимодействия бота с пользователями.

2. fastapi==0.115.0 — веб‑фреймворк для создания API. FastAPI используется как основа всего приложения, обрабатывая HTTP‑запросы, выполняя маршрутизацию и поддерживая взаимодействие игры с фронтендом.

3. pydantic==2.9.2 — библиотека для валидации и управления данными. Она помогает создавать схемы данных, используемые в API и валидации данных, поступающих от пользователя.

4. uvicorn==0.31.0 — ASGI‑сервер для запуска FastAPI‑приложения. Uvicorn необходим для выполнения приложения на локальном сервере и тестирования API перед деплоем.

5. jinja2==3.1.4 — движок шаблонов, используемый для рендеринга HTML‑страниц. В проекте Jinja2 будет генерировать интерфейс игры 2048 и другие веб‑страницы.

6. pydantic_settings==2.5.2 — расширение Pydantic для удобной работы с конфигурациями, в том числе с переменными окружения из.env файла.

7. aiosqlite==0.20.0 — асинхронная библиотека для работы с базой данных SQLite. Она будет использована в качестве асинхронного движка в SQLAlchemy для работы с SQLite.

8. alembic==1.13.3 — инструмент для управления миграциями базы данных. Alembic будет помогать отслеживать изменения структуры базы и синхронизировать их с кодом.

9. SQLAlchemy==2.0.35 — ORM для взаимодействия с базой данных, благодаря которой можно работать с SQL‑запросами через Python‑код, управляя таблицами и данными в объектном формате.

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

pip install -r requirements.txt

Файл конфигурации app/config.py

import os
from typing import List
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    BOT_TOKEN: str
    ADMIN_IDS: List[int]
    DB_URL: str = 'sqlite+aiosqlite:///data/db.sqlite3'
    BASE_SITE: str

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

    def get_webhook_url(self) -> str:
        """Возвращает URL вебхука с кодированием специальных символов."""
        return f"{self.BASE_SITE}/webhook"

      
# Получаем параметры для загрузки переменных среды
settings = Settings()
database_url = settings.DB_URL

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

Подготовка игры 2048

Как уже говорилось в начале статьи, для экономии времени мы не будем создавать игру 2048 с нуля, а воспользуемся готовым проектом. Я взял его здесь. Это старый проект, написанный более 10 лет назад на чистом JavaScript, HTML и CSS, без интеграции с FastAPI и поддержки MiniApp. Тем не менее, у него есть гибкая анимация, сохранение результатов в локальном хранилище и приятный интерфейс — всё, что нужно для нашего учебного проекта.

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

Адаптация игры под проект на FastAPI

1. Для клонирования репозитория используем команду:

   git clone https://github.com/edopedia/2048

2. Переносим скачанные файлы:

   • Папки js, meta и style перемещаем в app/static.

   • Файл index.html помещаем в app/templates.

3. Обновляем пути к статическим файлам в index.html, добавляя /static в пути к CSS и JavaScript. Например:

   <link href="style/main.css" rel="stylesheet" type="text/css">

   заменяем на:

   <link href="/static/style/main.css" rel="stylesheet" type="text/css">

   Изменения вносим для всех ссылок на статические файлы.

4. Создаем эндпоинт FastAPI для рендеринга страницы игры. В app/game создаем файл router.py со следующим содержимым:

   from fastapi import APIRouter, Request
   from fastapi.responses import HTMLResponse
   from fastapi.templating import Jinja2Templates
   
   
   router = APIRouter(prefix='', tags=['ИГРА'])
   templates = Jinja2Templates(directory='app/templates')
   
   
   @router.get("/", response_class=HTMLResponse)
   async def read_root(request: Request):
       return templates.TemplateResponse("index.html", {"request": request})

   Этот эндпоинт рендерит HTML-шаблон, расположенный в app/templates/index.html, обрабатывая GET-запросы к корневому маршруту (/).

5. Настраиваем главный файл приложения (app/main.py) для поддержки статических файлов и подключаем роутер игры:

   from fastapi import FastAPI
   from fastapi.staticfiles import StaticFiles
   from app.game.router import router as game_router
   
   app = FastAPI()
   
   # Монтируем статические файлы
   app.mount('/static', StaticFiles(directory='app/static'), 'static')
   
   # Подключаем роутер игры
   app.include_router(game_router)
   

   • Строка app.mount('/static', StaticFiles(directory='app/static'), 'static') настраивает маршрут /static для доступа к статическим файлам.

   • Строка app.include_router(game_router) подключает маршруты из game_router в основное приложение, организуя логику игры в отдельном модуле.

6. Запускаем FastAPI-приложение:

   uvicorn app.main:app --reload

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

Добавляем функционал для игры

Для расширения функциональности, совсем скоро, создадим следующие возможности:

1. Регистрация пользователя в базе данных.

2. Сохранение лучшего результата игрока.

3. Получение позиции пользователя в турнирной таблице.

4. Вывод топ-20 лучших игроков.

5. Создание страницы с таблицей лидеров.

Также нам предстоит интегрировать игру в Telegram-бота. Для реализации этого функционала подготовим базу данных и методы для работы с ней, используя SQLAlchemy, Aiosqlite и Alembic для миграций.

Настройка базы данных с использованием SQLAlchemy и Alembic

Шаг 1: Создание файла app/database.py

В этом файле мы пропишем основные настройки для работы с базой данных:

from datetime import datetime
from sqlalchemy import func, TIMESTAMP, Integer
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase
from sqlalchemy.ext.asyncio import AsyncAttrs, async_sessionmaker, create_async_engine, AsyncSession

from app.config import database_url

# Создание асинхронного движка для подключения к базе данных
engine = create_async_engine(url=database_url)
async_session_maker = async_sessionmaker(engine, class_=AsyncSession)


class Base(AsyncAttrs, DeclarativeBase):
    __abstract__ = True  # Абстрактный базовый класс, чтобы избежать создания отдельной таблицы

    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=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()
    )

    @classmethod
    @property
    def __tablename__(cls) -> str:
        return cls.__name__.lower() + 's'

• engine — асинхронный движок для подключения к базе данных по database_url, который позволяет работать в неблокирующем режиме.

• async_session_maker — фабрика асинхронных сессий, используется для создания сессий для запросов к базе данных.

• Base — абстрактный класс для моделей ORM. Включает колонки id, created_at и updated_at.

Класс Base будет родительским для всех моделей таблиц.

Шаг 2: Описание модели таблицы

Создадим файл app/game/models.py для описания модели User. Проект будет содержать только одну таблицу, где хранятся данные пользователей Telegram.

from sqlalchemy.orm import Mapped, mapped_column
from sqlalchemy import BigInteger
from typing import Optional
from app.database import Base


class User(Base):
    telegram_id: Mapped[int] = mapped_column(BigInteger, unique=True, nullable=False)
    username: Mapped[Optional[str]]
    first_name: Mapped[Optional[str]]
    last_name: Mapped[Optional[str]]
    best_score: Mapped[int] = mapped_column(default=0)

Класс User наследует Base, поэтому нет необходимости заново определять id, created_at и updated_at.

Колонка best_score имеет default=0, то есть по умолчанию будет подставляться 0 на стороне приложения. Чтобы это происходило на стороне базы данных, можно использовать server_default.

Шаг 3: Интеграция с Alembic

Для управления миграциями воспользуемся Alembic.

1. Переходим в папку app:

   cd app

2. Инициализируем Alembic с поддержкой асинхронного взаимодействия:

   alembic init -t async migration

3. Переносим файл alembic.ini в корень проекта и заменяем строку:

   script_location = migration

   на

   script_location = app/migration

4. Вносим изменения в файл app/migration/env.py для настройки Alembic.

Исходный код:

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 sys
from os.path import dirname, abspath

sys.path.insert(0, dirname(dirname(abspath(__file__))))

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 database_url
from app.database import Base
from app.game.models import User

config = context.config
config.set_main_option("sqlalchemy.url", database_url)
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

target_metadata = Base.metadata

5. Создаем первую миграцию:

   alembic revision --autogenerate -m "Initial revision"

6. Применяем миграцию:

   alembic upgrade head

Если все прошло успешно, в корне проекта появится файл db.sqlite3 с таблицей users, включающей все необходимые поля.

Шаг 4: Написание методов для работы с базой данных

Создайте файл app/game/dao.py, в котором будет описан класс с методами для работы с базой данных, реализующими все необходимые операции.

Методы работы с базой данных

Для полного понимания того, что здесь происходит, настоятельно рекомендую ознакомиться с моими статьями:

1. Асинхронный SQLAlchemy 2: простой пошаговый гайд по настройке, моделям, связям и миграциям с использованием Alembic

2. Асинхронный SQLAlchemy 2: пошаговый гайд по управлению сессиями, добавлению и извлечению данных с Pydantic

Подготовка класса

from pydantic import BaseModel
from sqlalchemy import select, desc, func
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import Base
from app.game.models import User


class UserDAO(Base):
    model = User

Класс UserDAO (Data Access Object) служит для управления доступом к данным модели User в базе данных. Он наследуется от Base, что позволяет использовать основные функции для работы с базой данных и моделями SQLAlchemy.

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

• Проверка, существует ли пользователь

• Добавление пользователя в базу данных

• Получение топ-20 пользователей с лучшим результатом

• Получение места в топе для конкретного пользователя

Начнем с метода для проверки существования пользователя.

Метод проверки существования пользователя

@classmethod
async def find_one_or_none(cls, session: AsyncSession, filters: BaseModel):
    # Найти одну запись по фильтрам
    filter_dict = filters.model_dump(exclude_unset=True)
    try:
        query = select(cls.model).filter_by(**filter_dict)
        result = await session.execute(query)
        record = result.scalar_one_or_none()
        return record
    except SQLAlchemyError as e:
        raise

Здесь представлен обновленный подход к написанию методов, и поэтому я подробно его объясню.

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

Кроме того, фильтры теперь передаются через Pydantic, а не через распакованный словарь **filters, как было раньше. Теперь мы создаем Pydantic-модель и передаем ее в качестве значения filters. О том, что такое Pydantic и как с ним работать, я подробно писал в статье «Pydantic 2: Полное руководство для Python-разработчиков — от основ до продвинутых техник».

Описание метода:

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

• Формируется стандартный запрос, и мы получаем информацию о пользователе или None, если пользователь не найден.

Подробнее читайте в статьях про SQLAlchemy.

Метод для добавления пользователя в базу данных

@classmethod
async def add(cls, session: AsyncSession, values: BaseModel):
    # Добавить одну запись
    values_dict = values.model_dump(exclude_unset=True)
    new_instance = cls.model(**values_dict)
    session.add(new_instance)
    try:
        await session.commit()
    except SQLAlchemyError as e:
        await session.rollback()
        raise e
    return new_instance

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

Для сохранения результата в базе данных используется session.commit(). В случае ошибки мы откатываем изменения с помощью session.rollback() и выбрасываем исключение, чтобы обработать ошибку на уровне вызова метода.

Метод для получения топ-20 игроков

@classmethod
async def get_top_scores(cls, session: AsyncSession, limit: int = 20):
    """
    Получить топ рекордов, отсортированных от самого высокого к низкому, с добавлением номера позиции.
    """
    try:
        query = (
            select(cls.model.telegram_id, cls.model.first_name, cls.model.best_score)
            .order_by(desc(cls.model.best_score))
            .limit(limit)
        )
        result = await session.execute(query)
        records = result.fetchall()

        # Добавление поля `rank` для нумерации позиций
        ranked_records = [
            {"rank": index + 1, "telegram_id": record.telegram_id, "first_name": record.first_name,
             "best_score": record.best_score}
            for index, record in enumerate(records)
        ]

        return ranked_records
    except SQLAlchemyError as e:
        raise e

Объяснение метода:

• В select мы указываем, какие значения колонок хотим получить. Для отображения топа пользователей достаточно telegram_id, first_name и best_score.

• Сортировка выполняется функцией desc, которая упорядочивает пользователей по best_score в порядке убывания. Параметр limit ограничивает список до 20 пользователей.

• Мы добавляем rank — место в турнирной таблице — на стороне приложения. Это удобно и делает запрос проще.

• Возвращается список словарей с данными о пользователях и их позициями.

Метод для получения места пользователя в списке рекордов

@classmethod
async def get_user_rank(cls, session: AsyncSession, telegram_id: int):
    """
    Получить место пользователя по telegram_id в списке рекордов.
    Возвращает словарь с полями rank и best_score.
    """
    try:
        # Подзапрос для вычисления рангов на основе best_score
        rank_subquery = (
            select(
                cls.model.telegram_id,
                cls.model.best_score,
                func.rank().over(order_by=desc(cls.model.best_score)).label("rank")
            )
            .order_by(desc(cls.model.best_score))
            .subquery()
        )

        # Запрос для получения ранга и best_score конкретного пользователя
        query = select(rank_subquery.c.rank, rank_subquery.c.best_score).where(
            rank_subquery.c.telegram_id == telegram_id
        )
        result = await session.execute(query)
        rank_row = result.fetchone()

        # Возвращаем словарь с рангом и лучшим результатом
        return {"rank": rank_row.rank, "best_score": rank_row.best_score} if rank_row else None
    except SQLAlchemyError as e:
        raise e

Описание метода get_user_rank:

• Этот метод делает не просто выборку данных, а формирует динамическое значение — ранг пользователя — на стороне базы данных.

1. Подзапрос для ранжирования (rank_subquery):

• Оконная функция rank() вычисляет позиции пользователей, отсортированных по их best_score.

• Это позволяет базе данных определить ранг пользователя на основе его best_score, что значительно эффективнее, чем делать это на стороне приложения.

2. Основной запрос:

• Основной запрос извлекает конкретного пользователя по telegram_id, возвращая его ранг и лучший результат (best_score).

3. Обработка результата:

• Если пользователь найден, метод возвращает словарь с его рангом и результатом. В случае отсутствия пользователя возвращается None.

4. Обработка ошибок:

• В случае ошибки SQLAlchemy выбрасывается исключение SQLAlchemyError, что позволяет централизованно обработать любые неполадки с базой данных.

Метод get_user_rank эффективно использует возможности базы данных для динамического ранжирования, минимизируя нагрузку на приложение и упрощая вычисления.

Я планирую подробно рассказать о том, как это все работает, в одной из своих будущих статей про SQLAlchemy.

Обновление лучшего результата пользователя

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

Пишем бота

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

Писать код будем в папке app/bot, а после этого подключим бота к FastAPI-приложению в файле main.py.

Создание файла create_bot.py

В папке bot создаем файл create_bot.py. В этом файле мы инициализируем два главных объекта для разработки ботов через aiogram 3: bot и dispatcher, а также пропишем функции, которые будут выполняться при запуске и завершении работы бота.

from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode

from app.config import settings

bot = Bot(token=settings.BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher()

async def start_bot():
    try:
        for admin_id in settings.ADMIN_IDS:
            await bot.send_message(admin_id, 'Я запущен?.')
    except Exception:
        pass

async def stop_bot():
    try:
        for admin_id in settings.ADMIN_IDS:
            await bot.send_message(admin_id, 'Бот остановлен. За что??')
    except Exception:
        pass

Здесь нам пригодился список администраторов ADMIN_IDS, который мы добавили в config.py.

Подготовка клавиатур для бота

Мы создадим две inline-клавиатуры. Для их описания создаем папку bot/keyboard и внутри нее файл kbs.py, в котором опишем клавиатуры.

from aiogram.types import InlineKeyboardMarkup, WebAppInfo
from aiogram.utils.keyboard import InlineKeyboardBuilder
from app.config import settings

def main_keyboard() -> InlineKeyboardMarkup:
    kb = InlineKeyboardBuilder()
    kb.button(text="? Старт игры 2048", web_app=WebAppInfo(url=settings.BASE_SITE))
    kb.button(text="? Лидеры 2048", web_app=WebAppInfo(url=f"{settings.BASE_SITE}/records"))
    kb.button(text="? Мой рекорд", callback_data="show_my_record")
    kb.adjust(1)
    return kb.as_markup()

def record_keyboard() -> InlineKeyboardMarkup:
    kb = InlineKeyboardBuilder()
    kb.button(text="? Старт игры 2048", web_app=WebAppInfo(url=settings.BASE_SITE))
    kb.button(text="? Рекоды других", web_app=WebAppInfo(url=f"{settings.BASE_SITE}/records"))
    kb.button(text="? Обновить мой рекорд", callback_data="show_my_record")
    kb.adjust(1)
    return kb.as_markup()

Мы использовали InlineKeyboardBuilder для простого создания inline-клавиатур. Наши клавиатуры включают как кнопки с callback_data (для просмотра своего рекорда и места в рейтинге, а также обновления рекорда на стороне Telegram), так и кнопки с WebAppInfo (ссылки на страницы MiniApp).

Описание хендлеров бота

Создадим папку handlers в bot, где создадим файл router.py для описания хендлеров бота.

Перед этим немного изменим файл app/database.py, добавив туда фабрику декораторов для удобного подключения к базе данных.

from functools import wraps

def connection(isolation_level=None):
    def decorator(method):
        @wraps(method)
        async def wrapper(*args, **kwargs):
            async with async_session_maker() as session:
                try:
                    # Устанавливаем уровень изоляции, если передан
                    if isolation_level:
                        await session.execute(text(f"SET TRANSACTION ISOLATION LEVEL {isolation_level}"))
                    # Выполняем декорированный метод
                    return await method(*args, session=session, **kwargs)
                except Exception as e:
                    await session.rollback()  # Откатываем сессию при ошибке
                    raise e  # Поднимаем исключение дальше
                finally:
                    await session.close()  # Закрываем сессию

        return wrapper
    return decorator

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

Для работы с методами взаимодействия с базой данных мы будем использовать модели Pydantic. Поэтому создадим в папке app/game файл schemas.py и опишем необходимые Pydantic-модели.

from pydantic import BaseModel


class TelegramIDModel(BaseModel):
    telegram_id: int

    
class UserModel(TelegramIDModel):
    username: str
    first_name: str
    last_name: str
    best_score: int

Первая схема TelegramIDModel принимает telegram_id пользователя, а вторая схема UserModel наследуется от первой и добавляет поля username, first_name, last_name и best_score.

Настройка роутера бота

Теперь мы можем приступить к описанию роутера бота.

Импортируем необходимые модули и классы:

from aiogram import Router, F
from aiogram.filters import CommandStart
from aiogram.types import Message, CallbackQuery
from app.game.dao import UserDAO
from app.game.schemas import TelegramIDModel, UserModel
from app.bot.keyboards.kbs import main_keyboard, record_keyboard
from app.database import connection
from sqlalchemy.ext.asyncio import AsyncSession

Инициализируем роутер:

router = Router()

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

@router.message(CommandStart())
@connection()
async def cmd_start(message: Message, session: AsyncSession, **kwargs):
    welcome_text = (
        "? Добро пожаловать в игру 2048! ?\n\n"
        "Здесь вы сможете насладиться увлекательной головоломкой и проверить свои навыки. Вот что вас ждёт:\n\n"
        "? Играйте в 2048 и двигайтесь к победе!\n"
        "? Смотрите свой текущий рекорд и стремитесь к новым вершинам\n"
        "? Узнавайте рекорды других игроков и соревнуйтесь за звание лучшего!\n\n"
        "Готовы начать? Будьте лучшим и достигните плитки 2048! ?"
    )

    try:
        user_id = message.from_user.id
        user_info = await UserDAO.find_one_or_none(session=session, filters=TelegramIDModel(telegram_id=user_id))

        if not user_info:
            # Добавляем нового пользователя
            values = UserModel(
                telegram_id=user_id,
                username=message.from_user.username,
                first_name=message.from_user.first_name,
                last_name=message.from_user.last_name,
                best_score=0
            )
            await UserDAO.add(session=session, values=values)

        await message.answer(welcome_text, reply_markup=main_keyboard())

    except Exception as e:
        await message.answer("Произошла ошибка при обработке вашего запроса. Пожалуйста, попробуйте снова позже.")

Описание логики метода:

• В методе, кроме стандартного декоратора message, мы использовали собственный декоратор connection. При использовании connection нужно добавить **kwargs, чтобы избежать ошибок, так как aiogram может передавать служебные параметры в обработчик.

• Если пользователя в базе данных нет, создается новый экземпляр UserModel и сохраняется в базе.

Теперь создадим вторую функцию, которая будет вызываться при переходе в «Мои рекорды»:

@router.callback_query(F.data == 'show_my_record')
@connection()
async def get_user_rating(call: CallbackQuery, session, **kwargs):
    await call.answer()
    await call.message.delete()

    # Получаем позицию пользователя в рейтинге
    record_info = await UserDAO.get_user_rank(session=session, telegram_id=call.from_user.id)
    rank = record_info['rank']
    best_score = record_info['best_score']

    # Формируем сообщение в зависимости от ранга
    if rank == 1:
        text = (
            f"? Поздравляем! Вы на первом месте с рекордом {best_score} очков! Вы — чемпион!\n\n"
            "Держите планку и защищайте свой титул. Нажмите кнопку ниже, чтобы начать игру и "
            "попробовать улучшить свой результат!"
        )
    elif rank == 2:
        text = (
            f"? Великолепно! Вы занимаете второе место с результатом {best_score} очков!\n\n"
            "Еще немного — и вершина ваша! Нажмите кнопку ниже, чтобы попробовать стать первым!"
        )
    elif rank == 3:
        text = (


            f"? Отличный результат! Вы на третьем месте с {best_score} очками!\n\n"
            "Почти вершина! Попробуйте свои силы еще раз, нажав кнопку ниже, и возьмите золото!"
        )
    else:
        text = (
            f"? Ваш рекорд: {best_score} очков. Вы находитесь на {rank}-ом месте в общем рейтинге.\n\n"
            "С каждым разом вы становитесь лучше! Нажмите кнопку ниже, чтобы попробовать "
            "подняться выше и побить свой рекорд!"
        )

    await call.message.answer(text, reply_markup=record_keyboard())

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

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

После, на основании места в рейтинге, формируется сообщение.

При клике на кнопку «Обновить рекорд» просто будет происходить повторный вызов этого метода.

Таким образом, все что касается части Aiogram 3 мы полностью закрыли. Дальнейшая разработка пойдет исключительно на стороне игры и FastApi. Теперь нам остается только подключить нашего телеграмм бота к FastApi приложению.

Подключение бота к FastAPI

Для интеграции бота на aiogram 3 с FastAPI в файле app/main.py нужно внести следующие корректировки. Подключите необходимые модули:

import logging
from contextlib import asynccontextmanager

from app.bot.create_bot import bot, dp, stop_bot, start_bot
from app.bot.handlers.router import router as bot_router
from app.config import settings
from app.game.router import router as game_router
from fastapi.staticfiles import StaticFiles
from aiogram.types import Update
from fastapi import FastAPI, Request

Настройте логирование:

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

Создайте функцию lifespan, используя @asynccontextmanager, для управления жизненным циклом приложения:

@asynccontextmanager
async def lifespan(app: FastAPI):
    logging.info("Starting bot setup...")
    dp.include_router(bot_router)
    await start_bot()
    webhook_url = settings.get_webhook_url()
    await bot.set_webhook(url=webhook_url,
                          allowed_updates=dp.resolve_used_update_types(),
                          drop_pending_updates=True)
    logging.info(f"Webhook set to {webhook_url}")
    yield
    logging.info("Shutting down bot...")
    await bot.delete_webhook()
    await stop_bot()
    logging.info("Webhook deleted")

Затем создайте экземпляр FastAPI и подключите жизненный цикл:

app = FastAPI(lifespan=lifespan)

Подключите папку для статических файлов:

app.mount('/static', StaticFiles(directory='app/static'), 'static')

Определите конечную точку для вебхука, которая будет получать обновления от Telegram и передавать их диспетчеру:

@app.post("/webhook")
async def webhook(request: Request) -> None:
    logging.info("Received webhook request")
    update = Update.model_validate(await request.json(), context={"bot": bot})
    await dp.feed_update(bot, update)
    logging.info("Update processed")

Включите маршрутизацию для игрового функционала:

app.include_router(game_router)

Жизненный цикл работы с FastAPI

Логика работы с FastAPI предполагает использование механизма жизненного цикла (lifespan), который управляется через декоратор @asynccontextmanager в функции lifespan(app: FastAPI). Этот цикл делится на две основные части:

1. Запуск приложения (до yield): выполняется один раз при старте приложения. Здесь мы:

   • Подключаем роутер бота bot_router.

   • Вызываем функции, необходимые сразу после запуска бота.

   • Устанавливаем вебхук для получения обновлений от Telegram.

2. Остановка приложения (после yield): выполняется при завершении работы приложения. Здесь мы:

   • Удаляем вебхук.

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

Преимущества и структура

Эта структура позволяет надежно интегрировать FastAPI с aiogram 3 для работы через вебхуки и создания ботов на основе MiniApp. Я подробно описывал эту связку в статье "Telegram Web App, FastAPI и вебхуки в одном приложении: создаем Telegram-бот с веб-интерфейсом для приема заявок", где можно найти расширенное описание и примеры использования.

Перезапуск приложения

Перезапустите приложение. Если все сделано корректно, бот отправит сообщение о запуске (не забудьте зайти в бота, чтобы он имел возможность отправлять вам сообщения). После запуска выполните команду /start, чтобы попасть в базу данных.

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

Оптимизация шаблонов

Прежде чем перейти к созданию страницы с таблицей лидеров, давайте поработаем с шаблоном index.html и проведём его оптимизацию, а также добавим необходимые для полноценного функционирования приложения API методы.

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

Создадим в папке app/templates файл base.html и заполним его следующим образом:

<!DOCTYPE html>
<html lang="ru">


<head>
    <meta charset="utf-8">
    <title>{% block title %}{% endblock %}</title>
    <link href="/static/style/main.css" rel="stylesheet" type="text/css">
    <link href="/static/style/my_style.css" rel="stylesheet" type="text/css">
    <link rel="shortcut icon" href="/static/favicon.ico">
    <link rel="apple-touch-icon" href="/static/meta/apple-touch-icon.png">
    <link rel="apple-touch-startup-image" href="/static/meta/apple-touch-startup-image-640x1096.png"
          media="(device-width: 320px) and (device-height: 568px) and (-webkit-device-pixel-ratio: 2)">
    <!-- iPhone 5+ -->
    <link rel="apple-touch-startup-image" href="/static/meta/apple-touch-startup-image-640x920.png"
          media="(device-width: 320px) and (device-height: 480px) and (-webkit-device-пиксельное-отношение: 2)">
    <!-- iPhone, retina -->

    <script src="https://telegram.org/js/telegram-web-app.js"></script>
    <script src="/static/js/tg_config.js"></script>
    <meta name="apple-mobile-web-app-capable" content="yes">
    <meta name="apple-mobile-web-app-status-bar-style" content="black">

    <meta name="HandheldFriendly" content="True">
    <meta name="MobileOptimized" content="320">
    <meta name="viewport"
          content="width=device-width, target-densitydpi=160dpi, initial-scale=1.0, maximum-scale=1, user-scalable=no, minimal-ui">
    {% block extra_head %}{% endblock %}
</head>


<body>
<div class="container">
    {% block content %}{% endblock %}
</div>

{% block extra_scripts %}{% endblock %}
</body>
</html>

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

<link href="/static/style/my_style.css" rel="stylesheet" type="text/css">

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

Обратите внимание на следующие строки:

<script src="https://telegram.org/js/telegram-web-app.js"></script>
<script src="/static/js/tg_config.js"></script>

Благодаря этим скриптам страницы игры превращаются в полноценное Telegram MiniApp. Здесь подключается библиотека telegram-web-app.js и наш JavaScript-файл tg_config.js, который будет рассмотрен чуть позже. Остальные стили и скрипты остаются практически без изменений.

Файл tg_config.js

window.addEventListener('load', (event) => {
    const tg = window.Telegram.WebApp;
    tg.ready(); // Подготовка WebApp
    tg.expand(); // Разворачиваем WebApp
    tg.disableVerticalSwipes();

    // Сохраняем userId в localStorage
    const userId = tg.initDataUnsafe.user?.id;
    if (userId) {
        localStorage.setItem("userId", userId);
        console.log("User ID сохранен:", userId);
    } else {
        console.error("User ID не найден в initDataUnsafe.");
    }
});

Здесь используется функция JavaScript, которая выполняет несколько важных действий для нашего WebApp:

1. Получение объекта tg:

   • В переменной tg сохраняется объект window.Telegram.WebApp, представляющий API Telegram WebApp. Это необходимо для дальнейшего взаимодействия с API, например, для изменения размеров окна и работы с пользовательскими данными.

2. Подготовка и настройка WebApp:

   • tg.ready(): Подготавливает WebApp к работе, сообщая Telegram, что приложение загружено и готово к использованию.

   • tg.expand(): Разворачивает WebApp на полную высоту, улучшая пользовательский опыт.

   • tg.disableVerticalSwipes(): Отключает вертикальную прокрутку, чтобы избежать случайных свайпов внутри WebApp. Это удобно, так как наше приложение поддерживает свайпы, в том числе сверху вниз. Без этой функции свайп вниз может свернуть окно приложения MiniApp.

3. Сохранение userId в localStorage:

   • Из tg.initDataUnsafe извлекается user.id (ID пользователя). Если userId доступен, он сохраняется в localStorage для последующего использования:

     localStorage.setItem("userId", userId);

     Таким образом, userId будет доступен, пока WebApp открыт на устройстве.

   • Если userId отсутствует, в консоли выводится ошибка:

     console.error("User ID не найден в initDataUnsafe.");

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

Метод tg.disableVerticalSwipes()

После тестирования на нескольких устройствах выяснилось, что метод disableVerticalSwipes() работает не всегда корректно: на некоторых устройствах он срабатывает, а на других — нет. Поэтому, чтобы обеспечить удобство игры для всех, я добавил дополнительные стрелки управления на экран.

Здесь мы не будем подробно рассматривать JavaScript, но правки были внесены в файл keyboard_input_manager.js. Вы сможете самостоятельно сравнить новую версию этого файла с той, что шла в комплекте с оригинальной игрой в исходном коде моего проекта.

Реализация сохранения лучшего результата в базу данных

Для реализации этого метода, для начала, давайте создадим новый эндпоинт FastApi приложения. Логика такая. Используя PUT-запрос, фронт (странички игры) будет отправлять запрос к бэку, передавая актуальный лучший результат.

Но, перед этим, нам необходимо выполнить небольшую подготовку. Во-первых, давайте опишем необходимые Pydantic-схемы:

class SetBestScoreRequest(BaseModel):
    score: int


class SetBestScoreResponse(BaseModel):
    status: str
    best_score: int

Все просто и, думаю, по названию понятно что к чему.

Теперь нам нужно ещё раз обратиться к файлу app/database.py. На этот раз мы опишем дополнительный метод, который позволит получать сессию уже в контексте FastApi приложения (декоратор connection тут не подойдет).

async def get_session() -> AsyncSession:
    async with async_session_maker() as session:
        try:
            yield session  # Возвращаем сессию для использования
        except Exception:
            await session.rollback()  # Откатываем транзакцию при ошибке
            raise
        finally:
            await session.close()  # Закрываем сессию

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

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

Опишем этот эндпоинт в файле app/game/router.py.

@router.put("/api/bestScore/{user_id}", response_model=SetBestScoreResponse, summary="Set Best Score")
async def set_best_score(
        user_id: int,
        request: SetBestScoreRequest,
        session: AsyncSession = Depends(get_session)
):
    """
    Установить лучший счет пользователя.
    Обновляет значение `best_score` в базе данных для текущего `user_id`.
    """
    score = request.score
    user = await UserDAO.find_one_or_none(session=session, filters=TelegramIDModel(telegram_id=user_id))
    user.best_score = score
    await session.commit()
    return SetBestScoreResponse(status="success", best_score=score)
  

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

Если простыми словами, то то, что вы передаете в функцию Depends, вызывается перед выполнением основного эндпоинта. В нашем случае сначала выполняется функция get_session, которую мы описали ранее.

Когда мы используем Depends(get_session), мы передаем саму функцию get_session, а не вызываем её напрямую. Это связано с тем, что Depends ожидает функцию (объект функции), которую оно сможет вызвать самостоятельно в момент выполнения запроса. Если бы мы передали get_session(), то передали бы результат её выполнения (то есть уже созданную сессию или None, если функция не возвращает значения), а не саму функцию.

Вот как это работает:

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

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

В этом примере Depends(get_session) будет вызывать get_session автоматически перед вызовом эндпоинта и передаст результат вызова (сессию) в параметр session.

Подробно механизм зависимостей в FastApi (Depends) я рассматривал в этой статье "Создание собственного API на Python (FastAPI): Авторизация, Аутентификация и роли пользователей".

Далее мы извлекаем значение best_score из того что передал фронт, а user_id (telegram_id в нашем случае) и пути (из того что указано после / bestScore.

После, внутри энпоинта, я использовал один трюк, который явно демонстрирует преимущества передачи сессии в Depends.

Сначала я воспользовался нашим методом find_one_or_none. Затем, используя мощь ООП Python, напрямую передал новое значение в колонке user для конкретного пользователя. После чего выполнил commit для сохранения результата. Этот подход подробнее рассмотрю в своей следующей статье по SQLAlchemy, которая будет посвящена теме обновления и удаления данных с таблиц.

Теперь мы можем создать обновленный HTML-шаблон страницы с игрой. Для этого в папке templates создадим папку pages и внутри нее файл index.html (можно переместить тот был или создать новый, так как там будет много изменений).

Заполним файл index.html.

{% extends "base.html" %}

{% block title %}Играть в 2048{% endblock %}

{% block content %}
<div class="heading">
    <h1 class="title">2048</h1>
    <div class="scores-container">
        <div class="score-container">0</div>
        <div class="best-container">0</div>
    </div>
</div>

<div class="above-game">
    <p class="game-intro">Соединяй числа и доберись до <strong>плитки 2048!</strong></p>
    <a class="restart-button">Новая игра</a>
</div>

<div class="game-container">
    <div class="game-message">
        <p></p>
        <div class="lower">
            <a class="keep-playing-button">Продолжить</a>
            <a class="retry-button">Попробовать снова</a>
        </div>
    </div>

    <div class="grid-container">
        <div class="grid-row">
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
        </div>
        <div class="grid-row">
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
        </div>
        <div class="grid-row">
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
        </div>
        <div class="grid-row">
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
            <div class="grid-cell"></div>
        </div>
    </div>

    <div class="tile-container">

    </div>
</div>

<div class="game-controls">
    <div class="arrow-row">
        <button class="arrow-btn up-btn">↑</button>
    </div>
    <div class="arrow-row">
        <button class="arrow-btn left-btn">←</button>
        <button class="arrow-btn down-btn">↓</button>
        <button class="arrow-btn right-btn">→</button>
    </div>
    <div class="button-group">
        <button class="icon-btn records-btn" onclick="window.location.href='/records'">
            <!-- Trophy Icon -->
            <svg width="24" height="24" viewBox="0 0 24 24">
                <path d="M17 3H7c0-1.1-.9-2-2-2H3c-1.1 0-2 .9-2 2v2c0 2.76 2.24 5 5 5h.18C6.07 11.19 6 11.58 6 12c0 3.31 2.69 6 6 6s6-2.69 6-6c0-.42-.07-.81-.18-1.18H19c2.76 0 5-2.24 5-5V3c0-1.1-.9-2-2-2h-2c-1.1 0-2 .9-2 2zM5 7c-1.65 0-3-1.35-3-3V3h2v4h1zm14-4h2v1c0 1.65-1.35 3-3 3V3zM12 18c-2.67 0-5.33 1.34-6 4h12c-.67-2.66-3.33-4-6-4z"
                      fill="currentColor"></path>
            </svg>
        </button>
        <button class="icon-btn reset-btn" id="clearStorageButton">
            <!-- Reset Icon -->
            <svg width="24" height="24" viewBox="0 0 24 24">
                <path d="M12 4V1L8 5l4 4V6c3.31 0 6 2.69 6 6 0 .34-.03.67-.08 1H18c0-3.31-2.69-6-6-6zM6 13c0 3.31 2.69 6 6 6v3l4-4-4-4v3c-2.76 0-5-2.24-5-5H6c0 .34.03.67.08 1H6c-.05-.33-.08-.66-.08-1z"
                      fill="currentColor"></path>
            </svg>
        </button>
        <button class="icon-btn exit-btn" onclick="window.Telegram.WebApp.close()">
            <!-- Exit Icon -->
            <svg width="24" height="24" viewBox="0 0 24 24">
                <path d="M16 13v-2H7V9l-5 4 5 4v-3h9zM19 3H5c-1.1 0-2 .9-2 2v4h2V5h14v14H5v-4H3v4c0 1.1.9 2 2 2h14c1.1 0 2-.9 2-2V5c0-1.1-.9-2-2-2z"
                      fill="currentColor"></path>
            </svg>
        </button>
    </div>
</div>
{% endblock %}

{% block extra_scripts %}
<script src="/static/js/bind_polyfill.js?v=1.0.1"></script>
<script src="/static/js/classlist_polyfill.js?v=1.0.1"></script>
<script src="/static/js/animframe_polyfill.js?v=1.0.1"></script>
<script src="/static/js/keyboard_input_manager.js?v=1.0.1"></script>
<script src="/static/js/html_actuator.js?v=1.0.1"></script>
<script src="/static/js/grid.js?v=1.0.1"></script>
<script src="/static/js/tile.js?v=1.0.1"></script>
<script src="/static/js/local_storage_manager.js?v=1.1.4"></script>
<script src="/static/js/game_manager.js?v=1.0.2"></script>
<script src="/static/js/application.js?v=1.0.3"></script>
<script src="/static/js/scan.js?v=1.0.1"></script>
{% endblock %}

Используя строку {% extends "base.html" %}, мы унаследовали наш новый шаблон от базового файла. Это позволило избежать необходимости импортировать стили и универсальные JavaScript-скрипты на каждой странице вручную.

Теперь из нового в шаблоне по сравнению с оригинальной игрой:

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

• Также добавлены кнопки для закрытия WebApp, обнуления списка рекордов и перехода на страницу с таблицей лидеров.

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

Первым делом, я внес корректировки в метод LocalStorageManager.prototype.setBestScore = async function (score). После стандартного поведения игры, я добавил отправку запроса к нашему API. Вот что получилось в итоге.

LocalStorageManager.prototype.setBestScore = async function (score) {
    this.storage.setItem(this.bestScoreKey, score);
    const userId = localStorage.getItem("userId");
    try {
        // Отправка нового результата на сервер
        const response = await fetch(`/api/bestScore/${userId}`, {
            method: "PUT",
            headers: {
                "Content-Type": "application/json"
            },
            body: JSON.stringify({score: score})
        });

        if (response.ok) {
            const data = await response.json();
            console.log("Best score updated on server:", data.best_score);
        } else {
            console.error("Failed to update best score on server.");
        }
    } catch (error) {
        console.error("Error updating best score on server:", error);
    }
};

Тут user_id (telegram_id) мы получаем из локального хранилища, так как это значение там точно есть, ведь об этом мы позаботились ещё на этапе инициализации приложения MiniApp. Другими словами, как только пользователь запустит наш MiniApp, его Telegram ID сразу попадет в локальное хранилище и будет доступен во всем приложении.

Далее, мы просто поместили «перехваченное» значение в запрос к нашему API и выполнили обновление.

Метода для очистки лучшего результата в исходнике игры не было и его пришлось писать отдельно.

Такой метод у меня получился:

LocalStorageManager.prototype.clearStorage = async function () {
    const userId = localStorage.getItem("userId");
    this.storage.clear();

    try {
        // Отправка нового результата на сервер
        const response = await fetch(`/api/bestScore/${userId}`, {
            method: "PUT",
            headers: {
                "Content-Type": "application/json"
            },
            body: JSON.stringify({score: 0}) // Убедитесь, что это объект { score: <значение> }
        });

        if (response.ok) {
            const data = await response.json();
            console.log("Best score updated on server:", data.best_score);
        } else {
            console.error("Failed to update best score on server.");
        }
    } catch (error) {
        console.error("Error updating best score on server:", error);
    }
};

Объяснение работы функций:

1. Очистка локального хранилища:
   Метод this.storage.clear(); очищает локальное хранилище, удаляя все сохраненные данные. В зависимости от возможностей устройства, используется localStorage, а если оно не поддерживается, — fakeStorage.

2. Сброс результата на сервере:

   • ID пользователя извлекается из localStorage (значение userId).

   • Отправляется PUT-запрос на сервер по адресу /api/bestScore/${userId}, передавая в теле запроса объект { score: 0 } для сброса лучшего результата.

   • При успешном обновлении сервер подтверждает операцию; если возникает ошибка, выводится соответствующее уведомление.

3. Обработчик для кнопки очистки:

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

Кроме того, я добавил обработчик события при клике на кнопку «Очистить рекорд»:

document.getElementById("clearStorageButton").addEventListener("click", function () {
    Telegram.WebApp.showConfirm("Вы уверены, что хотите очистить свой рекорд?", async function (confirmation) {
        if (confirmation) {
            const manager = new LocalStorageManager();
            await manager.clearStorage(); // Ждём завершения очистки

            Telegram.WebApp.showAlert("Ваш рекорд успешно очищен."); // Уведомление об успешной очистке
            location.reload(); // Перезагружаем страницу
        } else {
            Telegram.WebApp.showAlert("Очистка рекорда отменена."); // Уведомление об отмене
        }
    });
});

Этот код:

• Показывает подтверждение при нажатии на кнопку.

• В зависимости от ответа, вызывает clearStorage() и перезагружает страницу либо отменяет действие.

Не забываем внести изменения в эндпоинте для вывода игры:

@router.get("/", response_class=HTMLResponse)
async def read_root(request: Request):
    return templates.TemplateResponse("pages/index.html", {"request": request})

Тут мы изменили путь к странице. Теперь мы указали page/.

Теперь подготовим шаблон (страницу) для вывода топа игроков.

Для этого в папаке templates/pages создадим файл. Я назову его records.html

{% extends "base.html" %}

{% block title %}Рекорды 2048{% endblock %}


{% block content %}
<div class="heading">
    <h1 class="title">Топ-{{ records|length }} игроков</h1>
</div>

<div class="score-table">
    <table>
        <thead>
        <tr>
            <th>Место</th>
            <th>Telegram ID</th>
            <th>Имя</th>
            <th>Очки</th>
        </tr>
        </thead>
        <tbody>
        {% for record in records %}
        <tr>
            <td class="{% if record.rank == 1 %}first-place{% elif record.rank == 2 %}second-place{% elif record.rank == 3 %}third-place{% endif %}">
            <span class="rank-icon">
                {% if record.rank == 1 %}?{% elif record.rank == 2 %}?{% elif record.rank == 3 %}?{% else %}{{ record.rank }}{% endif %}
            </span>
            </td>
            <td>{{ record.telegram_id }}</td>
            <td>{{ record.first_name }}</td>
            <td>{{ record.best_score }}</td>
        </tr>
        {% endfor %}
        </tbody>
    </table>
</div>

<div class="button-group">
    <button class="icon-btn play-btn" onclick="window.location.href='/'">
        <!-- Play Icon -->
        <svg width="24" height="24" viewBox="0 0 24 24">
            <path d="M8 5v14l11-7z" fill="currentColor"></path>
        </svg>
    </button>
    <button class="icon-btn reset-btn" id="clearStorageButton">
        <!-- Reset Icon -->
        <svg width="24" height="24" viewBox="0 0 24 24">
            <path d="M12 4V1L8 5l4 4V6c3.31 0 6 2.69 6 6 0 .34-.03.67-.08 1H18c0-3.31-2.69-6-6-6zM6 13c0 3.31 2.69 6 6 6v3l4-4-4-4v3c-2.76 0-5-2.24-5-5H6c0 .34.03.67.08 1H6c-.05-.33-.08-.66-.08-1z"
                  fill="currentColor"></path>
        </svg>
    </button>
    <button class="icon-btn exit-btn" onclick="window.Telegram.WebApp.close()">
        <!-- Exit Icon -->
        <svg width="24" height="24" viewBox="0 0 24 24">
            <path d="M16 13v-2H7V9l-5 4 5 4v-3h9zM19 3H5c-1.1 0-2 .9-2 2v4h2V5h14v14H5v-4H3v4c0 1.1.9 2 2 2h14c1.1 0 2-.9 2-2V5c0-1.1-.9-2-2-2z"
                  fill="currentColor"></path>
        </svg>
    </button>
</div>
{% endblock %}

{% block extra_scripts %}
<script src="/static/js/local_storage_manager.js?v=1.1.4"></script>
{% endblock %}

Тут мы переложили всю логику на сторону FastApi, а, точнее Jinja2.

Единственное что мы импортировали файл:  <script src="/static/js/local_storage_manager.js?v=1.1.4"></script> для того чтоб сработала логика очистки лучшего результата для конкретного пользователя со всей остальной заложенной логикой.

Теперь опишем эндпоинт для отображения этой страницы (файл game/router.py).

@router.get("/records", response_class=HTMLResponse)
async def read_records(request: Request, session: AsyncSession = Depends(get_session)):
    # Получаем топовые рекорды с их позициями
    records = await UserDAO.get_top_scores(session=session)
    # Передаем актуальный список рекордов в шаблон
    return templates.TemplateResponse("pages/records.html", {"request": request, "records": records})

Мы снова воспользовались зависимостью Depends, чтобы получить сессию, и затем, используя метод get_top_scores, получили топ-20 игроков. Полученную информацию передали на страницу с помощью ключа records. Ранее, на примере с HTML-шаблоном, было показано, как эта переменная используется на практике.

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

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

Деплой игры 2048 на Amvera Cloud

Чтобы сервис Amvera знал, что запускать, необходимо подготовить файл настроек. Назовем его amvera.yml и разместим в корне проекта на одном уровне с файлами .env и requirements.txt. Заполним файл следующим образом:

meta:
  environment: python
  toolchain:
    name: pip
    version: 3.12
build:
  requirementsPath: requirements.txt
run:
  persistenceMount: /data
  containerPort: 8000
  command: uvicorn app.main:app --host 0.0.0.0 --port 8000

Этот файл описывает контейнерное окружение для Python-приложения на FastAPI, устанавливает зависимости и настраивает параметры запуска сервера uvicorn. Этих данных достаточно, чтобы Amvera понимала, как разворачивать и запускать наше FastAPI-приложение.

Теперь процесс деплоя будет состоять из следующих шагов:

1. Доставить файлы приложения с файлом настроек на сервис Amvera. Это можно сделать с помощью команд GIT или интерфейса на сайте (я выберу интерфейс).

2. Заменить ссылку, предоставленную NGROK, на бесплатное доменное имя, которое выдаст Amvera.

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

Пошаговый гайд по деплою проекта на Amvera Cloud

1. Зарегистрируйтесь на сайте Amvera Cloud и получите бонус в размере 111 рублей на основной баланс, если ранее у вас не было аккаунта.

2. Перейдите в раздел проектов и нажмите «Создать проект».

3. Введите название проекта и выберите тарифный план (для текущего проекта подойдет тариф «Начальный»).

4. Нажмите «Далее».

5. На следующем экране выберите опцию «Через интерфейс» и загрузите файлы проекта. Если вы предпочитаете работать с GIT-командами, выберите соответствующий вариант. На втором экране система Amvera предоставит подробные инструкции со всеми необходимыми командами для работы через GIT.

6. Нажмите «Далее».

7. Проверьте настройки и, если все верно, нажмите «Завершить» для окончания создания проекта.

Получение бесплатного домена и привязка к проекту

1. Перейдите в созданный проект и откройте вкладку «Настройки».

2. Нажмите «Добавить доменное имя» и получите бесплатный домен.

3. Скопируйте это доменное имя и откройте локальный файл .env. В этом файле замените доменное имя, предоставленное NGROK, на новое доменное имя от Amvera Cloud.

4. Вернитесь на вкладку «Репозиторий» в Amvera и загрузите измененный файл .env, чтобы перезаписать его с новым доменным именем.

5. В BotFather обновите ссылку для MiniApp и MenuButton, подставив новое доменное имя от Amvera.

6. Чтобы изменения вступили в силу, нажмите «Пересобрать проект». Проект будет пересобран с новым .env файлом и доменным именем.

7. Через пару минут, если все выполнено корректно, бот запустится и будет готов к работе.

Чтобы протестировать готовый проект бота-игры, переходите по ссылке: Игра 2048. Стать лучшим!

Заключение

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

Используя такие технологии, как FastAPI, Aiogram и SQLAlchemy, вы можете создавать мощные приложения, а вопросы интеграции больше не будут вызывать сложностей. В нашем примере мы использовали готовую игру, но, обладая достаточными знаниями JavaScript, вы сможете заменить её любым другим приложением: будь то игра любой сложности или функциональный инструмент.

Важно понимать каждый пройденный шаг: от настройки связки FastAPI и Aiogram до интеграции API-методов с фронтендом. Это заложит основу для разработки более сложных проектов в будущем.

Не забывайте, что исходный код проекта доступен бесплатно в моем Telegram-канале «Легкий путь в Python». Присоединяйтесь к сообществу, где вас ждут более тысячи единомышленников и уникальный контент. Если вам понравилось, оставляйте лайки и комментарии — ваша активность помогает развивать канал и создавать ещё больше полезных материалов!

До скорой встречи!