python - Django 克隆递归对象

Question

以前我想递归克隆对象时遇到问题。 我知道克隆对象的简单方法是这样的:

obj = Foo.objects.get(pk=<some_existing_pk>)
obj.pk = None
obj.save()

但是，我想做的更深入。例如，我有一个 models.py

class Post(TimeStampedModel):
    author = models.ForeignKey(User, related_name='posts',
                               on_delete=models.CASCADE)
    title = models.CharField(_('Title'), max_length=200)
    content = models.TextField(_('Content'))

    ...


class Comment(TimeStampedModel):
    author = models.ForeignKey(User, related_name='comments',
                               on_delete=models.CASCADE)
    post = models.ForeignKey(Post, on_delete=models.CASCADE)
    comment = models.TextField(_('Comment'))

    ...


class CommentAttribute(TimeStampedModel):
    comment = models.OneToOneField(Comment, related_name='comment_attribute',
                                   on_delete=models.CASCADE)
    is_bookmark = models.BooleanField(default=False)

    ...


class PostComment(TimeStampedModel):
    post = models.ForeignKey(Post, related_name='post_comments',
                             on_delete=models.CASCADE)
    comments = models.ManyToManyField(Comment)

    ...

When I clone the parent object from Post, the child objects like Comment, CommentAttribute and PostComment will also cloned by following new cloned Post objects. The child models are dynamically. So, I want to make it simple by creating the tool like object cloner.

下面的这段代码是我所做的；

from django.db.utils import IntegrityError


class ObjectCloner(object):
    """
    [1]. The simple way with global configuration:
    >>> cloner = ObjectCloner()
    >>> cloner.set_objects = [obj1, obj2]   # or can be queryset
    >>> cloner.include_childs = True
    >>> cloner.max_clones = 1
    >>> cloner.execute()

    [2]. Clone the objects with custom configuration per-each objects.
    >>> cloner = ObjectCloner()
    >>> cloner.set_objects = [
        {
            'object': obj1,
            'include_childs': True,
            'max_clones': 2
        },
        {
            'object': obj2,
            'include_childs': False,
            'max_clones': 1
        }
    ]
    >>> cloner.execute()
    """
    set_objects = []            # list/queryset of objects to clone.
    include_childs = True       # include all their childs or not.
    max_clones = 1              # maximum clone per-objects.

    def clone_object(self, object):
        """
        function to clone the object.
        :param `object` is an object to clone, e.g: <Post: object(1)>
        :return new object.
        """
        try:
            object.pk = None
            object.save()
            return object
        except IntegrityError:
            return None

    def clone_childs(self, object):
        """
        function to clone all childs of current `object`.
        :param `object` is a cloned parent object, e.g: <Post: object(1)>
        :return
        """
        # bypass the none object.
        if object is None:
            return

        # find the related objects contains with this current object.
        # e.g: (<ManyToOneRel: app.comment>,)
        related_objects = object._meta.related_objects

        if len(related_objects) > 0:
            for relation in related_objects:
                # find the related field name in the child object, e.g: 'post'
                remote_field_name = relation.remote_field.name

                # find all childs who have the same parent.
                # e.g: childs = Comment.objects.filter(post=object)
                childs = relation.related_model.objects.all()

                for old_child in childs:
                    new_child = self.clone_object(old_child)

                    if new_child is not None:
                        # FIXME: When the child field as M2M field, we gote this error.
                        # "TypeError: Direct assignment to the forward side of a many-to-many set is prohibited. Use comments.set() instead."
                        # how can I clone that M2M values?
                        setattr(new_child, remote_field_name, object)
                        new_child.save()

                    self.clone_childs(new_child)
        return

    def execute(self):
        include_childs = self.include_childs
        max_clones = self.max_clones
        new_objects = []

        for old_object in self.set_objects:
            # custom per-each objects by using dict {}.
            if isinstance(old_object, dict):
                include_childs = old_object.get('include_childs', True)
                max_clones = old_object.get('max_clones', 1)
                old_object = old_object.get('object')  # assigned as object or None.

            for _ in range(max_clones):
                new_object = self.clone_object(old_object)
                if new_object is not None:
                    if include_childs:
                        self.clone_childs(new_object)
                    new_objects.append(new_object)

        return new_objects

但是，问题是当子字段作为 M2M 字段时，我们得到了这个错误。

>>> cloner.set_objects = [post]
>>> cloner.execute()
Traceback (most recent call last):
  File "<console>", line 1, in <module>
  File "/home/agus/envs/env-django-cloner/django-object-cloner/object_cloner_demo/app/utils.py", line 114, in execute
    self.clone_childs(new_object)
  File "/home/agus/envs/env-django-cloner/django-object-cloner/object_cloner_demo/app/utils.py", line 79, in clone_childs
    self.clone_childs(new_child)
  File "/home/agus/envs/env-django-cloner/django-object-cloner/object_cloner_demo/app/utils.py", line 76, in clone_childs
    setattr(new_child, remote_field_name, object)
  File "/home/agus/envs/env-django-cloner/lib/python3.7/site-packages/django/db/models/fields/related_descriptors.py", line 546, in __set__
    % self._get_set_deprecation_msg_params(),
TypeError: Direct assignment to the forward side of a many-to-many set is prohibited. Use comments.set() instead.
>>> 

错误来自 setattr(...) 和 “改用 comments.set()”，但我仍然对如何更新 m2m 值感到困惑？

new_child = self.clone_object(old_child)

if new_child is not None:
    setattr(new_child, remote_field_name, object)
    new_child.save()

我也试过下面这个片段，但仍然有一个错误。克隆的 m2m 对象很多且未填充到 m2m 值中。

if new_child is not None:
    # check the object_type
    object_type = getattr(new_child, remote_field_name)

    if hasattr(object_type, 'pk'):
        # this mean is `object_type` as real object.
        # so, we can directly use the `setattr(...)`
        # to update the old relation value with new relation value.
        setattr(new_child, remote_field_name, object)

    elif hasattr(object_type, '_queryset_class'):
        # this mean is `object_type` as m2m queryset (ManyRelatedManager).
        # django.db.models.fields.related_descriptors.\
        # create_forward_many_to_many_manager.<locals>.ManyRelatedManager

        # check the old m2m values, and assign into new object.
        # FIXME: IN THIS CASE STILL GOT AN ERROR
        old_m2m_values = getattr(old_child, remote_field_name).all()
        object_type.add(*old_m2m_values)

    new_child.save()

Answer 1

我试图用一些有效的代码来解决这个有趣的问题......这比我最初想象的要难!

我偏离了您最初的解决方案，因为我在遵循 ObjectCloner 逻辑时遇到了一些困难。

下面给出了我能想到的最简单的解决方案；我没有使用类，而是选择了一个辅助函数 clone_object()，它处理单个对象。

您当然可以使用第二个函数来处理对象列表或查询集，方法是扫描序列并多次调用 clone_object()。

def clone_object(obj, attrs={}):

    # we start by building a "flat" clone
    clone = obj._meta.model.objects.get(pk=obj.pk)
    clone.pk = None

    # if caller specified some attributes to be overridden, 
    # use them
    for key, value in attrs.items():
        setattr(clone, key, value)

    # save the partial clone to have a valid ID assigned
    clone.save()

    # Scan field to further investigate relations
    fields = clone._meta.get_fields()
    for field in fields:

        # Manage M2M fields by replicating all related records 
        # found on parent "obj" into "clone"
        if not field.auto_created and field.many_to_many:
            for row in getattr(obj, field.name).all():
                getattr(clone, field.name).add(row)

        # Manage 1-N and 1-1 relations by cloning child objects
        if field.auto_created and field.is_relation:
            if field.many_to_many:
                # do nothing
                pass
            else:
                # provide "clone" object to replace "obj" 
                # on remote field
                attrs = {
                    field.remote_field.name: clone
                }
                children = field.related_model.objects.filter(**{field.remote_field.name: obj})
                for child in children:
                    clone_object(child, attrs)

    return clone

使用 Python 3.7.6 和 Django 3.0.6 测试的 POC 示例项目已保存在 github 上的公共(public)存储库中:

https://github.com/morlandi/test-django-clone