ref(typing): Let feature and search helpers take a service account

src/sentry/integrations/base.py

@@ -44,16 +44,14 @@
 from sentry.utils.audit import create_audit_entry
 
 if TYPE_CHECKING:
-    from django.contrib.auth.models import AnonymousUser
     from django.utils.functional import _StrPromise
 
+    from sentry.auth.principal import RequestPrincipal
     from sentry.identity.services.identity.model import RpcIdentity
     from sentry.integrations.pipeline import IntegrationPipeline  # noqa: F401
     from sentry.integrations.services.integration import RpcOrganizationIntegration
     from sentry.integrations.services.integration.model import RpcIntegration
     from sentry.models.organization import Organization
-    from sentry.users.models.user import User
-    from sentry.users.services.user import RpcUser
 
 
 logger = logging.getLogger(__name__)
@@ -623,7 +621,7 @@ def get_integration_types(provider: str) -> list[IntegrationDomain]:
 def is_provider_enabled(
     provider: IntegrationProvider,
     organization: Organization | RpcOrganization,
-    actor: User | RpcUser | AnonymousUser | None = None,
+    actor: RequestPrincipal | None = None,
 ) -> bool:
     from sentry import features
 

src/sentry/integrations/mixins/issues.py

@@ -9,7 +9,7 @@
 from typing import TYPE_CHECKING, Any, ClassVar
 
 from sentry import features
-from sentry.auth.principal import has_user
+from sentry.auth.principal import RequestPrincipal, has_user
 from sentry.integrations.base import IntegrationInstallation
 from sentry.integrations.models.external_issue import ExternalIssue
 from sentry.integrations.models.integration import Integration
@@ -217,7 +217,7 @@ def get_persisted_user_default_config_fields(self):
         """
         return []
 
-    def store_issue_last_defaults(self, project: Project, user: RpcUser | User, data):
+    def store_issue_last_defaults(self, project: Project, user: RequestPrincipal, data):
         """
         Stores the last used field defaults on a per-project basis. This
         accepts a dict of values that will be filtered to keys returned by

src/sentry/investigations/services/action_kinds/base.py

@@ -4,17 +4,17 @@
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from enum import StrEnum
-from typing import Any, Literal, NotRequired, TypedDict
+from typing import TYPE_CHECKING, Any, Literal, NotRequired, TypedDict
 
-from django.contrib.auth.models import AnonymousUser
 from rest_framework.request import Request
 
 from sentry.auth.access import Access
 from sentry.investigations.contracts import ActionPayloadSerializer
 from sentry.investigations.models import InvestigationOrchestrationRun
 from sentry.notebooks.models import Notebook, NotebookActionRequest
-from sentry.users.models.user import User
-from sentry.users.services.user.model import RpcUser
+
+if TYPE_CHECKING:
+    from sentry.auth.principal import RequestPrincipal
 
 
 class CannotApplyReason(StrEnum):
@@ -120,7 +120,7 @@ def cannot_apply_reasons(
         *,
         investigations: dict[int, Notebook],
         access: Access,
-        user: User | RpcUser | AnonymousUser,
+        user: RequestPrincipal,
     ) -> dict[int, CannotApplyReason | None]:
         """For each suggested request, why the viewer can't apply it, or None if they can."""
         return {item.id: CannotApplyReason.UNSUPPORTED_ACTION for item in action_requests}

src/sentry/investigations/services/action_kinds/connect_telemetry.py

@@ -1,9 +1,8 @@
 from __future__ import annotations
 
 from collections.abc import Sequence
-from typing import Literal, TypedDict
+from typing import TYPE_CHECKING, Literal, TypedDict
 
-from django.contrib.auth.models import AnonymousUser
 from django.db import router, transaction
 from django.db.models import QuerySet
 from rest_framework import serializers
@@ -21,8 +20,9 @@
 from sentry.models.organization import Organization
 from sentry.notebooks.models import Notebook, NotebookActionRequest, NotebookActionRequestState
 from sentry.seer.agent.monitoring_providers import org_monitoring_provider_registry
-from sentry.users.models.user import User
-from sentry.users.services.user.model import RpcUser
+
+if TYPE_CHECKING:
+    from sentry.auth.principal import RequestPrincipal
 
 CONNECT_TELEMETRY_ACTION = "connect_telemetry"
 TelemetryProvider = Literal["datadog", "gcp"]
@@ -83,7 +83,7 @@ def cannot_apply_reasons(
         *,
         investigations: dict[int, Notebook],
         access: Access,
-        user: User | RpcUser | AnonymousUser,
+        user: RequestPrincipal,
     ) -> dict[int, CannotApplyReason | None]:
         can_install = access.has_scope(_INSTALL_SCOPE)
         reasons: dict[int, CannotApplyReason | None] = {}

src/sentry/investigations/services/action_kinds/update_detector.py

@@ -3,9 +3,8 @@
 from collections import defaultdict
 from collections.abc import Collection, Mapping, Sequence
 from dataclasses import dataclass
-from typing import Any, Literal, TypedDict, cast
+from typing import TYPE_CHECKING, Any, Literal, TypedDict, cast
 
-from django.contrib.auth.models import AnonymousUser
 from django.db import router, transaction
 from rest_framework import serializers
 
@@ -42,8 +41,6 @@
 from sentry.notebooks.contracts import StrictContractSerializer
 from sentry.notebooks.models import Notebook, NotebookActionRequest
 from sentry.snuba.models import SnubaQuery
-from sentry.users.models.user import User
-from sentry.users.services.user.model import RpcUser
 from sentry.workflow_engine.endpoints.organization_detector_details import (
     check_metric_detector_allowed,
 )
@@ -56,6 +53,9 @@
 from sentry.workflow_engine.models.detector import DetectorQuerySet
 from sentry.workflow_engine.types import DetectorPriorityLevel
 
+if TYPE_CHECKING:
+    from sentry.auth.principal import RequestPrincipal
+
 UPDATE_DETECTOR_ACTION = "update_detector"
 
 
@@ -404,7 +404,7 @@ def cannot_apply_reasons(
         *,
         investigations: dict[int, Notebook],
         access: Access,
-        user: User | RpcUser | AnonymousUser,
+        user: RequestPrincipal,
     ) -> dict[int, CannotApplyReason | None]:
         source_detector_ids = {
             investigation_id: source_detector_id(source)

src/sentry/preprod/quotas.py

@@ -4,10 +4,9 @@
 from collections.abc import Callable
 from datetime import datetime, timedelta
 from enum import Enum
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import sentry_sdk
-from django.contrib.auth.models import AnonymousUser
 from django.utils import timezone
 
 from sentry import features, quotas
@@ -16,7 +15,9 @@
 from sentry.models.organization import Organization
 from sentry.preprod.artifact_search import artifact_matches_query
 from sentry.preprod.models import PreprodArtifact
-from sentry.users.models.user import User
+
+if TYPE_CHECKING:
+    from sentry.auth.principal import RequestPrincipal
 
 logger = logging.getLogger(__name__)
 
@@ -45,7 +46,7 @@ def get_size_retention_cutoff(organization: Organization) -> datetime:
 DISTRIBUTION_ENABLED_QUERY_KEY = "sentry:preprod_distribution_enabled_query"
 
 
-def has_size_quota(organization: Organization, actor: User | AnonymousUser | None = None) -> bool:
+def has_size_quota(organization: Organization, actor: RequestPrincipal | None = None) -> bool:
     result = quotas.backend.has_usage_quota(organization.id, DataCategory.SIZE_ANALYSIS)
     logger.info(
         "has_size_quota",
@@ -55,7 +56,7 @@ def has_size_quota(organization: Organization, actor: User | AnonymousUser | Non
 
 
 def has_installable_quota(
-    organization: Organization, actor: User | AnonymousUser | None = None
+    organization: Organization, actor: RequestPrincipal | None = None
 ) -> bool:
     if not features.has(
         "organizations:preprod-enforce-distribution-quota", organization, actor=actor

src/sentry/search/utils.py

@@ -4,7 +4,7 @@
 from collections.abc import Callable, Sequence
 from datetime import datetime, timedelta, timezone
 from enum import Enum
-from typing import Any, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 from django.contrib.auth.models import AnonymousUser
 from django.db import DataError, connections, router
@@ -33,6 +33,9 @@
 from sentry.utils.validators import is_event_id_or_list as _is_event_id_or_list
 from sentry.utils.validators import is_span_id_or_list as _is_span_id_or_list
 
+if TYPE_CHECKING:
+    from sentry.auth.principal import RequestPrincipal
+
 
 class InvalidQuery(Exception):
     pass
@@ -352,7 +355,7 @@ def parse_actor_or_none_value(
 _HACKY_INVALID_USER = RpcUser(id=0)
 
 
-def parse_user_value(value: str, user: User | RpcUser | AnonymousUser) -> RpcUser:
+def parse_user_value(value: str, user: RequestPrincipal) -> RpcUser:
     if value == "me":
         if isinstance(user, User):
             return serialize_rpc_user(user)

tests/sentry/investigations/test_contracts.py

@@ -6,9 +6,9 @@
 
 import orjson
 import pytest
-from django.contrib.auth.models import AnonymousUser
 
 from sentry.auth.access import Access
+from sentry.auth.principal import RequestPrincipal
 from sentry.db.models.fields.bounded import I32_MAX, I64_MAX
 from sentry.investigations.contracts import (
     MAX_PROJECTION_BYTES,
@@ -32,8 +32,6 @@
     CannotApplyReason,
 )
 from sentry.notebooks.models import Notebook, NotebookActionRequest
-from sentry.users.models.user import User
-from sentry.users.services.user.model import RpcUser
 from sentry.utils import json
 from tests.sentry.notebooks.test_contracts import golden_payload
 
@@ -536,7 +534,7 @@ def __init__(self) -> None:
         # The requests after_upsert ran for, in order.
         self.upserted: list[NotebookActionRequest] = []
         # The arguments from each cannot_apply_reasons call.
-        self.checked: list[tuple[list[int], list[int], Access, User | RpcUser | AnonymousUser]] = []
+        self.checked: list[tuple[list[int], list[int], Access, RequestPrincipal]] = []
         # Which of check_remote, check and apply ran, in order.
         self.calls: list[str] = []
         # What check and apply raise, if anything.
@@ -557,7 +555,7 @@ def cannot_apply_reasons(
         *,
         investigations: dict[int, Notebook],
         access: Access,
-        user: User | RpcUser | AnonymousUser,
+        user: RequestPrincipal,
     ) -> dict[int, CannotApplyReason | None]:
         self.checked.append(
             ([item.id for item in action_requests], sorted(investigations), access, user)