Harden image fetch algorithm

Prepare for database integration

modules/shared/DataClassJson.py

@@ -0,0 +1,185 @@
+from dataclasses import dataclass, field, fields
+from typing import Dict, Any, Optional
+import warnings
+
+# Определим базовый класс для удобного наследования
+@dataclass
+class DataClassJson:
+    _forwarding: Dict[str, type] = field(default_factory=dict)
+    _key_field: str = 'key'  # Поле, которое будет использоваться как ключ
+    fixed: bool = False
+
+    # Скрытые поля для хранения данных
+    _key: Optional[str] = None
+    other_data: Optional[Dict[str, Any]] = None
+
+    # Пример поля, которое будет использоваться в _forwarding
+    # Должно быть переопределено в дочерних классах
+    key: Optional[str] = None
+
+    def __post_init__(self):
+        if self._key is not None:
+            self.key = self._key
+
+    @property
+    def key(self) -> Optional[str]:
+        return self._key
+
+    @key.setter
+    def key(self, value: str):
+        self._key = value
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'DataClassJson':
+        # Создаем экземпляр класса
+        instance = cls()
+        instance.fixed = data.get('fixed', False)
+        instance.other_data = None
+
+        # Список всех полей
+        excluded_fields = {f.name for f in fields(DataClassJson)}
+        all_fields = {f.name for f in fields(cls) if f.name not in excluded_fields and not f.name.startswith('_')}
+
+        # Обрабатываем поля из _forwarding
+        handled_keys = set()
+        field_values = {}
+
+        for key, value in data.items():
+            if key in handled_keys:
+                continue
+
+            if key in instance._forwarding:
+                target_type = instance._forwarding[key]
+                if isinstance(value, dict):
+                    # Обрабатываем словарь
+                    sub_instance = target_type.from_dict(value)
+                    field_values[key] = sub_instance
+                    handled_keys.add(key)
+                elif isinstance(value, list):
+                    # Обрабатываем список словарей
+                    results = []
+                    for item in value:
+                        if isinstance(item, dict):
+                            sub_instance = target_type.from_dict(item)
+                            results.append(sub_instance)
+                        else:
+                            # Если элемент не словарь, записываем в other_data
+                            warnings.warn(f"Non-dict value {item} in list for field '{key}' will be added to 'other_data'")
+                            if instance.other_data is None:
+                                instance.other_data = {}
+                            instance.other_data[key] = item  # Сохраняем оригинал
+                    field_values[key] = results
+                    handled_keys.add(key)
+                else:
+                    # Если не словарь и не список, тоже добавляем в other_data
+                    warnings.warn(f"Non-dict/list value {value} for field '{key}' will be added to 'other_data'")
+                    if instance.other_data is None:
+                        instance.other_data = {}
+                    instance.other_data[key] = value
+            else:
+                # Обычное поле
+                if key in all_fields:
+                    field_values[key] = value
+                    handled_keys.add(key)
+                else:
+                    # Неизвестное поле, добавляем в other_data
+                    warnings.warn(f"Unknown field '{key}', adding to 'other_data'")
+                    if instance.other_data is None:
+                        instance.other_data = {}
+                    instance.other_data[key] = value
+
+        # Заполняем обычные поля
+        for key, value in field_values.items():
+            setattr(instance, key, value)
+
+        # Устанавливаем ключ, если есть
+        if hasattr(instance, '_key_field') and instance._key_field in data:
+            instance.key = data[instance._key_field]
+
+        # Проверяем флаг fixed и other_data
+        if instance.fixed and instance.other_data is not None:
+            raise ValueError("Cannot serialize with fixed=True and non-empty other_data")
+
+        return instance
+
+    def to_dict(self) -> Dict[str, Any]:
+        result = {}
+        excluded_fields = {f.name for f in fields(DataClassJson)}
+        field_names = [f.name for f in fields(self) if f.name not in excluded_fields and not f.name.startswith('_')]
+
+        for field_name in field_names:
+            if not hasattr(self, field_name):
+                result[field_name] = None
+                warnings.warn(f'object not have field {field_name}, something went wrong')
+                continue
+            value = getattr(self, field_name)
+            if not value:
+                result[field_name] = None
+                warnings.warn(f'object not have data in field {field_name}, it may be correct situation')
+                continue
+
+            if field_name in self._forwarding:
+                target_type = self._forwarding[field_name]
+                result[field_name] = list()
+                single = False
+                if not isinstance(value, list):
+                    single = True
+                    value = [value]
+                for v in value:
+                    try:
+                        v = v.to_dict()
+                    except Exception as e:
+                        warnings.warn(str(e))
+                    finally:
+                        result[field_name].append(v)
+                if single: result[field_name] = result[field_name][0]
+                continue
+            else: result[field_name] = value
+
+        # Добавляем other_data, если есть
+        if self.other_data and isinstance(self.other_data, dict):
+            for key, value in self.other_data.items():
+                if key not in result:
+                    result[key] = value
+                else:
+                    if not isinstance(result[key], list): result[key] = [result[key]]
+                    if not isinstance(value, list): value = [value]
+                    result[key].extend(value)
+
+        return result
+
+# Пример использования:
+@dataclass
+class Person(DataClassJson):
+    name: Optional[str] = None
+    age: Optional[int] = None
+    email: Optional[str] = None
+
+    def __post_init__(self):
+        super().__post_init__()
+        self._forwarding = {}
+        self._key_field = 'name'
+
+@dataclass
+class User(DataClassJson):
+    id: Optional[list] = None
+    username: Optional[str] = None
+    person: Optional[Person] = None
+
+    def __post_init__(self):
+        super().__post_init__()
+        self._forwarding = {'person': Person}
+        self._key_field = 'username'
+
+# Пример десериализации:
+if __name__ == "__main__":
+    data = {
+        "id": [1,2,3,4,5,6],
+        "username": "user1",
+        "person": None,
+        "extra_field": "should_be_in_other_data"
+    }
+
+    user = User.from_dict(data)
+    data2 = user.to_dict()
+    print(user.to_dict())