ref(typing): Let shared helpers take a service account

src/sentry/api/serializers/models/activity.py

@@ -2,10 +2,9 @@
 from datetime import datetime
 from typing import Any, TypeAlias, TypedDict
 
-from django.contrib.auth.models import AnonymousUser
-
 from sentry.api.serializers import Serializer, register, serialize
 from sentry.api.serializers.models.commit import CommitWithReleaseSerializer
+from sentry.auth.principal import RequestPrincipal
 from sentry.issues.action_log.types import ActionSource, GroupActionType
 from sentry.issues.models.groupactionlogentry import GroupActionLogEntry
 from sentry.models.activity import Activity
@@ -19,8 +18,6 @@
 from sentry.sentry_apps.services.app import app_service
 from sentry.sentry_apps.services.app.model import RpcSentryApp
 from sentry.types.activity import ActivityType
-from sentry.users.models.user import User
-from sentry.users.services.user.model import RpcUser
 from sentry.users.services.user.serial import serialize_generic_user
 from sentry.users.services.user.service import user_service
 from sentry.utils.action_log.activity_translator import activity_action_idempotency_key
@@ -78,7 +75,7 @@ def serialize_activity_source(source: str | None) -> str | None:
 def serialize_group_notes(
     group: Group,
     notes: Sequence[Activity],
-    user: User | RpcUser | AnonymousUser | None,
+    user: RequestPrincipal | None,
 ) -> list[GroupNoteSerializerResponse]:
     if not notes:
         return []

src/sentry/api/serializers/models/groupactionlogentry.py

@@ -12,6 +12,7 @@
     serialize_group_notes,
 )
 from sentry.api.serializers.models.commit import CommitWithReleaseSerializer
+from sentry.auth.principal import RequestPrincipal
 from sentry.issues.action_log.read_metrics import (
     ActivityReadFallbackReason,
     ActivityReadResult,
@@ -119,7 +120,7 @@ def _serialized_comment_id(obj: GroupActionLogEntry) -> str | None:
 
 def get_serialized_activity_items(
     group: "Group",
-    user: User | RpcUser | AnonymousUser | None,
+    user: RequestPrincipal | None,
     *,
     endpoint: str,
     limit: int = 99,

src/sentry/auth/staff.py

@@ -5,10 +5,9 @@
 import logging
 from collections.abc import Iterable
 from datetime import datetime, timedelta, timezone
-from typing import Any, Final
+from typing import TYPE_CHECKING, Any, Final
 
 from django.conf import settings
-from django.contrib.auth.models import AnonymousUser
 from django.core.signing import BadSignature
 from django.http import HttpRequest, HttpResponse
 from django.utils import timezone as django_timezone
@@ -19,6 +18,9 @@
 from sentry.auth.elevated_mode import ElevatedMode, InactiveReason
 from sentry.auth.system import is_system_auth
 from sentry.users.models.user import User
+
+if TYPE_CHECKING:
+    from sentry.auth.principal import RequestPrincipal
 from sentry.utils.auth import has_completed_sso
 
 logger = logging.getLogger("sentry.staff")
@@ -56,7 +58,7 @@ def is_active_staff(request: HttpRequest | Request) -> bool:
 
 
 # TODO(schew2381): Delete after staff is GA'd and the options are removed
-def has_staff_option(user: User | AnonymousUser) -> bool:
+def has_staff_option(user: RequestPrincipal) -> bool:
     """
     This checks two options, the first being whether or not staff has been GA'd.
     If not, it falls back to checking the second option which by email specifies which

src/sentry/demo_mode/utils.py

@@ -1,10 +1,14 @@
-from django.contrib.auth.models import AnonymousUser
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 
 from sentry import options
 from sentry.models.organization import Organization
 from sentry.organizations.services.organization import RpcOrganization
 from sentry.users.models.user import User
-from sentry.users.services.user.model import RpcUser
+
+if TYPE_CHECKING:
+    from sentry.auth.principal import RequestPrincipal
 
 READONLY_SCOPES = frozenset(
     [
@@ -23,11 +27,14 @@ def is_demo_mode_enabled() -> bool:
     return options.get("demo-mode.enabled")
 
 
-def is_demo_user(user: User | AnonymousUser | None | RpcUser) -> bool:
+def is_demo_user(user: RequestPrincipal | None) -> bool:
+    # circular: sentry.auth.principal imports this module
+    from sentry.auth.principal import acting_user_id
+
     if not user:
         return False
 
-    return user.id in options.get("demo-mode.users")
+    return acting_user_id(user) in options.get("demo-mode.users")
 
 
 def is_demo_org(organization: Organization | RpcOrganization | None) -> bool:

src/sentry/issues/derived/gate.py

@@ -1,4 +1,6 @@
-from django.contrib.auth.models import AnonymousUser
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
 
 from sentry import features
 from sentry.issues.action_log.read_metrics import (
@@ -8,8 +10,9 @@
 )
 from sentry.models.options.project_option import ProjectOption
 from sentry.models.project import Project
-from sentry.users.models.user import User
-from sentry.users.services.user.model import RpcUser
+
+if TYPE_CHECKING:
+    from sentry.auth.principal import RequestPrincipal
 
 # ProjectOption key whose value tracks group action-log backfill state. True means complete,
 # false means pending, and a missing option uses the project's epoch default. Projects created
@@ -34,7 +37,7 @@ def derived_should_be_correct(project: Project) -> bool:
 
 def should_serve_action_log_activity(
     project: Project,
-    actor: User | RpcUser | AnonymousUser | None = None,
+    actor: RequestPrincipal | None = None,
     *,
     endpoint: str,
 ) -> bool:

src/sentry/models/team.py

@@ -4,7 +4,6 @@
 from typing import TYPE_CHECKING, ClassVar
 
 from django.conf import settings
-from django.contrib.auth.models import AnonymousUser
 from django.db import models, router, transaction
 from django.utils import timezone
 from django.utils.translation import gettext_lazy as _
@@ -25,16 +24,15 @@
 from sentry.utils.snowflake import save_with_snowflake_id, snowflake_id_model
 
 if TYPE_CHECKING:
+    from sentry.auth.principal import RequestPrincipal
     from sentry.models.organization import Organization
-    from sentry.users.models.user import User
-    from sentry.users.services.user import RpcUser
 
 
 class TeamManager(BaseManager["Team"]):
     def get_for_user(
         self,
         organization: Organization,
-        user: User | RpcUser | AnonymousUser,
+        user: RequestPrincipal,
         scope: str | None = None,
         is_team_admin: bool = False,
     ) -> Sequence[Team]:

src/sentry/notebooks/access.py

@@ -1,19 +1,20 @@
 from __future__ import annotations
 
-from django.contrib.auth.models import AnonymousUser
+from typing import TYPE_CHECKING
 
 from sentry import features
 from sentry.models.organization import Organization
-from sentry.users.models.user import User
-from sentry.users.services.user.model import RpcUser
+
+if TYPE_CHECKING:
+    from sentry.auth.principal import RequestPrincipal
 
 INVESTIGATIONS_FEATURE = "organizations:investigations"
 # Any one of these lets a member read and manage investigations.
 INVESTIGATION_SCOPES = frozenset({"org:read", "org:write", "org:admin"})
 
 
 def has_investigations_access(
-    organization: Organization, actor: User | RpcUser | AnonymousUser | None = None
+    organization: Organization, actor: RequestPrincipal | None = None
 ) -> bool:
     """Whether investigations are available to `actor` in `organization`.
 

src/sentry/seer/agent/client.py

@@ -8,7 +8,6 @@
 from datetime import datetime
 from typing import Any, Literal
 
-from django.contrib.auth.models import AnonymousUser
 from django.db.models import Q, QuerySet
 from django.utils import timezone as django_timezone
 from django.utils.timezone import now
@@ -17,6 +16,7 @@
 from urllib3 import BaseHTTPResponse
 
 from sentry import features, options
+from sentry.auth.principal import RequestPrincipal, acting_user_id
 from sentry.constants import ENABLE_SEER_CODING_DEFAULT, ObjectStatus
 from sentry.identity import default_manager as identity_manager
 from sentry.identity.mcp import McpIdentityProvider
@@ -66,8 +66,6 @@
 from sentry.seer.signed_seer_api import SeerViewerContext
 from sentry.tasks.seer.context_engine_index import build_service_map, index_org_project_knowledge
 from sentry.tasks.seer.explorer_index import dispatch_explorer_index_projects
-from sentry.users.models.user import User
-from sentry.users.services.user import RpcUser
 from sentry.utils.prompts import (
     get_prompt_activities_for_user,
     seer_monitoring_provider_dont_ask_feature,
@@ -315,7 +313,7 @@ def execute(cls, organization: Organization, run_id: int) -> None:
     def __init__(
         self,
         organization: Organization,
-        user: User | RpcUser | AnonymousUser | None = None,
+        user: RequestPrincipal | None = None,
         project: Project | None = None,
         group: Group | None = None,
         category_key: str | None = None,
@@ -339,6 +337,7 @@ def __init__(
     ):
         self.organization = organization
         self.user = user
+        self._user_id = acting_user_id(user) if user is not None else None
         self.project = project
         self.group = group
         self.custom_tools = custom_tools or []
@@ -391,8 +390,8 @@ def __init__(
 
     def _build_viewer_context(self) -> SeerViewerContext:
         context = SeerViewerContext(organization_id=self.organization.id)
-        if self.user and hasattr(self.user, "id") and self.user.id is not None:
-            context["user_id"] = self.user.id
+        if self._user_id is not None:
+            context["user_id"] = self._user_id
         return context
 
     def start_run(
@@ -515,11 +514,7 @@ def start_run(
             )
         )
 
-        user_id = (
-            self.user.id
-            if self.user and hasattr(self.user, "id") and self.user.id is not None
-            else None
-        )
+        user_id = self._user_id
 
         def _create_agent_run(run: SeerRun) -> None:
             if record_in_history:
@@ -587,11 +582,7 @@ def start_feature_run(
         Explicit agent_run_options override any options derived from organization
         configuration.
         """
-        user_id = (
-            self.user.id
-            if self.user and hasattr(self.user, "id") and self.user.id is not None
-            else None
-        )
+        user_id = self._user_id
 
         def _create_agent_run(run: SeerRun) -> None:
             SeerAgentRun.objects.create(
@@ -830,11 +821,7 @@ def continue_run(
         if ui_tools:
             chat_body["ui_tools"] = ui_tools
 
-        monitoring_user_id = (
-            self.user.id
-            if self.user and not isinstance(self.user, AnonymousUser) and self.user.id is not None
-            else None
-        )
+        monitoring_user_id = self._user_id
         monitoring_provider_connections = get_monitoring_provider_connections(
             self.organization, monitoring_user_id
         )
@@ -932,7 +919,7 @@ def get_runs(
         start: datetime | None = None,
         end: datetime | None = None,
     ) -> QuerySet[SeerRun]:
-        user_id = int(self.user.id) if self.user is not None and self.user.id is not None else None
+        user_id = self._user_id
         queryset = queryset_for_query(query, self.organization, user_id)
         queryset = queryset.filter(
             Q(agent__project_id__isnull=True) | Q(agent__project_id__in=accessible_project_ids)
@@ -954,13 +941,8 @@ def latest_run(
             run__organization_id=self.organization.id,
             source=self.category_key or "",
         )
-        if (
-            only_current_user
-            and self.user
-            and hasattr(self.user, "id")
-            and self.user.id is not None
-        ):
-            qs = qs.filter(run__user_id=int(self.user.id))
+        if only_current_user and self._user_id is not None:
+            qs = qs.filter(run__user_id=self._user_id)
         if group_id is not None:
             qs = qs.filter(group_id=group_id)
         elif self.category_value is not None:

src/sentry/seer/agent/client_utils.py

@@ -15,13 +15,13 @@
 
 import orjson
 from django.conf import settings
-from django.contrib.auth.models import AnonymousUser
 from django.db import router, transaction
 from django.utils.timezone import now
 from rest_framework.request import Request
 from urllib3 import BaseHTTPResponse, HTTPConnectionPool
 
 from sentry import features
+from sentry.auth.principal import RequestPrincipal, has_user
 from sentry.constants import ObjectStatus
 from sentry.hybridcloud.models.outbox import (
     CellOutbox,
@@ -392,7 +392,7 @@ def get_agent_state_from_pr_id(
 
 def has_seer_agent_access_with_detail(
     organization: Organization | RpcOrganization,
-    actor: SentryUser | AnonymousUser | RpcUser | None = None,
+    actor: RequestPrincipal | None = None,
 ) -> tuple[bool, str | None]:
     """
     Check if the actor has access to Seer Agent.
@@ -421,7 +421,7 @@ def has_seer_agent_access_with_detail(
 
 
 def collect_user_org_context(
-    user: SentryUser | RpcUser | AnonymousUser | None,
+    user: RequestPrincipal | None,
     organization: Organization,
     request: Request | None = None,
 ) -> UserOrgContext:
@@ -442,7 +442,7 @@ def collect_user_org_context(
         for p in all_projects
     ]
 
-    if user is None or isinstance(user, AnonymousUser):
+    if user is None or not has_user(user):
         return {
             "org_slug": organization.slug,
             "all_org_projects": all_org_projects,

src/sentry/seer/autofix/commit_author.py

@@ -2,10 +2,10 @@
 from collections.abc import Sequence
 from typing import NotRequired, TypedDict
 
-from django.contrib.auth.models import AnonymousUser
 from django.db.models import Q
 from sentry_sdk import traces
 
+from sentry.auth.principal import RequestPrincipal, has_user
 from sentry.integrations.models.external_actor import ExternalActor
 from sentry.integrations.types import ExternalProviders
 from sentry.seer.autofix.pr_iteration.feedback import Feedback
@@ -18,8 +18,6 @@
 )
 from sentry.seer.autofix.pr_iteration.feedback_sources.user_ui import UserUIFeedbackSource
 from sentry.seer.utils import get_github_username_for_user
-from sentry.users.models.user import User
-from sentry.users.services.user.model import RpcUser
 from sentry.users.services.user.service import user_service
 from sentry.utils import json, metrics
 
@@ -103,13 +101,13 @@ def _is_github_enterprise_only(user_id: int, organization_id: int) -> bool:
 
 
 def commit_author_for_user(
-    user: User | RpcUser | AnonymousUser | None,
+    user: RequestPrincipal | None,
     organization_id: int,
     *,
     referrer: str,
 ) -> SeerCommitAuthor | None:
     """The acting user's GitHub identity, or ``None`` to let Seer author the commit."""
-    if user is None or isinstance(user, AnonymousUser) or user.id is None:
+    if user is None or not has_user(user) or user.id is None:
         _record_outcome("no_acting_user")
         return None