src.db.utils.admin

 1"""Функкционал для работы с БД. Администратор"""
 2
 3import logging
 4
 5from sqlalchemy import and_, not_, select, update
 6
 7from core.metric import async_speed_metric
 8from db.database import execute_query
 9from db.models import UserActivity, UserData
10from db.utils.redis import CashManager
11
12logger = logging.getLogger()
13
14
15@async_speed_metric
16async def set_admin(user_id):
17    """Устанавливает пользователя администратором.
18
19    Удаляет кэшированные данные пользователя и обновляет его статус в базе данных.
20
21    Args:
22        user_id (int): Идентификатор пользователя, которого необходимо установить администратором.
23
24    Returns:
25        UserData | None: Объект UserData, если обновление прошло успешно, иначе None.
26    """
27    await CashManager(UserData).delete(user_id)
28
29    query = (
30        update(UserData)
31        .values(admin=True)
32        .filter_by(telegram_id=user_id)
33        .returning(UserData)
34    )
35    return (await execute_query(query)).scalar_one_or_none()
36
37
38@async_speed_metric
39async def get_valid_users(my_id):
40    """Получает список всех активных пользователей, кроме указанного.
41
42    Фильтрует пользователей по различным критериям, включая статус активности и настройки.
43
44    Args:
45        my_id (int): Идентификатор пользователя, который будет исключен из списка.
46
47    Returns:
48        list[UserData]: Список объектов UserData, представляющих активных пользователей.
49    """
50    query = select(UserData).where(
51        and_(
52            not_(UserData.mute),
53            UserData.telegram_id != my_id,
54            UserData.active != UserActivity.freezed,
55            UserData.active != UserActivity.banned,
56            UserData.active != UserActivity.deleted,
57            UserData.stage != 0,
58        )
59    )
60    results: list[UserData] = (await execute_query(query)).scalars().all()
61    return results
62
63
64@async_speed_metric
65async def get_all_users():
66    """Получает список всех пользователей.
67
68    Запрашивает все записи пользователей из базы данных.
69
70    Returns:
71        list[UserData]: Список всех объектов UserData.
72    """
73    query = select(UserData)
74    results: list[UserData] = (await execute_query(query)).scalars().all()
75    return results
76
77
78@async_speed_metric
79async def get_admins():
80    """Получает список всех администраторов.
81
82    Фильтрует пользователей по статусу администратора и активности.
83
84    Returns:
85        list[UserData]: Список объектов UserData, представляющих администраторов.
86    """
87    query = select(UserData).where(
88        and_(
89            UserData.admin,
90            UserData.active != UserActivity.banned,
91            UserData.active != UserActivity.deleted,
92        )
93    )
94
95    return (await execute_query(query)).scalars().all()