feat:add deepcopy tool

Author: goasleep

scrapegraphai/utils/copy.py

@@ -0,0 +1,78 @@
+import copy
+from typing import Any, Dict, Optional
+
+
+def safe_deepcopy(obj: Any, memo: Optional[Dict[int, Any]] = None) -> Any:
+    """
+    Attempts to create a deep copy of the object using `copy.deepcopy`
+    whenever possible. If that fails, it falls back to custom deep copy
+    logic or returns the original object.
+
+    Args:
+        obj (Any): The object to be copied, which can be of any type.
+        memo (Optional[Dict[int, Any]]): A dictionary used to track objects
+            that have already been copied to handle circular references.
+            If None, a new dictionary is created.
+
+    Returns:
+        Any: A deep copy of the object if possible; otherwise, a shallow
+        copy if deep copying fails; if neither is possible, the original
+        object is returned.
+    """
+
+    if memo is None:
+        memo = {}
+
+    if id(obj) in memo:
+        return memo[id(obj)]
+
+    try:
+        # Try to use copy.deepcopy first
+        return copy.deepcopy(obj, memo)
+    except (TypeError, AttributeError):
+        # If deepcopy fails, handle specific types manually
+
+        # Handle dictionaries
+        if isinstance(obj, dict):
+            new_obj = {}
+            memo[id(obj)] = new_obj
+            for k, v in obj.items():
+                new_obj[k] = safe_deepcopy(v, memo)
+            return new_obj
+
+        # Handle lists
+        elif isinstance(obj, list):
+            new_obj = []
+            memo[id(obj)] = new_obj
+            for v in obj:
+                new_obj.append(safe_deepcopy(v, memo))
+            return new_obj
+
+        # Handle tuples (immutable, but might contain mutable objects)
+        elif isinstance(obj, tuple):
+            new_obj = tuple(safe_deepcopy(v, memo) for v in obj)
+            memo[id(obj)] = new_obj
+            return new_obj
+
+        # Handle frozensets (immutable, but might contain mutable objects)
+        elif isinstance(obj, frozenset):
+            new_obj = frozenset(safe_deepcopy(v, memo) for v in obj)
+            memo[id(obj)] = new_obj
+            return new_obj
+
+        # Handle objects with attributes
+        elif hasattr(obj, "__dict__"):
+            new_obj = obj.__new__(obj.__class__)
+            for attr in obj.__dict__:
+                setattr(new_obj, attr, safe_deepcopy(getattr(obj, attr), memo))
+            memo[id(obj)] = new_obj
+            return new_obj
+
+        # Attempt shallow copy as a fallback
+        try:
+            return copy.copy(obj)
+        except (TypeError, AttributeError):
+            pass
+
+        # If all else fails, return the original object
+        return obj

tests/utils/copy_utils_test.py

@@ -0,0 +1,170 @@
+import copy
+import pytest
+
+# Assuming the custom_deepcopy function is imported or defined above this line
+from scrapegraphai.utils.copy import safe_deepcopy
+
+
+class NormalObject:
+    def __init__(self, value):
+        self.value = value
+        self.nested = [1, 2, 3]
+
+    def __deepcopy__(self, memo):
+        raise TypeError("Forcing fallback")
+
+
+class NonDeepcopyable:
+    def __init__(self, value):
+        self.value = value
+
+    def __deepcopy__(self, memo):
+        raise TypeError("Forcing shallow copy fallback")
+
+
+class WithoutDict:
+    __slots__ = ["value"]
+
+    def __init__(self, value):
+        self.value = value
+
+    def __deepcopy__(self, memo):
+        raise TypeError("Forcing shallow copy fallback")
+
+    def __copy__(self):
+        return self
+
+
+class NonCopyableObject:
+    __slots__ = ["value"]
+
+    def __init__(self, value):
+        self.value = value
+
+    def __deepcopy__(self, memo):
+        raise TypeError("fail deep copy ")
+
+    def __copy__(self):
+        raise TypeError("fail shallow copy")
+
+
+def test_deepcopy_simple_dict():
+    original = {"a": 1, "b": 2, "c": [3, 4, 5]}
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj == original
+    assert copy_obj is not original
+    assert copy_obj["c"] is not original["c"]
+
+
+def test_deepcopy_simple_list():
+    original = [1, 2, 3, [4, 5]]
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj == original
+    assert copy_obj is not original
+    assert copy_obj[3] is not original[3]
+
+
+def test_deepcopy_with_tuple():
+    original = (1, 2, [3, 4])
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj == original
+    assert copy_obj is not original
+    assert copy_obj[2] is not original[2]
+
+
+def test_deepcopy_with_frozenset():
+    original = frozenset([1, 2, 3, (4, 5)])
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj == original
+    assert copy_obj is not original
+
+
+def test_deepcopy_with_object():
+    original = NormalObject(10)
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj.value == original.value
+    assert copy_obj is not original
+    assert copy_obj.nested is not original.nested
+
+
+def test_deepcopy_with_custom_deepcopy_fallback():
+    original = {"origin": NormalObject(10)}
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj is not original
+    assert copy_obj["origin"].value == original["origin"].value
+
+
+def test_shallow_copy_fallback():
+    original = {"origin": NonDeepcopyable(10)}
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj is not original
+    assert copy_obj["origin"].value == original["origin"].value
+
+
+def test_circular_reference():
+    original = []
+    original.append(original)
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj is not original
+    assert copy_obj[0] is copy_obj
+
+
+def test_memoization():
+    original = {"a": 1, "b": 2}
+    memo = {}
+    copy_obj = safe_deepcopy(original, memo)
+    assert copy_obj is memo[id(original)]
+
+
+def test_deepcopy_object_without_dict():
+    original = {"origin": WithoutDict(10)}
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj["origin"].value == original["origin"].value
+    assert copy_obj is not original
+    assert copy_obj["origin"] is original["origin"]
+    assert (
+        hasattr(copy_obj["origin"], "__dict__") is False
+    )  # Ensure __dict__ is not present
+
+    original = [WithoutDict(10)]
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj[0].value == original[0].value
+    assert copy_obj is not original
+    assert copy_obj[0] is original[0]
+
+    original = (WithoutDict(10),)
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj[0].value == original[0].value
+    assert copy_obj is not original
+    assert copy_obj[0] is original[0]
+
+    original_item = WithoutDict(10)
+    original = set([original_item])
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj is not original
+    copy_obj_item = copy_obj.pop()
+    assert copy_obj_item.value == original_item.value
+    assert copy_obj_item is original_item
+
+    original_item = WithoutDict(10)
+    original = frozenset([original_item])
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj is not original
+    copy_obj_item = list(copy_obj)[0]
+    assert copy_obj_item.value == original_item.value
+    assert copy_obj_item is original_item
+
+def test_memo():
+    obj = NormalObject(10)
+    original = {"origin": obj}
+    memo = {id(original):obj}
+    copy_obj = safe_deepcopy(original, memo)
+    assert copy_obj is memo[id(original)]
+
+def test_unhandled_type():
+    original = {"origin": NonCopyableObject(10)}
+    copy_obj = safe_deepcopy(original)
+    assert copy_obj["origin"].value == original["origin"].value
+    assert copy_obj is not original
+    assert copy_obj["origin"] is original["origin"]
+    assert hasattr(copy_obj, "__dict__") is False  # Ensure __dict__ is not present