Edit on GitHub

src.db.utils.notifications

Функционал для работы с БД. Работа с пользовательскими уведомлениями

 1"""Функционал для работы с БД. Работа с пользовательскими уведомлениями"""
 2
 3import logging
 4
 5from sqlalchemy import delete, desc, insert, select
 6
 7from core.metric import async_speed_metric
 8from db.database import execute_query
 9from db.models import Notifications
10
11logger = logging.getLogger()
12
13
14@async_speed_metric
15async def get_notifications(user_id):
16    # result: Notifications = await CashManager(Notifications).get({user_id: ...})
17    # if result:
18    #     return result[0]
19
20    query = (
21        select(Notifications)
22        .where(Notifications.user_id == user_id)
23        .order_by(desc(Notifications.date))
24    )
25    result: list[Notifications] = (await execute_query(query)).scalars().all()
26
27    # if result:
28    #     await CashManager(Notifications).add({user_id: result.__ustr_dict__})
29
30    return result
31
32
33@async_speed_metric
34async def add_notification(notification: Notifications.ValidationSchema):
35    # await CashManager(Notifications).delete(user_id)
36
37    query = (
38        insert(Notifications)
39        .values(notification.model_dump(exclude="id"))
40        .returning(Notifications)
41    )
42
43    return (await execute_query(query)).scalar_one_or_none()
44
45
46@async_speed_metric
47async def remove_notification(notification_id: int):
48    # await CashManager(Notifications).delete(user_id)
49
50    query = delete(Notifications).where(Notifications.id == notification_id)
51
52    await execute_query(query)
logger = <RootLogger root (DEBUG)>
@async_speed_metric
async def get_notifications(user_id): 
15@async_speed_metric
16async def get_notifications(user_id):
17    # result: Notifications = await CashManager(Notifications).get({user_id: ...})
18    # if result:
19    #     return result[0]
20
21    query = (
22        select(Notifications)
23        .where(Notifications.user_id == user_id)
24        .order_by(desc(Notifications.date))
25    )
26    result: list[Notifications] = (await execute_query(query)).scalars().all()
27
28    # if result:
29    #     await CashManager(Notifications).add({user_id: result.__ustr_dict__})
30
31    return result
@async_speed_metric
async def add_notification(notification: db.models.notifications.Notifications.ValidationSchema): 
34@async_speed_metric
35async def add_notification(notification: Notifications.ValidationSchema):
36    # await CashManager(Notifications).delete(user_id)
37
38    query = (
39        insert(Notifications)
40        .values(notification.model_dump(exclude="id"))
41        .returning(Notifications)
42    )
43
44    return (await execute_query(query)).scalar_one_or_none()
@async_speed_metric
async def remove_notification(notification_id: int): 
47@async_speed_metric
48async def remove_notification(notification_id: int):
49    # await CashManager(Notifications).delete(user_id)
50
51    query = delete(Notifications).where(Notifications.id == notification_id)
52
53    await execute_query(query)