Finished draft with mixin

src/sentry/api/analytics.py

@@ -1,11 +1,4 @@
-import logging
-
-from rest_framework.request import Request
-
 from sentry import analytics
-from sentry.utils.http import get_api_relative_path, origin_from_request, query_string
-
-logger = logging.getLogger(__name__)
 
 
 class OrganizationSavedSearchCreatedEvent(analytics.Event):
@@ -47,41 +40,15 @@ class DevToolbarRequestEvent(analytics.Event):
         analytics.Attribute("path"),  # path to endpoint
         analytics.Attribute("query"),  # string or dict?
         analytics.Attribute("origin"),
-        analytics.Attribute("organization_slug"),
-        analytics.Attribute("project_slug"),
+        analytics.Attribute("organization_id", required=False),
+        analytics.Attribute("organization_slug", required=False),
+        analytics.Attribute("project_id", required=False),
+        analytics.Attribute("project_slug", required=False),
+        analytics.Attribute("issue_id", required=False),
         analytics.Attribute("user_id"),  # needed to aggregate/send to amplitude(?)
     )
 
 
-def track_devtoolbar_api_analytics(
-    request: Request,
-    scope: str | None,
-    org_slug: str | None = None,
-    project_slug: str | None = None,
-):
-    """
-    @param scope - "organization", "project", "group", or None.
-    """
-    try:
-        origin = origin_from_request(request)
-        query = query_string(request)  # starts with ?
-        path = get_api_relative_path(request, scope)
-        analytics.record(
-            "devtoolbar.request",
-            path=path,
-            query=query,
-            origin=origin,
-            organization_slug=org_slug,
-            project_slug=project_slug,
-            user_id=str(request.user.id) if request.user else None,
-        )
-    except Exception:
-        logger.exception(
-            "devtoolbar: failed to record api analytics event.",
-            extra={"org_slug": org_slug, "project_slug": project_slug},
-        )
-
-
 analytics.register(OrganizationSavedSearchCreatedEvent)
 analytics.register(OrganizationSavedSearchDeletedEvent)
 analytics.register(GroupSimilarIssuesEmbeddingsCountEvent)

src/sentry/api/base.py

@@ -43,9 +43,12 @@
 from sentry.utils.dates import to_datetime
 from sentry.utils.http import (
     absolute_uri,
+    get_api_path_from_request,
     is_using_customer_domain,
     is_valid_origin,
     origin_from_request,
+    parse_id_or_slug_param,
+    query_string,
 )
 from sentry.utils.sdk import capture_exception, merge_context_into_scope
 
@@ -707,6 +710,55 @@ def track_set_commits_local(self, request: Request, organization_id=None, projec
         )
 
 
+class DevtoolbarAnalyticsMixin:
+    # Must come before the base Endpoint in the inheritance chain. Ex: MyEndpoint(DevtoolbarAnalyticsMixin, Endpoint)
+    def dispatch(self, request, *args, **kwargs):
+        org_id, org_slug, project_id, project_slug = None, None, None, None
+        try:
+            if request.headers.get("queryReferrer") == "devtoolbar":
+                org_id_or_slug = kwargs.get(
+                    "organization_id_or_slug", kwargs.get("organization_slug")
+                )
+                org_id, org_slug = parse_id_or_slug_param(org_id_or_slug)
+
+                project_id_or_slug = kwargs.get("project_id_or_slug")
+                project_id, project_slug = parse_id_or_slug_param(project_id_or_slug)
+
+                issue_id = kwargs.get("issue_id")
+                scope = (
+                    "group"
+                    if issue_id
+                    else "project"
+                    if project_id_or_slug
+                    else "organization"
+                    if org_id_or_slug
+                    else None
+                )
+
+                origin = origin_from_request(request)
+                query = query_string(request)  # starts with '?'
+                path = get_api_path_from_request(request, scope)
+                analytics.record(
+                    "devtoolbar.request",
+                    path=path,
+                    query=query,
+                    origin=origin,
+                    organization_id=str(org_id) if org_id else None,
+                    organization_slug=org_slug,
+                    project_id=str(project_id) if project_id else None,
+                    project_slug=project_slug,
+                    issue_id=issue_id,
+                    user_id=str(request.user.id) if request.user else None,
+                )
+        except Exception:
+            logger.exception(
+                "devtoolbar: failed to record api analytics event.",
+                extra={"org_slug": org_slug, "project_slug": project_slug},
+            )
+
+        super().dispatch(*args, **kwargs)  # type: ignore[attr-defined]
+
+
 class EndpointSiloLimit(SiloLimit):
     def modify_endpoint_class(self, decorated_class: type[Endpoint]) -> type:
         dispatch_override = self.create_override(decorated_class.dispatch)

src/sentry/api/bases/group.py

@@ -5,7 +5,7 @@
 from rest_framework.request import Request
 
 from sentry.api.api_owners import ApiOwner
-from sentry.api.base import Endpoint
+from sentry.api.base import DevtoolbarAnalyticsMixin, Endpoint
 from sentry.api.bases.project import ProjectPermission
 from sentry.api.exceptions import ResourceDoesNotExist
 from sentry.integrations.tasks import create_comment, update_comment
@@ -35,7 +35,7 @@ def has_object_permission(self, request: Request, view, group):
         return super().has_object_permission(request, view, group.project)
 
 
-class GroupEndpoint(Endpoint):
+class GroupEndpoint(DevtoolbarAnalyticsMixin, Endpoint):
     owner = ApiOwner.ISSUES
     permission_classes = (GroupPermission,)
 

src/sentry/api/bases/organization.py

@@ -11,7 +11,7 @@
 from rest_framework.permissions import BasePermission
 from rest_framework.request import Request
 
-from sentry.api.base import Endpoint
+from sentry.api.base import DevtoolbarAnalyticsMixin, Endpoint
 from sentry.api.exceptions import ResourceDoesNotExist
 from sentry.api.helpers.environments import get_environments
 from sentry.api.permissions import SentryPermission, StaffPermissionMixin
@@ -228,7 +228,7 @@ class OrganizationMetricsPermission(OrganizationPermission):
     }
 
 
-class ControlSiloOrganizationEndpoint(Endpoint):
+class ControlSiloOrganizationEndpoint(DevtoolbarAnalyticsMixin, Endpoint):
     """
     A base class for endpoints that use an organization scoping but lives in the control silo
     """
@@ -319,7 +319,7 @@ def _validate_fetched_projects(
         raise PermissionDenied
 
 
-class OrganizationEndpoint(Endpoint):
+class OrganizationEndpoint(DevtoolbarAnalyticsMixin, Endpoint):
     permission_classes: tuple[type[BasePermission], ...] = (OrganizationPermission,)
 
     def get_projects(

src/sentry/api/bases/project.py

@@ -7,7 +7,7 @@
 from rest_framework.request import Request
 from rest_framework.response import Response
 
-from sentry.api.base import Endpoint
+from sentry.api.base import DevtoolbarAnalyticsMixin, Endpoint
 from sentry.api.exceptions import ProjectMoved, ResourceDoesNotExist
 from sentry.api.helpers.environments import get_environments
 from sentry.api.permissions import StaffPermissionMixin
@@ -113,7 +113,7 @@ class ProjectMetricsExtractionRulesPermission(ProjectPermission):
     }
 
 
-class ProjectEndpoint(Endpoint):
+class ProjectEndpoint(DevtoolbarAnalyticsMixin, Endpoint):
     permission_classes: tuple[type[BasePermission], ...] = (ProjectPermission,)
 
     def convert_args(

src/sentry/utils/http.py

@@ -228,21 +228,30 @@ def is_using_customer_domain(request: HttpRequest) -> TypeGuard[_HttpRequestWith
     return bool(hasattr(request, "subdomain") and request.subdomain)
 
 
-def get_api_path_from_request(request: HttpRequest, scope: str | None) -> str:
+def get_api_path_from_request(request: HttpRequest, endpoint_scope: str | None) -> str:
     """
     Returns request.path without the specific org, project, or group identifiers, depending on `scope`.
 
-    @param request - the request object.
-    @param scope - "organization", "project", "group", or None.
+    @param request        - the request object.
+    @param endpoint_scope - "organization", "project", "group", or None.
     """
     segments = request.path.split("/")
-    if scope is not None:
+    if endpoint_scope is not None:
         segments[2] = "<organization_id_or_slug>"
 
-    if scope == "project":
+    if endpoint_scope == "project":
         segments[3] = "<project_id_or_slug>"
-    elif scope == "group":
+    elif endpoint_scope == "group":
         # segments[3] is either "issues" or "groups"
         segments[4] = "<group_id>"
 
     return "/" + "/".join(segments) + "/"
+
+
+def parse_id_or_slug_param(id_or_slug: str | None) -> tuple[int | None, str | None]:
+    if not id_or_slug:
+        return None, None
+
+    if id_or_slug.isnumeric():
+        return int(id_or_slug), None
+    return None, id_or_slug