feat: add deepcopy implementation for BoundedAttributes

CHANGELOG.md

@@ -26,6 +26,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4973](https://github.com/open-telemetry/opentelemetry-python/pull/4973))
 - `opentelemetry-exporter-prometheus`: Fix metric name prefix
   ([#4895](https://github.com/open-telemetry/opentelemetry-python/pull/4895))
+- `opentelemetry-api`, `opentelemetry-sdk`: Add deepcopy support for `BoundedAttributes` and `BoundedList`
+  ([#4934](https://github.com/open-telemetry/opentelemetry-python/pull/4934))
 
 ## Version 1.40.0/0.61b0 (2026-03-04)
 

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,20 @@ def __iter__(self):  # type: ignore
     def __len__(self) -> int:
         return len(self._dict)
 
+    def __deepcopy__(self, memo: dict) -> "BoundedAttributes":
+        copy_ = BoundedAttributes(
+            maxlen=self.maxlen,
+            immutable=self._immutable,
+            max_value_len=self.max_value_len,
+            extended_attributes=self._extended_attributes,
+        )
+        memo[id(self)] = copy_
+        with self._lock:
+            # Assign _dict directly to avoid re-cleaning already clean values
+            # and to bypass the immutability guard in __setitem__
+            copy_._dict = copy.deepcopy(self._dict, memo)
+            copy_.dropped = self.dropped
+        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,30 @@ def __str__(self):
         self.assertEqual(
             "<DummyWSGIRequest method=GET path=/example/>", cleaned_value
         )
+
+    def test_deepcopy(self):
+        bdict = BoundedAttributes(4, self.base, immutable=False)
+        bdict.dropped = 10
+        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/src/opentelemetry/sdk/util/__init__.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import copy
 import datetime
 import threading
 from collections import deque
@@ -55,6 +56,14 @@ def __init__(self, maxlen: Optional[int]):
         self._dq = deque(maxlen=maxlen)  # type: deque
         self._lock = threading.Lock()
 
+    def __deepcopy__(self, memo):
+        copy_ = BoundedList(0)
+        memo[id(self)] = copy_
+        with self._lock:
+            copy_.dropped = self.dropped
+            copy_._dq = copy.deepcopy(self._dq, memo)
+        return copy_
+
     def __repr__(self):
         return f"{type(self).__name__}({list(self._dq)}, maxlen={self._dq.maxlen})"
 

opentelemetry-sdk/src/opentelemetry/sdk/util/__init__.pyi

@@ -13,6 +13,7 @@
 # limitations under the License.
 
 from typing import (
+    Any,
     Iterable,
     Iterator,
     Mapping,
@@ -44,6 +45,7 @@ class BoundedList(Sequence[_T]):
 
     dropped: int
     def __init__(self, maxlen: Optional[int]): ...
+    def __deepcopy__(self, memo: dict[int, Any]) -> BoundedList[_T]: ...
     def insert(self, index: int, value: _T) -> None: ...
     @overload
     def __getitem__(self, i: int) -> _T: ...

opentelemetry-sdk/tests/test_util.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import copy
 import unittest
 
 from opentelemetry.sdk.util import BoundedList
@@ -142,3 +143,21 @@ def test_no_limit(self):
 
         for num in range(100):
             self.assertEqual(blist[num], num)
+
+    # pylint: disable=protected-access
+    def test_deepcopy(self):
+        blist = BoundedList(maxlen=10)
+        blist.append(1)
+        blist.append([2, 3])
+        blist.dropped = 5
+
+        blist_copy = copy.deepcopy(blist)
+
+        self.assertIsNot(blist, blist_copy)
+        self.assertIsNot(blist._dq, blist_copy._dq)
+        self.assertIsNot(blist._lock, blist_copy._lock)
+        self.assertEqual(list(blist), list(blist_copy))
+        self.assertEqual(blist.dropped, blist_copy.dropped)
+        self.assertEqual(blist._dq.maxlen, blist_copy._dq.maxlen)
+        self.assertIsNot(blist[1], blist_copy[1])
+        self.assertEqual(blist[1], blist_copy[1])

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
@@ -58,7 +59,7 @@
     ParentBased,
     StaticSampler,
 )
-from opentelemetry.sdk.util import BoundedDict, ns_to_iso_str
+from opentelemetry.sdk.util import BoundedDict, BoundedList, ns_to_iso_str
 from opentelemetry.sdk.util.instrumentation import InstrumentationInfo
 from opentelemetry.test.spantestutil import (
     get_span_with_dropped_attributes_events_links,
@@ -708,6 +709,69 @@ 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 = BoundedList(10)
+        events.extend(
+            (
+                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))
+        self.assertIsNot(span_copy.events, span.events)
+        self.assertEqual(span_copy.events[0].name, span.events[0].name)
+        self.assertEqual(
+            span_copy.events[0].attributes, span.events[0].attributes
+        )
+
+        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