feat(err): add django middleware

posthog/client.py

@@ -19,6 +19,8 @@
     exc_info_from_error,
     exceptions_from_error_tuple,
     handle_in_app,
+    exception_is_already_captured,
+    mark_exception_as_captured,
 )
 from posthog.feature_flags import InconclusiveMatchError, match_feature_flag_properties
 from posthog.poller import Poller
@@ -652,9 +654,7 @@ def capture_exception(
             properties = properties or {}
 
             # Check if this exception has already been captured
-            if exception is not None and hasattr(
-                exception, "__posthog_exception_captured"
-            ):
+            if exception is not None and exception_is_already_captured(exception):
                 self.log.debug("Exception already captured, skipping")
                 return
 
@@ -708,8 +708,8 @@ def capture_exception(
             )
 
             # Mark the exception as captured to prevent duplicate captures
-            if exception is not None and isinstance(exception, BaseException):
-                setattr(exception, "__posthog_exception_captured", True)
+            if exception is not None:
+                mark_exception_as_captured(exception)
 
             return res
         except Exception as e:

posthog/exception_utils.py

@@ -779,6 +779,27 @@ def set_in_app_in_frames(frames, in_app_exclude, in_app_include, project_root=No
     return frames
 
 
+def exception_is_already_captured(error):
+    # type: (Union[BaseException, ExcInfo]) -> bool
+    if isinstance(error, BaseException):
+        return hasattr(error, "__posthog_exception_captured")
+    elif isinstance(error, tuple) and len(error) == 3:
+        return error[2] is not None and hasattr(
+            error[2], "__posthog_exception_captured"
+        )
+    else:
+        return False
+
+
+def mark_exception_as_captured(error):
+    # type: (Union[BaseException, ExcInfo]) -> None
+    if isinstance(error, BaseException):
+        setattr(error, "__posthog_exception_captured", True)
+    elif isinstance(error, tuple) and len(error) == 3:
+        if error[2] is not None:
+            setattr(error[2], "__posthog_exception_captured", True)
+
+
 def exc_info_from_error(error):
     # type: (Union[BaseException, ExcInfo]) -> ExcInfo
     if isinstance(error, tuple) and len(error) == 3:

posthog/integrations/django.py

@@ -0,0 +1,115 @@
+from typing import TYPE_CHECKING, cast
+from posthog import scopes
+
+if TYPE_CHECKING:
+    from django.http import HttpRequest, HttpResponse  # noqa: F401
+    from typing import Callable, Dict, Any, Optional  # noqa: F401
+
+
+class PosthogContextMiddleware:
+    """Middleware to automatically track Django requests.
+
+    This middleware wraps all calls with a posthog scope. It attempts to extract the following from the request headers:
+    - Session ID as $session_id, (extracted from `X-POSTHOG-SESSION-ID`)
+    - Distinct ID as $distinct_id, (extracted from `X-POSTHOG-DISTINCT-ID`)
+    - Request URL as $current_url
+    - Request Method as request_method
+
+    The scope will also auto-capture exceptions and send them to PostHog, unless you disable it by setting
+    `POSTHOG_MW_CAPTURE_EXCEPTIONS` to `False` in your Django settings.
+
+    The middleware behaviour is customisable through 3 additional functions:
+    - `POSTHOG_MW_EXTRA_TAGS`, which is a Callable[[HttpRequest], Dict[str, Any]] expected to return a dictionary of additional tags to be added to the scope.
+    - `POSTHOG_MW_REQUEST_FILTER`, which is a Callable[[HttpRequest], bool] expected to return a boolean indicating whether the request should be tracked or not.
+    - `POSTHOG_MW_TAG_MAP`, which is a Callable[[Dict[str, Any]], Dict[str, Any]], which you can use to modify the tags before they're added to the scope.
+
+    You can use the `POSTHOG_MW_TAG_MAP` function to remove any default tags you don't want to capture, or override them with your own values.
+    """
+
+    def __init__(self, get_response):
+        # type: (Callable[[HttpRequest], HttpResponse]) -> None
+        self.get_response = get_response
+
+        from django.conf import settings
+
+        if hasattr(settings, "POSTHOG_MW_EXTRA_TAGS") and callable(
+            settings.POSTHOG_MW_EXTRA_TAGS
+        ):
+            self.extra_tags = cast(
+                "Callable[[HttpRequest], Dict[str, Any]]",
+                settings.POSTHOG_MW_EXTRA_TAGS,
+            )
+        else:
+            self.extra_tags = None  # type: Optional[Callable[[HttpRequest], Dict[str, Any]]]
+
+        if hasattr(settings, "POSTHOG_MW_REQUEST_FILTER") and callable(
+            settings.POSTHOG_MW_REQUEST_FILTER
+        ):
+            self.request_filter = cast(
+                "Callable[[HttpRequest], bool]", settings.POSTHOG_MW_REQUEST_FILTER
+            )
+        else:
+            self.request_filter = None  # type: Optional[Callable[[HttpRequest], bool]]
+
+        if hasattr(settings, "POSTHOG_MW_TAG_MAP") and callable(
+            settings.POSTHOG_MW_TAG_MAP
+        ):
+            self.tag_map = cast(
+                "Callable[[Dict[str, Any]], Dict[str, Any]]",
+                settings.POSTHOG_MW_TAG_MAP,
+            )
+        else:
+            self.tag_map = None  # type: Optional[Callable[[Dict[str, Any]], Dict[str, Any]]]
+
+        if hasattr(settings, "POSTHOG_MW_CAPTURE_EXCEPTIONS") and isinstance(
+            settings.POSTHOG_MW_CAPTURE_EXCEPTIONS, bool
+        ):
+            self.capture_exceptions = settings.POSTHOG_MW_CAPTURE_EXCEPTIONS
+        else:
+            self.capture_exceptions = True
+
+    def extract_tags(self, request):
+        # type: (HttpRequest) -> Dict[str, Any]
+        tags = {}
+
+        # Extract session ID from X-POSTHOG-SESSION-ID header
+        session_id = request.META.get("HTTP_X_POSTHOG_SESSION_ID")
+        if session_id:
+            tags["$session_id"] = session_id
+
+        # Extract distinct ID from X-POSTHOG-DISTINCT-ID header
+        distinct_id = request.META.get("HTTP_X_POSTHOG_DISTINCT_ID")
+        if distinct_id:
+            tags["$distinct_id"] = distinct_id
+
+        # Extract current URL
+        absolute_url = request.build_absolute_uri()
+        if absolute_url:
+            tags["$current_url"] = absolute_url
+
+        # Extract request method
+        if request.method:
+            tags["request_method"] = request.method
+
+        # Apply extra tags if configured
+        if self.extra_tags:
+            extra = self.extra_tags(request)
+            if extra:
+                tags.update(extra)
+
+        # Apply tag mapping if configured
+        if self.tag_map:
+            tags = self.tag_map(tags)
+
+        return tags
+
+    def __call__(self, request):
+        # type: (HttpRequest) -> HttpResponse
+        if self.request_filter and not self.request_filter(request):
+            return self.get_response(request)
+
+        with scopes.new_context(self.capture_exceptions):
+            for k, v in self.extract_tags(request).items():
+                scopes.tag(k, v)
+
+            return self.get_response(request)

posthog/test/integrations/test_middleware.py

@@ -0,0 +1,206 @@
+import unittest
+from unittest.mock import Mock
+
+from posthog.integrations.django import PosthogContextMiddleware
+
+
+class MockRequest:
+    """Mock Django HttpRequest object"""
+
+    def __init__(
+        self, meta=None, method="GET", path="/test", host="example.com", is_secure=False
+    ):
+        self.META = meta or {}
+        self.method = method
+        self.path = path
+        self._host = host
+        self._is_secure = is_secure
+
+    def build_absolute_uri(self):
+        scheme = "https" if self._is_secure else "http"
+        return f"{scheme}://{self._host}{self.path}"
+
+
+class TestPosthogContextMiddleware(unittest.TestCase):
+    def create_middleware(
+        self,
+        extra_tags=None,
+        request_filter=None,
+        tag_map=None,
+        capture_exceptions=True,
+    ):
+        """Helper to create middleware instance without calling __init__"""
+        middleware = PosthogContextMiddleware.__new__(PosthogContextMiddleware)
+        middleware.get_response = Mock()
+        middleware.extra_tags = extra_tags
+        middleware.request_filter = request_filter
+        middleware.tag_map = tag_map
+        middleware.capture_exceptions = capture_exceptions
+        return middleware
+
+    def test_extract_tags_basic(self):
+        """Test basic tag extraction from request"""
+        middleware = self.create_middleware()
+        request = MockRequest(
+            meta={
+                "HTTP_X_POSTHOG_SESSION_ID": "session-123",
+                "HTTP_X_POSTHOG_DISTINCT_ID": "user-456",
+            },
+            method="POST",
+            path="/api/test",
+            host="example.com",
+            is_secure=True,
+        )
+
+        tags = middleware.extract_tags(request)
+
+        self.assertEqual(tags["$session_id"], "session-123")
+        self.assertEqual(tags["$distinct_id"], "user-456")
+        self.assertEqual(tags["$current_url"], "https://example.com/api/test")
+        self.assertEqual(tags["request_method"], "POST")
+
+    def test_extract_tags_missing_headers(self):
+        """Test tag extraction when PostHog headers are missing"""
+        middleware = self.create_middleware()
+        request = MockRequest(meta={}, method="GET", path="/home")
+
+        tags = middleware.extract_tags(request)
+
+        self.assertNotIn("$session_id", tags)
+        self.assertNotIn("$distinct_id", tags)
+        self.assertEqual(tags["$current_url"], "http://example.com/home")
+        self.assertEqual(tags["request_method"], "GET")
+
+    def test_extract_tags_partial_headers(self):
+        """Test tag extraction with only some PostHog headers present"""
+        middleware = self.create_middleware()
+        request = MockRequest(
+            meta={"HTTP_X_POSTHOG_SESSION_ID": "session-only"}, method="PUT"
+        )
+
+        tags = middleware.extract_tags(request)
+
+        self.assertEqual(tags["$session_id"], "session-only")
+        self.assertNotIn("$distinct_id", tags)
+        self.assertEqual(tags["request_method"], "PUT")
+
+    def test_extract_tags_with_extra_tags(self):
+        """Test tag extraction with extra_tags function"""
+
+        def extra_tags_func(request):
+            return {"custom_tag": "custom_value", "user_id": "789"}
+
+        middleware = self.create_middleware(extra_tags=extra_tags_func)
+        request = MockRequest(
+            meta={"HTTP_X_POSTHOG_SESSION_ID": "session-123"}, method="GET"
+        )
+
+        tags = middleware.extract_tags(request)
+
+        self.assertEqual(tags["$session_id"], "session-123")
+        self.assertEqual(tags["custom_tag"], "custom_value")
+        self.assertEqual(tags["user_id"], "789")
+
+    def test_extract_tags_with_tag_map(self):
+        """Test tag extraction with tag_map function"""
+
+        def tag_map_func(tags):
+            # Remove session_id and add a mapped version
+            if "$session_id" in tags:
+                tags["mapped_session"] = f"mapped_{tags['$session_id']}"
+                del tags["$session_id"]
+            return tags
+
+        middleware = self.create_middleware(tag_map=tag_map_func)
+        request = MockRequest(
+            meta={"HTTP_X_POSTHOG_SESSION_ID": "session-123"}, method="GET"
+        )
+
+        tags = middleware.extract_tags(request)
+
+        self.assertNotIn("$session_id", tags)
+        self.assertEqual(tags["mapped_session"], "mapped_session-123")
+
+    def test_extract_tags_with_both_extra_and_map(self):
+        """Test tag extraction with both extra_tags and tag_map"""
+
+        def extra_tags_func(request):
+            return {"extra": "value"}
+
+        def tag_map_func(tags):
+            tags["modified"] = True
+            return tags
+
+        middleware = self.create_middleware(
+            extra_tags=extra_tags_func, tag_map=tag_map_func
+        )
+        request = MockRequest(
+            meta={"HTTP_X_POSTHOG_DISTINCT_ID": "user-123"}, method="DELETE"
+        )
+
+        tags = middleware.extract_tags(request)
+
+        self.assertEqual(tags["$distinct_id"], "user-123")
+        self.assertEqual(tags["extra"], "value")
+        self.assertEqual(tags["modified"], True)
+        self.assertEqual(tags["request_method"], "DELETE")
+
+    def test_extract_tags_extra_tags_returns_none(self):
+        """Test tag extraction when extra_tags returns None"""
+
+        def extra_tags_func(request):
+            return None
+
+        middleware = self.create_middleware(extra_tags=extra_tags_func)
+        request = MockRequest(method="GET")
+
+        tags = middleware.extract_tags(request)
+
+        self.assertEqual(tags["request_method"], "GET")
+        # Should not crash when extra_tags returns None
+
+    def test_extract_tags_extra_tags_returns_empty_dict(self):
+        """Test tag extraction when extra_tags returns empty dict"""
+
+        def extra_tags_func(request):
+            return {}
+
+        middleware = self.create_middleware(extra_tags=extra_tags_func)
+        request = MockRequest(method="PATCH")
+
+        tags = middleware.extract_tags(request)
+
+        self.assertEqual(tags["request_method"], "PATCH")
+
+    def test_extract_tags_url_encoding(self):
+        """Test URL building with different scenarios"""
+        middleware = self.create_middleware()
+
+        # Test with query parameters in path
+        request = MockRequest(
+            path="/search?q=test&page=1", host="api.example.com", is_secure=True
+        )
+
+        tags = middleware.extract_tags(request)
+
+        self.assertEqual(
+            tags["$current_url"], "https://api.example.com/search?q=test&page=1"
+        )
+
+    def test_extract_tags_case_sensitivity(self):
+        """Test that header extraction is case sensitive for META keys"""
+        middleware = self.create_middleware()
+        request = MockRequest(
+            meta={
+                "HTTP_X_POSTHOG_SESSION_ID": "correct-session",
+                "http_x_posthog_session_id": "wrong-session",  # lowercase won't match
+            }
+        )
+
+        tags = middleware.extract_tags(request)
+
+        self.assertEqual(tags["$session_id"], "correct-session")
+
+
+if __name__ == "__main__":
+    unittest.main()