110 lines
3.6 KiB
Python
110 lines
3.6 KiB
Python
from enum import Enum
|
||
from typing import Optional
|
||
from api.schemas.endpoints.account import UserUpdate
|
||
from api.db.tables.account import KeyType, KeyStatus
|
||
from api.schemas.endpoints.account_keyring import AccountKeyringUpdate
|
||
from api.db.tables.account import AccountRole, AccountStatus
|
||
from api.schemas.endpoints.list_events import ListEventUpdate
|
||
from api.db.tables.events import EventState, EventStatus
|
||
|
||
|
||
def update_user_data_changes(update_data: UserUpdate, user) -> Optional[dict]:
|
||
"""
|
||
Сравнивает данные для обновления с текущими значениями пользователя.
|
||
Возвращает:
|
||
- None, если нет изменений
|
||
- Словарь {поле: новое_значение} для измененных полей
|
||
"""
|
||
update_values = {}
|
||
changes = {}
|
||
|
||
for field, value in update_data.model_dump(exclude_unset=True).items():
|
||
if value is None:
|
||
continue
|
||
|
||
if isinstance(value, (AccountRole, AccountStatus)):
|
||
update_values[field] = value.value
|
||
else:
|
||
update_values[field] = value
|
||
|
||
for field, new_value in update_values.items():
|
||
if not hasattr(user, field):
|
||
continue
|
||
|
||
current_value = getattr(user, field)
|
||
|
||
if isinstance(current_value, Enum):
|
||
current_value = current_value.value
|
||
|
||
if current_value != new_value:
|
||
changes[field] = new_value
|
||
|
||
return changes if changes else None
|
||
|
||
|
||
def update_key_data_changes(update_data: AccountKeyringUpdate, key) -> Optional[dict]:
|
||
"""
|
||
Сравнивает данные для обновления с текущими значениями пользователя.
|
||
Возвращает:
|
||
- None, если нет изменений
|
||
- Словарь {поле: новое_значение} для измененных полей
|
||
"""
|
||
update_values = {}
|
||
changes = {}
|
||
|
||
for field, value in update_data.model_dump(exclude_unset=True).items():
|
||
if value is None:
|
||
continue
|
||
|
||
if isinstance(value, (KeyType, KeyStatus)):
|
||
update_values[field] = value.value
|
||
else:
|
||
update_values[field] = value
|
||
|
||
for field, new_value in update_values.items():
|
||
if not hasattr(key, field):
|
||
continue
|
||
|
||
current_value = getattr(key, field)
|
||
|
||
if isinstance(current_value, Enum):
|
||
current_value = current_value.value
|
||
|
||
if current_value != new_value:
|
||
changes[field] = new_value
|
||
|
||
return changes if changes else None
|
||
|
||
def update_listevents_data_changes(update_data: ListEventUpdate, listevents) -> Optional[dict]:
|
||
"""
|
||
Сравнивает данные для обновления с текущими значениями listevents.
|
||
Возвращает:
|
||
- None, если нет изменений
|
||
- Словарь {поле: новое_значение} для измененных полей
|
||
"""
|
||
update_values = {}
|
||
changes = {}
|
||
|
||
for field, value in update_data.model_dump(exclude_unset=True).items():
|
||
if value is None:
|
||
continue
|
||
|
||
if isinstance(value, (EventState, EventStatus)):
|
||
update_values[field] = value.value
|
||
else:
|
||
update_values[field] = value
|
||
|
||
for field, new_value in update_values.items():
|
||
if not hasattr(listevents, field):
|
||
continue
|
||
|
||
current_value = getattr(listevents, field)
|
||
|
||
if isinstance(current_value, Enum):
|
||
current_value = current_value.value
|
||
|
||
if current_value != new_value:
|
||
changes[field] = new_value
|
||
|
||
return changes if changes else None
|