src/db.py - main - Hetvild/Client-Birthday-Alert

src/db.py
80 строк3 KB
bil
init
28 янв 2026, 09:52
28 янв 2026, 09:52d793dde
from datetime import datetime, date
from typing import Optional

from sqlalchemy import create_engine, Column, Integer, Date
from sqlalchemy.orm import declarative_base, sessionmaker

# Создаем подключение к базе данных sqlite
engine = create_engine("sqlite:///data/data.sqlite", echo=True)


# Создаем базовый класс для модели
Base = declarative_base()

# Создаем фабрику сессий
SessionMaker = sessionmaker(autocommit=False, autoflush=False, bind=engine)


class DateCompletion(Base):
    __tablename__ = "date_completion"
    id = Column(Integer, primary_key=True)
    date_of_completion = Column(Date, nullable=False)


# Декларативно создаем таблицы
# Base.metadata.create_all(engine)


# Создаем запись в таблице
def write_data_to_db(date: str):
    """
    Функция создает запись в базу данных
    :param date:
    :return:
    """
    with SessionMaker() as session:
        dt = datetime.fromisoformat(date)
        date_completion = DateCompletion(date_of_completion=dt)
        session.add(date_completion)
        session.commit()


def get_date_completion_by_id() -> Optional[date | None]:
    """
    Получает дату завершения из таблицы `DateCompletion` по умолчанию (id=1).

    :return: Дата завершения или None, если запись не найдена.
    """
    try:
        with SessionMaker() as session:
            record = (
                session.query(DateCompletion).filter(DateCompletion.id == 1).first()
            )

            if record:
                return record.date_of_completion
            else:
                return None

    except Exception as e:
        # Логирование ошибки рекомендуется добавить в реальном проекте
        print(f"Ошибка при получении даты завершения: {e}")
        return None


def update_date_completion_by_id(date: str) -> bool | None:
    """
    Функция обновляет дату в базе данных по ID.
    :param date:
    :return:
    """
    with SessionMaker() as session:
        date_completion = (
            session.query(DateCompletion).filter(DateCompletion.id == 1).first()
        )
        if date_completion:
            date_completion.date_of_completion = datetime.fromisoformat(date)
            session.commit()
            return True
        else:
            return False