Django，级联移动到一个单独的table而不是级联删除
Django, cascading move to a separate table instead of cascading delete
我想在 delete
时保留数据
而不是 soft-delete（使用 is_deleted 字段），我想将数据移动到另一个 table（对于已删除的行）
我也不知道策略的名称是什么。叫归档？二-table删除?
为了完成这项工作，
我需要能够做到
对于给定的对象（将被删除），找到所有其他对象具有该对象的外键或一对一键。 （这可以通过 完成，实际上比那个更难，那个代码是不够的）
插入一个新对象并让在#1 中找到的所有对象指向这个新对象
删除对象
（本质上我正在做级联移动而不是级联删除，1~3 步应该以递归方式完成）
为此创建一个支持对象和查询集 delete() 和 undelete() 的混入是最方便的。
有人创造过这样的吗？
如果您正在寻找特定服务或功能的任何第 3 方 django 包，如果您对现有包一无所知，可以随时在 www.djangopackages.com 中搜索。它还会为您提供包之间的比较table，以帮助您做出正确的选择。
基于 table here：django-reversion 是最常用的，有 stable 版本，活跃社区 github 最后更新是 3 天之前，这意味着该项目维护良好且可靠。
要安装 django-reversion，请按照以下步骤操作：
1.Install 点数：pip install django-reversion.
2.Add 'reversion' 到 INSTALLED_APPS.
3.Run manage.py migrate
检查 here 了解更多详细信息和配置
我自己实现了这个，我正在分享我的发现。
存档
第一次归档相当容易，因为我放宽了对归档表的外键约束。
您不能像在现实世界中那样保留存档世界中的所有约束，因为您要删除的对象所引用的内容不会在存档世界中。 （因为不会被删除）
这可以通过 mixin（系统地）完成
基本上，您使用级联创建存档对象，然后删除原始对象。
取消归档
另一方面，解压比较困难，因为你需要确认外键约束。
这无法系统地完成。
这与Django rest framework 等序列化程序不会神奇地创建相关对象的原因相同。你必须知道对象图和约束。
这就是为什么没有库或 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):
#
# 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):
#
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
我想在 delete
而不是 soft-delete（使用 is_deleted 字段），我想将数据移动到另一个 table（对于已删除的行）
我也不知道策略的名称是什么。叫归档？二-table删除?
为了完成这项工作，
我需要能够做到
对于给定的对象（将被删除），找到所有其他对象具有该对象的外键或一对一键。 （这可以通过 完成，实际上比那个更难，那个代码是不够的）
插入一个新对象并让在#1 中找到的所有对象指向这个新对象
删除对象
（本质上我正在做级联移动而不是级联删除，1~3 步应该以递归方式完成）
为此创建一个支持对象和查询集 delete() 和 undelete() 的混入是最方便的。
有人创造过这样的吗？
如果您正在寻找特定服务或功能的任何第 3 方 django 包，如果您对现有包一无所知，可以随时在 www.djangopackages.com 中搜索。它还会为您提供包之间的比较table，以帮助您做出正确的选择。 基于 table here：django-reversion 是最常用的，有 stable 版本，活跃社区 github 最后更新是 3 天之前，这意味着该项目维护良好且可靠。
要安装 django-reversion，请按照以下步骤操作：
1.Install 点数：pip install django-reversion.
2.Add 'reversion' 到 INSTALLED_APPS.
3.Run manage.py migrate
检查 here 了解更多详细信息和配置
我自己实现了这个，我正在分享我的发现。
存档
第一次归档相当容易，因为我放宽了对归档表的外键约束。
您不能像在现实世界中那样保留存档世界中的所有约束，因为您要删除的对象所引用的内容不会在存档世界中。 （因为不会被删除）
这可以通过 mixin（系统地）完成
基本上，您使用级联创建存档对象，然后删除原始对象。
取消归档
另一方面，解压比较困难，因为你需要确认外键约束。
这无法系统地完成。
这与Django rest framework 等序列化程序不会神奇地创建相关对象的原因相同。你必须知道对象图和约束。
这就是为什么没有库或 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):
#
# 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):
#
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