Создание бота с использованием баз данных и API FastAPI

Добро пожаловать на новую веху нашего курса! Сегодня мы будем создавать Telegram-бота, который умеет работать с базой данных и API FastAPI.

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

Боты могут стать настоящими помощниками в бизнесе и повседневной жизни.

Представьте себе официанта (бота) на кухне. Если у него нет блокнота (базы данных), он забудет, что вы заказали, и принесёт всё подряд.

Если же у него есть блокнот, он запишет ваш заказ и отнесёт его на кухню.

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

Примеры реального применения:

• Хранение пользователей, чтобы бот знал, кому он "говорит".
• Сохранение состояний чатов, чтобы поддерживать продолжительные диалоги.
• Учет данных, введенных пользователями (например, список дел или заметки).

---

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

Прежде чем писать код, убедитесь, что у вас установлен FastAPI, PostgreSQL (или SQLite), а также ORM (SQLAlchemy в нашем случае).

pip install fastapi uvicorn sqlalchemy psycopg2 python-telegram-bot

Если вы хотите использовать SQLite, вместо psycopg2 вам ничего дополнительно ставить не нужно — SQLite уже встроен в Python.

---

Шаг 1: Настройка базы данных

Создадим файл db.py, в котором будем управлять подключением к базе данных. Используем SQLAlchemy для работы с таблицами.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

DATABASE_URL = "sqlite:///./test.db"  # Для PostgreSQL: "postgresql://user:password@localhost/dbname"

engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

Здесь мы создаём подключение к базе данных. В случае SQLite файл будет храниться в папке проекта.

Определение таблицы пользователей

Теперь создадим таблицу для хранения данных о пользователях. Например, ID чата и имя.

from sqlalchemy import Column, Integer, String
from db import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    telegram_id = Column(Integer, unique=True, index=True)
    name = Column(String, index=True)

Инициализация базы данных

Создадим файл main.py, в котором будем управлять созданием таблиц:

from db import Base, engine

# Запуск приложения FastAPI
if __name__ == "__main__":
    Base.metadata.create_all(bind=engine)

Запустите этот файл, чтобы база данных и таблица были созданы. SQLite создаст файл test.db в каталоге проекта.

---

Шаг 2: Разработка API FastAPI

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

Эндпоинт добавления пользователя

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from db import SessionLocal
from models import User

app = FastAPI()

# Зависимость для передачи сессии базы данных
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/users/")
async def create_user(telegram_id: int, name: str, db: Session = Depends(get_db)):
    db_user = User(telegram_id=telegram_id, name=name)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

Этот эндпоинт сохраняет пользователя в базу данных (ищите его по адресу /users/).

Получение пользователей

@app.get("/users/")
async def get_users(db: Session = Depends(get_db)):
    return db.query(User).all()

Теперь у нас есть API для добавления и просмотра пользователей. Следующим шагом будет подключение этого API к Telegram-боту.

---

Шаг 3: Интеграция Telegram-бота с базой данных

Установим бот и API. В теле бота мы будем использовать вышеописанные эндпоинты.

from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler
import httpx

API_URL = "http://127.0.0.1:8000"  # URL FastAPI

async def start(update: Update, context):
    user = update.effective_user
    chat_id = update.effective_chat.id

    # Сохраняем пользователя в FastAPI
    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{API_URL}/users/",
            params={"telegram_id": chat_id, "name": user.first_name}
        )
        if response.status_code == 200:
            await update.message.reply_text(f"Привет, {user.first_name}! Ты зарегистрирован.")
        else:
            await update.message.reply_text("Ошибка регистрации. Попробуйте позже.")

application = ApplicationBuilder().token("ВАШ_ТОКЕН").build()
application.add_handler(CommandHandler("start", start))

application.run_polling()

Теперь при вводе команды /start, пользователь будет добавлен в базу данных.

Здесь мы используем HTTP-клиент httpx для асинхронной работы с API.

---

Шаг 4: Расширение функциональности

Добавим команду /users, которая будет возвращать список всех зарегистрированных пользователей:

async def get_users(update: Update, context):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{API_URL}/users/")
        users = response.json()
        user_list = "\n".join([f"{user['name']} (ID: {user['telegram_id']})" for user in users])
        await update.message.reply_text(f"Зарегистрированные пользователи:\n{user_list}")

application.add_handler(CommandHandler("users", get_users))

Теперь вы можете ввести /users, чтобы получить список всех пользователей.

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

---

Итог

В результате мы создали Telegram-бота, который:

1. Регистрирует пользователей в базе данных через FastAPI.
2. Может получать список всех зарегистрированных пользователей.
3. Работает асинхронно как с FastAPI, так и внутри Telegram.

Эта архитектура гибкая и масштабируемая. Вы можете легко добавлять новые эндпоинты в FastAPI и команды в бота.

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