fix: ModificationAudit 로직 고도화

Author: MU-Software

app/core/util/django_orm.py

@@ -0,0 +1,258 @@
+import collections.abc
+import contextlib
+import copy
+import datetime
+import types
+import typing
+import uuid
+
+from django.apps.registry import apps as django_apps
+from django.core.exceptions import ObjectDoesNotExist
+from django.db import models
+from django.db.models.fields.files import FieldFile
+from django.forms import model_to_dict
+
+
+def arbitrary_value_to_basic_type(value: typing.Any) -> str | int | float | bool | None:
+    """Convert an arbitrary value to a basic type that can be JSON serialized."""
+    if isinstance(value, (int, float, bool)):
+        return value
+    elif isinstance(value, str):
+        return value
+    elif isinstance(value, (datetime.datetime, datetime.date, datetime.time)):
+        return value.isoformat()
+    elif isinstance(value, uuid.UUID):
+        return str(value)
+    elif isinstance(value, FieldFile):
+        return value.name
+    elif value is None:
+        return None
+    else:
+        raise TypeError(f"Unsupported type for JSON serialization: {type(value)}")
+
+
+def model_to_identifier(instance: models.Model) -> str:
+    return f"mdl:{instance._meta.app_label}:{instance._meta.model_name}:{instance.pk}"
+
+
+def identifier_to_model(identifier: str) -> models.Model | None:
+    if not identifier.startswith("mdl:"):
+        raise ValueError(f"Invalid model identifier: {identifier}")
+
+    with contextlib.suppress(ValueError, LookupError, ObjectDoesNotExist):
+        _, app_label, model_name, pk = identifier.split(":", 3)
+        model_class: type[models.Model] = django_apps.get_model(app_label, model_name)
+        return model_class.objects.get(pk=pk)
+
+    return None
+
+
+def _model_to_jsonable_dict(  # noqa: C901
+    instance: models.Model,
+    converted_models: dict[str, dict[str, typing.Any]],
+    exclude: set[str] | None = None,
+    nested: bool = False,
+) -> dict:
+    all_fields = {
+        z for z in {f.name for f in instance._meta.fields} | set(instance._meta.fields_map.keys()) if z != "+"
+    }
+    exclude = {"created_at", "created_by", "updated_at", "updated_by", "deleted_at", "deleted_by"} | (exclude or set())
+
+    model_dict = model_to_dict(instance, exclude=exclude)
+    jsonable_model_dict: dict[str, str | int | float | bool | None] = {
+        "id": str(instance.pk) if isinstance(instance.pk, uuid.UUID) else instance.pk,
+        "pk": str(instance.pk) if isinstance(instance.pk, uuid.UUID) else instance.pk,
+        "_meta": types.SimpleNamespace(
+            app_label=instance._meta.app_label,
+            model_name=instance._meta.model_name,
+            verbose_name=instance._meta.verbose_name,
+            verbose_name_plural=instance._meta.verbose_name_plural,
+        ),
+    }
+
+    for field, value in model_dict.items():
+        if isinstance(value, (datetime.datetime, datetime.date, datetime.time)):
+            jsonable_model_dict[field] = value.isoformat()
+        elif isinstance(value, (uuid.UUID, int, str)):
+            # Is this field just a UUID | int | str, or is it a ForeignKey/OneToOneField?
+            # django.forms.model_to_dict will return a UUID for Proxy model fields,
+            # so we need to check if getattr(instance, field) is a model.Model instance.
+            model_attr = getattr(instance, field, None)
+            if isinstance(model_attr, models.Model):
+                key = model_to_identifier(model_attr)
+
+                jsonable_model_dict[f"{field}_id"] = (
+                    str(model_attr.pk) if isinstance(model_attr.pk, uuid.UUID) else model_attr.pk
+                )
+                jsonable_model_dict[field] = key
+                if key not in converted_models:
+                    converted_models[key] = _model_to_jsonable_dict(model_attr, converted_models, None, True)
+            else:
+                # If it's just a UUID | int field, we can store it directly.
+                jsonable_model_dict[field] = str(value) if isinstance(value, uuid.UUID) else value
+        elif isinstance(value, models.Model):
+            # simple ForeignKey or OneToOneField case.
+            key = model_to_identifier(value)
+            jsonable_model_dict[f"{field}_id"] = str(value.pk) if isinstance(value.pk, uuid.UUID) else value.pk
+            jsonable_model_dict[field] = key
+            if key not in converted_models:
+                converted_models[key] = _model_to_jsonable_dict(value, converted_models, None, True)
+        elif isinstance(value, (float, bool, str)) or value is None:
+            jsonable_model_dict[field] = value
+        elif isinstance(value, FieldFile):
+            jsonable_model_dict[field] = value.name
+        elif isinstance(value, collections.abc.Iterable):
+            # Is this field a ManyToOneRel | ManyToManyField, or is it a json or array field?
+            model_attr = getattr(instance, field, None)
+            if isinstance(model_attr, models.manager.BaseManager):
+                value = typing.cast(collections.abc.Iterable[models.Model], value)
+                jsonable_value = []
+                for v in value:
+                    key = model_to_identifier(v)
+                    jsonable_value.append(key)
+                    if key not in converted_models:
+                        converted_models[key] = _model_to_jsonable_dict(v, converted_models, None, True)
+                jsonable_model_dict[field] = jsonable_value
+            else:
+                jsonable_model_dict[field] = [arbitrary_value_to_basic_type(v) for v in value]
+        else:
+            raise TypeError(f"Unsupported field type: {type(value)} for field '{field}'")
+
+    if not nested:
+        for leftover_field in all_fields - exclude - jsonable_model_dict.keys():
+            # Possibly a many-to-many or many-to-one relation that was not included in the model_to_dict output.
+            model_attr = getattr(instance, leftover_field, None)
+            if not isinstance(model_attr, models.manager.BaseManager):
+                continue
+            if not (model_attr_objs := model_attr.all()):
+                continue
+
+            model_identifier_list = []
+            for model in model_attr_objs:
+                key = model_to_identifier(model)
+                model_identifier_list.append(key)
+                if key not in converted_models:
+                    converted_models[key] = _model_to_jsonable_dict(model, converted_models, None, True)
+            jsonable_model_dict[leftover_field] = model_identifier_list
+
+    return jsonable_model_dict
+
+
+def model_to_jsonable_dict(instance: models.Model):
+    key = model_to_identifier(instance)
+    converted_models: dict[str, dict[str, typing.Any]] = {}
+    converted_models[key] = _model_to_jsonable_dict(instance, converted_models)
+    return {"key": key, "model_data": converted_models}
+
+
+def get_diff_data_from_jsonized_models(
+    models_asis: dict[str, dict[str, typing.Any]],
+    models_tobe: dict[str, dict[str, typing.Any]],
+) -> dict[str, dict[str, typing.Any]]:
+    diff_models_data: dict[str, dict[str, typing.Any]] = {}
+
+    for model_identifier in set(models_asis.keys()).union(models_tobe.keys()):
+        model_asis = models_asis.get(model_identifier, {})
+        model_tobe = models_tobe.get(model_identifier, {})
+        if not (model_asis and model_tobe):
+            continue
+
+        diff_models_data[model_identifier] = {}
+        # Compare the contents of the models
+        for field_name in set(model_asis.keys()).union(model_tobe.keys()):
+            if field_name in model_asis and field_name not in model_tobe:
+                continue
+            if field_name not in model_asis and field_name in model_tobe:
+                diff_models_data[model_identifier][field_name] = model_tobe[field_name]
+                continue
+
+            value_a = model_asis[field_name]
+            value_b = model_tobe[field_name]
+            if value_a == value_b:
+                continue
+            if type(value_a) != type(value_b):  # noqa: E721
+                raise TypeError(
+                    f"Type mismatch for field '{field_name}' in model '{model_identifier}': "
+                    f"{type(value_a)} != {type(value_b)}"
+                )
+            diff_models_data[model_identifier][field_name] = value_b
+
+    return {k: v for k, v in diff_models_data.items() if v}
+
+
+def apply_diff_to_jsonized_models(
+    models_asis: dict[str, dict[str, typing.Any]],
+    models_diff: dict[str, dict[str, typing.Any]],
+) -> dict[str, dict[str, typing.Any]]:
+    updated_models = copy.deepcopy(models_asis)
+
+    for model_identifier, diff_data in models_diff.items():
+        if model_identifier not in updated_models:
+            updated_models[model_identifier] = diff_data
+            continue
+
+        if not isinstance(updated_models[model_identifier], dict):
+            raise TypeError(
+                f"Expected a dict for model '{model_identifier}', got {type(updated_models[model_identifier])}"
+            )
+
+        for field_name, new_value in diff_data.items():
+            updated_models[model_identifier][field_name] = new_value
+
+    return updated_models
+
+
+def json_to_simplenamespace(model_data: dict[str, dict[str, typing.Any]], key: str) -> types.SimpleNamespace:
+    # Resolve models first
+    resolved_models: dict[str, types.SimpleNamespace] = {}
+    for model_identifier, model_datum in model_data.items():
+        resolved_models[model_identifier] = types.SimpleNamespace(**model_datum)
+    # link identifiers in resolved models to their SimpleNamespace instances
+    for resolved_model in resolved_models.values():
+        for attr_name, attr_value in resolved_model.__dict__.items():
+            if isinstance(attr_value, str) and attr_value.startswith("mdl:"):
+                setattr(resolved_model, attr_name, resolved_models[attr_value])
+            elif isinstance(attr_value, list) and all(
+                isinstance(item, str) and item.startswith("mdl:") for item in attr_value
+            ):
+                resolved_many_rel_models = [resolved_models[item] for item in attr_value]
+                setattr(resolved_model, attr_name, resolved_many_rel_models)
+
+    return resolved_models[key]
+
+
+def apply_diff_to_model(models_data: dict[str, dict[str, typing.Any]]) -> list[models.Model]:
+    result_instances: list[models.Model] = []
+
+    for model_identifier, model_data in models_data.items():
+        if not (model_instance := identifier_to_model(model_identifier)):
+            raise ValueError(f"Model class not found for identifier: {model_identifier}")
+
+        # Apply the data to the model instance
+        for field_name, value in model_data.items():
+            if isinstance(value, str) and value.startswith("mdl:"):
+                # If the value is a model identifier, resolve it to a model instance
+                if not (related_model_instance := identifier_to_model(value)):
+                    raise ValueError(f"Related model not found for identifier: {value}")
+                setattr(model_instance, field_name, related_model_instance)
+            elif isinstance(value, list) and all(isinstance(item, str) and item.startswith("mdl:") for item in value):
+                # If the value is a list of model identifiers, resolve them to model instances
+                related_model_instances = [identifier_to_model(item) for item in value]
+                if None in related_model_instances:
+                    raise ValueError(f"One or more related models not found for identifiers: {value}")
+
+                old_related_models = {item.pk: item for item in getattr(model_instance, field_name, [])}
+                new_related_models = {item.pk: item for item in related_model_instances}
+
+                field = getattr(model_instance, field_name)
+                for del_pk in old_related_models.keys() - new_related_models.keys():
+                    field.remove(old_related_models[del_pk])
+                for add_pk in new_related_models.keys() - old_related_models.keys():
+                    field.add(new_related_models[add_pk])
+            else:
+                setattr(model_instance, field_name, value)
+
+        model_instance.save()
+        result_instances.append(model_instance)
+
+    return result_instances

app/participant_portal_api/migrations/0002_historicalmodificationaudit_original_data_and_more.py

@@ -0,0 +1,22 @@
+# Generated by Django 5.2 on 2025-07-11 09:01
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("participant_portal_api", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="historicalmodificationaudit",
+            name="original_data",
+            field=models.JSONField(default=dict),
+        ),
+        migrations.AddField(
+            model_name="modificationaudit",
+            name="original_data",
+            field=models.JSONField(default=dict),
+        ),
+    ]

app/participant_portal_api/models.py

@@ -1,51 +1,22 @@
-import contextlib
+import types
 import typing
 
 from core.models import BaseAbstractModel, BaseAbstractModelQuerySet
+from core.util.django_orm import (
+    apply_diff_to_jsonized_models,
+    apply_diff_to_model,
+    json_to_simplenamespace,
+    model_to_identifier,
+)
 from django.contrib.contenttypes.fields import GenericForeignKey
 from django.contrib.contenttypes.models import ContentType
-from django.core.exceptions import FieldDoesNotExist
 from django.db import models
 from event.presentation.models import Presentation, PresentationSpeaker
-from rest_framework import serializers
 from user.models import UserExt
 
 T = typing.TypeVar("T", bound=models.Model)
 
 
-def _apply_dict_to_model(instance: T, data: dict, save: bool = False) -> T:
-    model_class = type(instance)
-
-    for field_name, value in data.items():
-        with contextlib.suppress(FieldDoesNotExist):
-            field = model_class._meta.get_field(field_name)
-
-            if isinstance(field, models.ForeignKey):
-                # One to One or One to Many case
-                if isinstance(value, dict):
-                    if not (sub_instance := field.related_model.objects.filter(pk=value.get("id")).first()):
-                        continue
-                    setattr(instance, field_name, _apply_dict_to_model(sub_instance, value), save)
-                elif isinstance(value, (int, str)):
-                    if not (sub_instance := field.related_model.objects.filter(pk=value).first()):
-                        continue
-                    setattr(instance, field_name, sub_instance.pk if field_name.endswith("_id") else sub_instance)
-            elif isinstance(field, models.ManyToOneRel):
-                if save:
-                    if not all(isinstance(v, dict) and "id" in v for v in value):
-                        continue
-                    for sub_value in value:
-                        getattr(instance, field).filter(pk=sub_value.pop("id")).update(**sub_value)
-            else:
-                # 일반 필드 업데이트
-                setattr(instance, field_name, value)
-
-    if save:
-        instance.save()
-
-    return instance
-
-
 class ModificationAuditQuerySet(BaseAbstractModelQuerySet):
     def filter_by_instance(self, instance: models.Model) -> typing.Self:
         return self.filter_active().filter(
@@ -74,6 +45,7 @@ class Status(models.TextChoices):
 
     action = models.CharField(max_length=16, choices=Action.choices, default=Action.UPDATE)
     status = models.CharField(max_length=32, choices=Status.choices, default=Status.REQUESTED)
+    original_data = models.JSONField(default=dict, blank=False)
     modification_data = models.JSONField(default=dict, blank=False)
 
     instance_type = models.ForeignKey(ContentType, on_delete=models.PROTECT)
@@ -98,32 +70,23 @@ class Meta:
     def __str__(self) -> str:
         return str(self.instance)
 
-    def get_applied_data(self, serializer_class: type[serializers.ModelSerializer]) -> dict:
-        one_to_many: dict[str, dict[str, dict[str, typing.Any]]] = {
-            k: {sv["id"]: sv for sv in v}
-            for k, v in self.modification_data.items()
-            if isinstance(v, list) and all(isinstance(i, dict) and "id" in i for i in v)
-        }
-
-        modified_instance = _apply_dict_to_model(instance=self.instance, data=self.modification_data, save=False)
-        modified_data = serializer_class(instance=modified_instance).data
-
-        for field_name, mod_values in one_to_many.items():
-            if field_name not in modified_data:
-                continue
-
-            for field_value in modified_data[field_name]:
-                if not (isinstance(field_value, dict) and (value_id := field_value.get("id"))):
-                    continue
+    @property
+    def instance_identifier(self) -> str:
+        return model_to_identifier(self.instance)
 
-                if value_id in mod_values:
-                    # 기존 값에 수정된 값을 병합합니다.
-                    field_value.update(mod_values[value_id])
+    @property
+    def fake_original_instance(self) -> types.SimpleNamespace:
+        return json_to_simplenamespace(self.original_data)
 
-        return modified_data
+    @property
+    def fake_modified_instance(self) -> types.SimpleNamespace:
+        updated_data = apply_diff_to_jsonized_models(self.original_data, self.modification_data)
+        return json_to_simplenamespace(updated_data)
 
     def apply_modification(self) -> models.Model:
-        return _apply_dict_to_model(instance=self.instance, data=self.modification_data, save=True)
+        apply_diff_to_model(self.modification_data)
+        self.instance.refresh_from_db()
+        return self.instance
 
 
 class ModificationAuditComment(BaseAbstractModel):