🚀 Разбор Telegram-бота: Часть 1 — База данных и модели (SQLAlchemy)
Привет, кодеры! В прошлом посте я рассказал о своём новом проекте — Telegram-боте для планирования задач и расписания. Он помогает фрилансерам и студентам держать дела в порядке: разовые задачи с напоминаниями, еженедельные занятия, отчёты и настройки. Бот на Python, с использованием telebot для интерфейса, APScheduler для тайминга и SQLAlchemy для хранения данных.
Сегодня начинаем разбирать код по частям, как обещал. Стартуем с фундамента: базы данных. Почему SQLAlchemy? Она гибкая, поддерживает разные БД (SQLite для теста, PostgreSQL для продакшена), и ORM делает работу с данными как с объектами Python — удобно и без сырого SQL.🗂
Структура: db.py и models.pyСначала config.py (там настройки, включая URL для БД). Вот базовый snippet:# config.py
import os
from dotenv import load_dotenv
import pytz
load_dotenv()
BOT_TOKEN = os.getenv("BOT_TOKEN") # обязательно
DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///planner.db")
TIMEZONE = os.getenv("TIMEZONE", "Europe/Moscow")
TZ = pytz.timezone(TIMEZONE)
DATABASE_URL — это строка подключения. Для SQLite просто файл, для Postgres — 'postgresql://user:pass@host/db'.
.env файл хранит токен и секреты (не коммить в Git!).
Теперь db.py: создаём движок и сессию SQLAlchemy.
# db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, scoped_session
from config import DATABASE_URL
engine = create_engine(DATABASE_URL, echo=False, future=True)
SessionLocal = scoped_session(sessionmaker(bind=engine, autoflush=False, autocommit=False))
SessionLocal — это фабрика сессий. Мы используем scoped_session для thread-safety (важно для бота с многопоточностью).
Ключевой файл — models.py. Здесь определяем таблицы: пользователи, задачи и расписание (классы/занятия). Используем Enum для статусов задач.# models.py (фрагмент)
from sqlalchemy import Column, Integer, String, DateTime, Boolean, ForeignKey, Text, Enum, Time
from sqlalchemy.orm import declarative_base, relationship
import enum
from datetime import datetime, time
Base = declarative_base()
class TaskStatus(enum.Enum):
pending = "pending"
done = "done"
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True) # internal id
tg_id = Column(Integer, unique=True, index=True, nullable=False) # telegram user id
name = Column(String, nullable=True)
default_remind_minutes = Column(Integer, default=30) # default reminder before event in minutes
tasks = relationship("Task", back_populates="user", cascade="all, delete-orphan")
classes = relationship("ClassSchedule", back_populates="user", cascade="all, delete-orphan")
class Task(Base):
__tablename__ = "tasks"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
title = Column(String, nullable=False)
description = Column(Text, nullable=True)
starts_at = Column(DateTime, nullable=False, index=True) # <- timezone=True
remind_before_minutes = Column(Integer, nullable=True)
status = Column(Enum(TaskStatus), default=TaskStatus.pending)
user = relationship("User", back_populates="tasks")
# ClassSchedule для еженедельных занятий (фрагмент)
class ClassSchedule(Base):
__tablename__ = "class_schedules"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False)
title = Column(String, nullable=False)
weekday = Column(Integer, nullable=False) # 0..6
start_time = Column(Time, nullable=False, index=True)
duration_minutes = Column(Integer, default=90)
location = Column(String, nullable=True)
active = Column(Boolean, default=True)
remind_before_minutes = Column(Integer, nullable=True)
user = relationship("User", back_populates="classes")
