feat: add deepcopy implementation for BoundedAttributes

Author: herin049

opentelemetry-api/src/opentelemetry/attributes/__init__.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import copy
 import logging
 import threading
 from collections import OrderedDict
@@ -318,5 +319,18 @@ def __iter__(self):  # type: ignore
     def __len__(self) -> int:
         return len(self._dict)
 
+    def __deepcopy__(self, memo: dict) -> "BoundedAttributes":
+        with self._lock:
+            attributes = copy.deepcopy(self._dict, memo)
+        copy_ = BoundedAttributes(
+            self.maxlen,
+            attributes,
+            self._immutable,
+            self.max_value_len,
+            self._extended_attributes,
+        )
+        memo[id(self)] = copy_
+        return copy_
+
     def copy(self):  # type: ignore
         return self._dict.copy()  # type: ignore

opentelemetry-api/tests/attributes/test_attributes.py

@@ -14,6 +14,7 @@
 
 # type: ignore
 
+import copy
 import unittest
 from typing import MutableSequence
 
@@ -320,3 +321,29 @@ def __str__(self):
         self.assertEqual(
             "<DummyWSGIRequest method=GET path=/example/>", cleaned_value
         )
+
+    def test_deepcopy(self):
+        bdict = BoundedAttributes(4, self.base, immutable=False)
+        bdict_copy = copy.deepcopy(bdict)
+
+        for key in bdict_copy:
+            self.assertEqual(bdict_copy[key], bdict[key])
+
+        self.assertEqual(bdict_copy.dropped, bdict.dropped)
+        self.assertEqual(bdict_copy.maxlen, bdict.maxlen)
+        self.assertEqual(bdict_copy.max_value_len, bdict.max_value_len)
+
+        bdict_copy["name"] = "Bob"
+        self.assertNotEqual(bdict_copy["name"], bdict["name"])
+
+        bdict["age"] = 99
+        self.assertNotEqual(bdict["age"], bdict_copy["age"])
+
+    def test_deepcopy_preserves_immutability(self):
+        bdict = BoundedAttributes(
+            maxlen=4, attributes=self.base, immutable=True
+        )
+        bdict_copy = copy.deepcopy(bdict)
+
+        with self.assertRaises(TypeError):
+            bdict_copy["invalid"] = "invalid"

opentelemetry-sdk/tests/trace/test_trace.py

@@ -15,6 +15,7 @@
 # pylint: disable=too-many-lines
 # pylint: disable=no-member
 
+import copy
 import shutil
 import subprocess
 import unittest
@@ -708,6 +709,63 @@ def test_link_dropped_attributes(self):
         )
         self.assertEqual(link2.dropped_attributes, 0)
 
+    def test_deepcopy(self):
+        context = trace_api.SpanContext(
+            trace_id=0x000000000000000000000000DEADBEEF,
+            span_id=0x00000000DEADBEF0,
+            is_remote=False,
+        )
+        attributes = BoundedAttributes(
+            10, {"key1": "value1", "key2": 42}, immutable=False
+        )
+        events = [
+            trace.Event("event1", {"ekey": "evalue"}),
+            trace.Event("event2", {"ekey2": "evalue2"}),
+        ]
+        links = [
+            trace_api.Link(
+                context=trace_api.INVALID_SPAN_CONTEXT,
+                attributes={"lkey": "lvalue"},
+            )
+        ]
+
+        span = trace.ReadableSpan(
+            name="test-span",
+            context=context,
+            attributes=attributes,
+            events=events,
+            links=links,
+            status=Status(StatusCode.OK),
+        )
+
+        span_copy = copy.deepcopy(span)
+
+        self.assertEqual(span_copy.name, span.name)
+        self.assertEqual(span_copy.status.status_code, span.status.status_code)
+        self.assertEqual(span_copy.context.trace_id, span.context.trace_id)
+        self.assertEqual(span_copy.context.span_id, span.context.span_id)
+
+        self.assertEqual(dict(span_copy.attributes), dict(span.attributes))
+        attributes["key1"] = "mutated"
+        self.assertNotEqual(
+            span_copy.attributes["key1"], span.attributes["key1"]
+        )
+
+        self.assertEqual(len(span_copy.events), len(span.events))
+        events[0] = trace.Event("mutated-event", {"mutated": "value"})
+        self.assertNotEqual(span_copy.events[0].name, events[0].name)
+        self.assertEqual(span_copy.events[0].name, "event1")
+
+        self.assertEqual(len(span_copy.links), len(span.links))
+        self.assertEqual(
+            span_copy.links[0].attributes, span.links[0].attributes
+        )
+        links[0] = trace_api.Link(
+            context=trace_api.INVALID_SPAN_CONTEXT,
+            attributes={"mutated": "link"},
+        )
+        self.assertNotIn("mutated", span_copy.links[0].attributes)
+
 
 class DummyError(Exception):
     pass