Створення бота з використанням баз даних та 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 і команди в бота.

Наприклад, наступний крок — зробити щось складніше, типу системи нотаток або задач.