kb-meeting-bot/app/data_provider.py

from sqlite3.dbapi2 import IntegrityError
from dateutil import parser
from aiogram.types.user import User
from app.dates import get_now_datetime_str, is_actual_date
from app.messages import Event, Notifications
from app.db import delete, fetchall, insert, select_by_keys

def add_notifications2db(notification_object: Notifications, meeting_id: int):
    for n in [notification_object.befor_day, notification_object.befor_4hours, notification_object.befor_5mins]:
        insert("notifications", {
            "when_notify": n.date_time_srting,
            "meeting_id": meeting_id,
            "chat_id": n.chat_id,
            "message": n.message
            })


def add_event2db(event: Event) -> int:
    return insert("meetings", {
        "description": event.agenda,
        "meeting_datetime": event.date_time,
        "participants": event.participants,
        "leader_id": event.organizer.id,
        "chat_id": event.chat_id,
        "meeting_link": event.meeting_url
        })


def check_notifications_now() -> list:
    """
    Достает из БД все напоминания на текущую дату-время. Возвращает список со 
    всеми напоминавниями
    """
    date_time_str = get_now_datetime_str()
    notifications = select_by_keys("notifications", 
            ["id", "chat_id", "message"],
            {"when_notify": date_time_str})
    if type(notifications) == dict: notifications = [notifications]
    for n in notifications:
        delete("notifications", n["id"])
    return notifications

def get_user_by_id(user_id: int):
    return select_by_keys(
        'users',
        ["id", "firstname", "lastname", "login"],
        {'id': str(user_id)})


def get_all_users():
    return fetchall("users", ["id", "firstname", "lastname", "login"])


def del_user_by_id(user_id: int):
    delete('users', user_id)


def add_user2db(user: User):
    try:
        return insert("users", {
                "id":           user.id,
                "firstname":    user.first_name,
                "lastname":     user.last_name,
                "login":        user.username
            })
    except IntegrityError:
        pass

def delete_meeting_and_notifications(meeting_id: int, user_id: int):
    """
    Принимает айди встречи и пользователя, от которого поступил запрос на 
    удаление, получает из БД айди пользователя создавшего встречу. Удаляет встречу
    если пользователь один и тот же. Также удаляет все напоминания
    """
    if user_id != 555:
        meeting = select_by_keys("meetings",
                ["leader_id"],
                {"id": str(meeting_id)})
        if not meeting:
            raise Exception(
                    "Не могу найти эту встречу\. Возможно, ее уже отменили")
        if meeting["leader_id"] != user_id:
            raise Exception(
                    "Вы не создавали эту встречу и не можете её отменить")
    notifications = select_by_keys("notifications",
            ["id"],
            {"meeting_id": meeting_id})
    if type(notifications) == dict:
        notifications = [notifications]
    for n in notifications:
        delete("notifications", n["id"])
    delete("meetings", meeting_id)



def filter_actual_events(event: dict) -> bool:
    """Фильтрует события, оставляет те, которые еще не произошли"""
    return is_actual_date(event["meeting_datetime"])

def get_all_users_meetings(user_id: int) -> list:
    """
    Принимает айди пользователя и возвращает все не прошедшие встречи созданные им
    """
    events = select_by_keys("meetings",
            ["id", "description", "meeting_datetime", "chat_id"],
            {"leader_id": user_id}
            )
    if type(events) != list: events = [events]
    for e in events:
        e["meeting_datetime"] = parser.parse(e["meeting_datetime"])
    events: list = sorted(events, key=lambda e: e["meeting_datetime"].timestamp())
    return list(filter(filter_actual_events, events))