Django，级联移动到单独的表而不是级联删除

Posted 2023-02-23

【中文标题】Django，级联移动到单独的表而不是级联删除

【英文标题】：Django, cascading move to a separate table instead of cascading delete

【发布时间】：2016-03-26 14:29:36

【问题描述】：

我想在我们delete时保留数据

而不是 soft-delete（使用 is_deleted 字段），我想将数据移动到另一个表（对于已删除的行）

https://***.com/a/26125927/433570

我也不知道策略的名称是什么。叫存档？两表删除？

为了完成这项工作，

我需要能够做到

对于给定的对象（将被删除），查找具有该对象的外键或一对一键的所有其他对象。 （这可以通过https://***.com/a/2315053/433570 完成，实际上比这更难，代码还不够）

插入一个新对象并让 #1 中找到的所有对象都指向这个新对象

删除对象

（本质上我是在做级联移动而不是级联删除，1~3步应该以递归方式完成）

为此创建一个支持delete() 和undelete() 的对象和查询集的mixin 是最方便的。

有人做过这样的吗？

【问题讨论】：

我编辑了这个问题，那么如何让这个问题脱离put-on-hold 状态？

你好，请重新打开这个:(

【参考方案1】：

我自己实现了这个，我正在分享我的发现。

存档

第一次归档相当容易，因为我放宽了归档表的外键约束。

您不能像在现实世界中那样保留存档世界中的所有约束，因为您要删除的对象所引用的内容不会在存档世界中。 （因为它不会被删除）

这可以通过 mixin 来完成（系统地）

基本上，您使用 cascade 创建存档对象，然后删除原始对象。

取消归档

另一方面，取消归档更难，因为您需要确认外键约束。 这不能系统地完成。

这与 Django rest 框架等序列化程序不会神奇地创建相关对象的原因相同。你必须知道对象图和约束。

这就是为什么没有库或 mixin 支持这一点的原因。

不管怎样，我在下面分享我的 mixin 代码。

 class DeleteModelQuerySet(object):
     '''
     take a look at django.db.models.deletion
     '''

     def hard_delete(self):
         super().delete()

     def delete(self):
         if not self.is_archivable():
             super().delete()
             return

         archive_object_ids = []
         seen = []

         collector = NestedObjects(using='default')  # or specific database
         collector.collect(list(self))
         collector.sort()

         with transaction.atomic():

             for model, instances in six.iteritems(collector.data):

                 if model in self.model.exclude_models_from_archive():
                     continue

                 assert hasattr(model, "is_archivable"), 
                     "model  doesn't know about archive".format(model)
                 

                 if not model.is_archivable():
                     # just delete
                     continue

                 for instance in instances:

                     if instance in seen:
                         continue
                     seen.append(instance)

                     for ptr in six.itervalues(instance._meta.parents):
                         # add parents to seen
                         if ptr:
                             seen.append(getattr(instance, ptr.name))

                     archive_object = model.create_archive_object(instance)
                     archive_object_ids.append(archive_object.id)

             # real delete
             super().delete()

         archive_objects = self.model.get_archive_model().objects.filter(id__in=archive_object_ids)
         return archive_objects

     def undelete(self):

         with transaction.atomic():
             self.unarchive()

             super().delete()

     def is_archivable(self):
         # if false, we hard delete instead of archive
         return self.model.is_archivable()

     def unarchive(self):

         for obj_archive in self:
             self.model.create_live_object(obj_archive)


 class DeleteModelMixin(models.Model):

     @classmethod
     def is_archivable(cls):
         # override if you don't want to archive and just delete
         return True

     def get_deletable_objects(self):
         collector = NestedObjects(using='default')  # or specific database
         collector.collect(list(self))
         collector.sort()
         deletable_data = collector.data

         return deletable_data

     @classmethod
     def create_archive_object(cls, obj):
         # http://***.com/q/21925671/433570
         # d = cls.objects.filter(id=obj.id).values()[0]

         d = obj.__dict__.copy()
         remove_fields = []
         for field_name, value in six.iteritems(d):
             try:
                 obj._meta.get_field(field_name)
             except FieldDoesNotExist:
                 remove_fields.append(field_name)
         for remove_field in remove_fields:
             d.pop(remove_field)

         cls.convert_to_archive_dictionary(d)

         # print(d)

         archive_object = cls.get_archive_model().objects.create(**d)
         return archive_object

     @classmethod
     def create_live_object(cls, obj):

         # index error, dont know why..
         # d = cls.objects.filter(id=obj.id).values()[0]

         d = obj.__dict__.copy()

         remove_fields = [cls.convert_to_archive_field_name(field_name) + '_id' for field_name in cls.get_twostep_field_names()]
         for field_name, value in six.iteritems(d):
             try:
                 obj._meta.get_field(field_name)
             except FieldDoesNotExist:
                 remove_fields.append(field_name)

         for remove_field in remove_fields:
             d.pop(remove_field)

         cls.convert_to_live_dictionary(d)

         live_object = cls.get_live_model().objects.create(**d)
         return live_object

     @classmethod
     def get_archive_model_name(cls):
         return 'Archive'.format(cls._meta.model_name)

     @classmethod
     def get_live_model_name(cls):

         if cls._meta.model_name.endswith("archive"):
             length = len("Archive")
             return cls._meta.model_name[:-length]
         return cls._meta.model_name

     @classmethod
     def get_archive_model(cls):
         # http://***.com/a/26126935/433570
         return apps.get_model(app_label=cls._meta.app_label, model_name=cls.get_archive_model_name())

     @classmethod
     def get_live_model(cls):
         return apps.get_model(app_label=cls._meta.app_label, model_name=cls.get_live_model_name())

     @classmethod
     def is_archive_model(cls):
         if cls._meta.model_name.endswith("Archive"):
             return True
         return False

     @classmethod
     def is_live_model(cls):
         if cls.is_archive_model():
             return False
         return True

     def make_referers_point_to_archive(self, archive_object, seen):

         instance = self

         for related in get_candidate_relations_to_delete(instance._meta):
             accessor_name = related.get_accessor_name()

             if accessor_name.endswith('+') or accessor_name.lower().endswith("archive"):
                 continue

             referers = None

             if related.one_to_one:
                 referer = getattr(instance, accessor_name, None)
                 if referer:
                     referers = type(referer).objects.filter(id=referer.id)
             else:
                 referers = getattr(instance, accessor_name).all()

             refering_field_name = '_archive'.format(related.field.name)

             if referers:
                 assert hasattr(referers, 'is_archivable'), 
                     "referers is not archivable: referer_cls".format(
                         referer_cls=referers.model
                     )
                 

                 archive_referers = referers.delete(seen=seen)
                 if referers.is_archivable():
                     archive_referers.update(**refering_field_name: archive_object)

     def hard_delete(self):
         super().delete()

     def delete(self, *args, **kwargs):
         self._meta.model.objects.filter(id=self.id).delete()

     def undelete(self, commit=True):
         self._meta.model.objects.filter(id=self.id).undelete()

     def unarchive(self, commit=True):
         self._meta.model.objects.filter(id=self.id).unarchive()

     @classmethod
     def get_archive_field_names(cls):
         raise NotImplementedError('get_archive_field_names() must be implemented')

     @classmethod
     def convert_to_archive_dictionary(cls, d):

         field_names = cls.get_archive_field_names()
         for field_name in field_names:
             field_name = '_id'.format(field_name)
             archive_field_name = cls.convert_to_archive_field_name(field_name)
             d[archive_field_name] = d.pop(field_name)

     @classmethod
     def convert_to_live_dictionary(cls, d):

         field_names = list(set(cls.get_archive_field_names()) - set(cls.get_twostep_field_names()))

         for field_name in field_names:
             field_name = '_id'.format(field_name)
             archive_field_name = cls.convert_to_archive_field_name(field_name)
             d[field_name] = d.pop(archive_field_name)

     @classmethod
     def convert_to_archive_field_name(cls, field_name):
         if field_name.endswith('_id'):
             length = len('_id')
             return '_archive_id'.format(field_name[:-length])
         return '_archive'.format(field_name)

     @classmethod
     def convert_to_live_field_name(cls, field_name):
         if field_name.endswith('_archive_id'):
             length = len('_archive_id')
             return '_id'.format(field_name[:-length])
         if field_name.endswith('archive'):
             length = len('_archive')
             return ''.format(field_name[:-length])
         return None

     @classmethod
     def get_twostep_field_names(cls):
         return []

     @classmethod
     def exclude_models_from_archive(cls):
         # excluded model can be deleted if referencing to me
         # or just lives if I reference him
         return []

     class Meta:
         abstract = True

【讨论】：

我对这段代码做了一个要点并添加了缺失的导入：gist.github.com/adamlwgriffiths/4fb6a47bb84ec6e34424

【参考方案2】：

如果您正在寻找任何用于特定服务或功能的 3rd 方 django-package，如果您对现有的没有任何想法，您可以随时搜索 www.djangopackages.com。它还将为您提供包之间的比较表，以帮助您做出正确的选择。 基于table here: django-reversion 是最常用的，有稳定的版本，在 github 活跃社区，最后一次更新是 3 天前，这意味着项目维护得很好，很可靠。

要安装 django-reversion，请按以下步骤操作：

1.使用pip安装：pip install django-reversion.

2.将“reversion”添加到INSTALLED_APPS。

3.运行manage.py migrate

查看here了解更多详细信息和配置

【讨论】：

我的意思是soft-delete 是指使用is_delete 的方法。我编辑了操作。很抱歉造成混乱。

我看了一下django-reversion，我觉得不支持delete-undelete

这些似乎都使用deleted_on 字段，而不是将模型移动到不同的表。