Add WSGIRequest support to OpenTelemetry attributes

Author: rads-1996

CHANGELOG.md

@@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4695](https://github.com/open-telemetry/opentelemetry-python/pull/4695)).
 - docs: linked the examples with their github source code location and added Prometheus example
   ([#4728](https://github.com/open-telemetry/opentelemetry-python/pull/4728))
+- opentelemetry-api: Fix for issue #4732 - Invalid type WSGIRequest for attribute 'request' value opentelemetry
+  ([#4733](https://github.com/open-telemetry/opentelemetry-python/pull/4733))
 
 ## Version 1.36.0/0.57b0 (2025-07-29)
 

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

@@ -20,10 +20,23 @@
 
 from opentelemetry.util import types
 
+
+def _get_wsgi_request_type():
+    """Get WSGIRequest type if Django is available, otherwise return None."""
+    try:
+        # pylint: disable=import-outside-toplevel
+        from django.core.handlers.wsgi import WSGIRequest  # type: ignore
+
+        return WSGIRequest
+    except ImportError:
+        return None
+
+
 # bytes are accepted as a user supplied value for attributes but
 # decoded to strings internally.
 _VALID_ATTR_VALUE_TYPES = (bool, str, bytes, int, float)
 # AnyValue possible values
+_WSGIRequest = _get_wsgi_request_type()
 _VALID_ANY_VALUE_TYPES = (
     type(None),
     bool,
@@ -33,7 +46,7 @@
     str,
     Sequence,
     Mapping,
-)
+) + ((_WSGIRequest,) if _WSGIRequest is not None else ())
 
 
 _logger = logging.getLogger(__name__)
@@ -121,12 +134,25 @@ def _clean_attribute(
 def _clean_extended_attribute_value(
     value: types.AnyValue, max_len: Optional[int]
 ) -> types.AnyValue:
+    # pylint: disable=too-many-branches
     # for primitive types just return the value and eventually shorten the string length
     if value is None or isinstance(value, _VALID_ATTR_VALUE_TYPES):
         if max_len is not None and isinstance(value, str):
             value = value[:max_len]
         return value
 
+    if _WSGIRequest is not None and isinstance(value, _WSGIRequest):
+        wsgi_data = {
+            "method": getattr(value, "method", None),
+            "path": getattr(value, "path", None),
+            "path_info": getattr(value, "path_info", None),
+            "content_type": getattr(value, "content_type", None),
+            "user": str(getattr(value, "user", None))
+            if hasattr(value, "user")
+            else None,
+        }
+        return {k: v for k, v in wsgi_data.items() if v is not None}
+
     if isinstance(value, Mapping):
         cleaned_dict: dict[str, types.AnyValue] = {}
         for key, element in value.items():

opentelemetry-api/tests/attributes/test_attributes.py

@@ -14,16 +14,35 @@
 
 # type: ignore
 
+import io
 import unittest
 from typing import MutableSequence
 
 from opentelemetry.attributes import (
     BoundedAttributes,
     _clean_attribute,
     _clean_extended_attribute,
+    _get_wsgi_request_type,
 )
 
 
+def _get_django_settings():
+    """Get Django settings if available, otherwise return None."""
+    try:
+        # pylint: disable=import-outside-toplevel
+        from django.conf import (
+            settings,  # pyright: ignore[reportMissingImports]
+        )
+
+        return settings
+    except ImportError:
+        return None
+
+
+_WSGIRequest = _get_wsgi_request_type()
+_settings = _get_django_settings()
+
+
 class TestAttributes(unittest.TestCase):
     # pylint: disable=invalid-name
     def assertValid(self, value, key="k"):
@@ -182,6 +201,62 @@ def test_mapping(self):
             _clean_extended_attribute("headers", mapping, None), expected
         )
 
+    def test_wsgi_request_attribute(self):
+        if _WSGIRequest is None or _settings is None:
+            self.skipTest("Django not available")
+
+        if not _settings.configured:
+            _settings.configure(
+                DEBUG=True,
+                SECRET_KEY="test-secret-key",
+                USE_TZ=True,
+                ROOT_URLCONF=[],
+                MIDDLEWARE=[],
+            )
+
+            # Create a minimal WSGI environ dict
+            environ = {
+                "REQUEST_METHOD": "GET",
+                "PATH_INFO": "/test",
+                "QUERY_STRING": "",
+                "CONTENT_TYPE": "",
+                "CONTENT_LENGTH": "",
+                "HTTP_HOST": "testserver",
+                "wsgi.version": (1, 0),
+                "wsgi.url_scheme": "http",
+                "wsgi.input": io.StringIO(),
+                "wsgi.errors": io.StringIO(),
+                "wsgi.multithread": False,
+                "wsgi.multiprocess": False,
+                "wsgi.run_once": False,
+                "SERVER_NAME": "testserver",
+                "SERVER_PORT": "80",
+            }
+
+            # Create a WSGIRequest object
+            wsgi_request = _WSGIRequest(environ)
+            expected_cleaned = {
+                "method": "GET",
+                "path": "/test",
+                "path_info": "/test",
+                "content_type": "",
+            }
+
+            cleaned_value = _clean_extended_attribute(
+                "request", wsgi_request, None
+            )
+            self.assertEqual(cleaned_value, expected_cleaned)
+
+            cleaned_sequence = _clean_extended_attribute(
+                "requests", [wsgi_request], None
+            )
+            self.assertEqual(cleaned_sequence, (expected_cleaned,))
+
+            cleaned_mapping = _clean_extended_attribute(
+                "data", {"request": wsgi_request}, None
+            )
+            self.assertEqual(cleaned_mapping, {"request": expected_cleaned})
+
 
 class TestBoundedAttributes(unittest.TestCase):
     # pylint: disable=consider-using-dict-items