Added WSGIRequest to _VALID_ANY_VALUE_TYPES

Author: rads-1996

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

@@ -17,7 +17,7 @@
 from collections import OrderedDict
 from collections.abc import MutableMapping
 from typing import Mapping, Optional, Sequence, Tuple, Union
-
+from django.core.handlers.wsgi import WSGIRequest
 from opentelemetry.util import types
 
 # bytes are accepted as a user supplied value for attributes but
@@ -33,6 +33,7 @@
     str,
     Sequence,
     Mapping,
+    WSGIRequest,
 )
 
 
@@ -121,11 +122,22 @@ 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 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] = {}

opentelemetry-api/tests/attributes/test_attributes.py

@@ -181,7 +181,70 @@ def test_mapping(self):
         self.assertEqual(
             _clean_extended_attribute("headers", mapping, None), expected
         )
-
+    
+    def test_wsgi_request_attribute(self):
+        # Test WSGIRequest type validation
+        try:
+            from django.core.handlers.wsgi import WSGIRequest
+            from django.conf import settings
+            import io
+            
+            # Configure Django settings if not already configured
+            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)
+            
+            # Test that WSGIRequest gets cleaned to a dictionary format
+            expected_cleaned = {
+                'method': 'GET',
+                'path': '/test',
+                'path_info': '/test',
+                'content_type': ''
+            }
+            
+            # Test WSGIRequest cleaning directly
+            from opentelemetry.attributes import _clean_extended_attribute
+            cleaned_value = _clean_extended_attribute("request", wsgi_request, None)
+            self.assertEqual(cleaned_value, expected_cleaned)
+            
+            # Test WSGIRequest in sequences - should be cleaned to dict
+            cleaned_sequence = _clean_extended_attribute("requests", [wsgi_request], None)
+            self.assertEqual(cleaned_sequence, (expected_cleaned,))
+            
+            # Test WSGIRequest in mappings - should be cleaned to dict
+            cleaned_mapping = _clean_extended_attribute("data", {"request": wsgi_request}, None)
+            self.assertEqual(cleaned_mapping, {"request": expected_cleaned})
+            
+        except ImportError:
+            # Skip test if django is not available
+            self.skipTest("Django not available")
 
 class TestBoundedAttributes(unittest.TestCase):
     # pylint: disable=consider-using-dict-items