fix typing - kinda went all over

src/sentry/sentry_apps/api/endpoints/sentry_app_details.py

@@ -25,6 +25,8 @@
 from sentry.sentry_apps.logic import SentryAppUpdater
 from sentry.sentry_apps.models.sentry_app import SentryApp
 from sentry.sentry_apps.models.sentry_app_installation import SentryAppInstallation
+from sentry.users.models.user import User
+from sentry.users.services.user.model import RpcUser
 from sentry.utils.audit import create_audit_entry
 
 logger = logging.getLogger(__name__)
@@ -91,6 +93,10 @@ def put(self, request: Request, sentry_app) -> Response:
 
  if serializer.is_valid():
  result = serializer.validated_data
+
+ assert isinstance(request.user, User) or isinstance(
+ request.user, RpcUser
+ ), "User must be authenticated to update a Sentry App"
  updated_app = SentryAppUpdater(
  sentry_app=sentry_app,
  name=result.get("name"),

src/sentry/sentry_apps/api/endpoints/sentry_app_publish_request.py

@@ -1,3 +1,5 @@
+from collections.abc import Iterable
+
 from rest_framework.request import Request
 from rest_framework.response import Response
 
@@ -10,6 +12,8 @@
 from sentry.models.avatars.sentry_app_avatar import SentryAppAvatar, SentryAppAvatarTypes
 from sentry.models.organizationmapping import OrganizationMapping
 from sentry.sentry_apps.logic import SentryAppUpdater
+from sentry.users.models.user import User
+from sentry.users.services.user.model import RpcUser
 from sentry.utils import email
 
 
@@ -54,6 +58,9 @@ def post(self, request: Request, sentry_app) -> Response:
  status=400,
  )
 
+ assert isinstance(request.user, User) or isinstance(
+ request.user, RpcUser
+ ), "User must be authenticated to update a Sentry App"
  SentryAppUpdater(
  sentry_app=sentry_app,
  status=SentryAppStatus.PUBLISH_REQUEST_INPROGRESS_STR,
@@ -65,7 +72,8 @@ def post(self, request: Request, sentry_app) -> Response:
  org_slug = "<unknown>" if org_mapping is None else org_mapping.slug
  message = f"User {request.user.email} of organization {org_slug} wants to publish {sentry_app.slug}\n"
 
- for question_pair in request.data.get("questionnaire"):
+ questionnaire: Iterable[dict[str, str]] = request.data.get("questionnaire", [])
+ for question_pair in questionnaire:
  message += "\n\n>{}\n{}".format(question_pair["question"], question_pair["answer"])
 
  subject = "Sentry Integration Publication Request from %s" % org_slug

src/sentry/sentry_apps/api/endpoints/sentry_app_requests.py

@@ -18,15 +18,15 @@
 INVALID_DATE_FORMAT_MESSAGE = "Invalid date format. Format must be YYYY-MM-DD HH:MM:SS."
 
 
-def filter_by_date(request: Mapping[str, Any], start: float, end: float) -> bool:
+def filter_by_date(request: Mapping[str, Any], start: datetime, end: datetime) -> bool:
  date_str = request.get("date")
  if not date_str:
  return False
  timestamp = datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S.%f+00:00").replace(microsecond=0)
  return start <= timestamp <= end
 
 
-def filter_by_organization(request: Mapping[str, Any], organization: Organization) -> bool:
+def filter_by_organization(request: Mapping[str, Any], organization: Organization | None) -> bool:
  if not organization:
  return True
  return request["organization_id"] == organization.id
@@ -57,27 +57,31 @@ def get(self, request: Request, sentry_app) -> Response:
  :qparam string start: Optionally specify a date to begin at. Format must be YYYY-MM-DD HH:MM:SS
  :qparam string end: Optionally specify a date to end at. Format must be YYYY-MM-DD HH:MM:SS
  """
+
  date_format = "%Y-%m-%d %H:%M:%S"
- now = datetime.now().strftime(date_format)
- default_start = "2000-01-01 00:00:00"
+ start_time: datetime = datetime.strptime("2000-01-01 00:00:00", date_format)
+ end_time: datetime = datetime.now()
 
  event_type = request.GET.get("eventType")
  errors_only = request.GET.get("errorsOnly")
  org_slug = request.GET.get("organizationSlug")
- start = request.GET.get("start", default_start)
- end = request.GET.get("end", now)
+ start_parameter = request.GET.get("start", None)
+ end_parameter = request.GET.get("end", None)
 
  try:
- start = datetime.strptime(start, date_format)
+ start_time = (
+ datetime.strptime(start_parameter, date_format) if start_parameter else start_time
+ )
  except ValueError:
  return Response({"detail": INVALID_DATE_FORMAT_MESSAGE}, status=400)
 
  try:
- end = datetime.strptime(end, date_format)
+
+ end_time = datetime.strptime(end_parameter, date_format) if end_parameter else end_time
  except ValueError:
  return Response({"detail": INVALID_DATE_FORMAT_MESSAGE}, status=400)
 
- kwargs = {}
+ kwargs: dict[Any, Any] = {}
  if event_type:
  if event_type not in EXTENDED_VALID_EVENTS:
  return Response({"detail": "Invalid event type."}, status=400)
@@ -95,7 +99,9 @@ def get(self, request: Request, sentry_app) -> Response:
 
  filtered_requests = []
  for i, req in enumerate(buffer.get_requests(**kwargs)):
- if filter_by_date(req, start, end) and filter_by_organization(req, organization):
+ if filter_by_date(req, start_time, end_time) and filter_by_organization(
+ req, organization
+ ):
  filtered_requests.append(BufferedRequest(id=i, data=req))
 
  return Response(serialize(filtered_requests, request.user, RequestSerializer(sentry_app)))

src/sentry/sentry_apps/api/endpoints/sentry_apps.py

@@ -18,6 +18,8 @@
 from sentry.constants import SentryAppStatus
 from sentry.sentry_apps.logic import SentryAppCreator
 from sentry.sentry_apps.models.sentry_app import SentryApp
+from sentry.users.models.user import User
+from sentry.users.services.user.model import RpcUser
 from sentry.users.services.user.service import user_service
 
 logger = logging.getLogger(__name__)
@@ -115,6 +117,9 @@ def post(self, request: Request, organization) -> Response:
  data["author"] = data["author"] or organization.name
 
  try:
+ assert isinstance(request.user, User) or isinstance(
+ request.user, RpcUser
+ ), "User must be authenticated to create a Sentry App"
  sentry_app = SentryAppCreator(
  name=data["name"],
  author=data["author"],

src/sentry/sentry_apps/installations.py

@@ -19,6 +19,7 @@
 from sentry.sentry_apps.services.hook import hook_service
 from sentry.tasks.sentry_apps import installation_webhook
 from sentry.users.models.user import User
+from sentry.users.services.user.model import RpcUser
 from sentry.utils import metrics
 
 
@@ -28,7 +29,7 @@ class SentryAppInstallationTokenCreator:
  expires_at: datetime.date | None = None
  generate_audit: bool = False
 
- def run(self, user: User, request: HttpRequest | None = None) -> ApiToken:
+ def run(self, user: User | RpcUser, request: HttpRequest | None = None) -> ApiToken:
  with transaction.atomic(router.db_for_write(ApiToken)):
  self._check_token_limit()
  api_token = self._create_api_token()
@@ -100,7 +101,7 @@ class SentryAppInstallationCreator:
  slug: str
  notify: bool = True
 
- def run(self, *, user: User, request: HttpRequest | None) -> SentryAppInstallation:
+ def run(self, *, user: User | RpcUser, request: HttpRequest | None) -> SentryAppInstallation:
  metrics.incr("sentry_apps.installation.attempt")
  with transaction.atomic(router.db_for_write(ApiGrant)):
  api_grant = self._create_api_grant()

src/sentry/sentry_apps/logic.py

@@ -101,7 +101,7 @@ class SentryAppUpdater:
  popularity: int | None = None
  features: list[int] | None = None
 
- def run(self, user: User) -> SentryApp:
+ def run(self, user: User | RpcUser) -> SentryApp:
  with transaction.atomic(router.db_for_write(User)):
  self._update_name()
  self._update_author()
@@ -122,7 +122,7 @@ def run(self, user: User) -> SentryApp:
  self.record_analytics(user, new_schema_elements)
  return self.sentry_app
 
- def _update_features(self, user: User) -> None:
+ def _update_features(self, user: User | RpcUser) -> None:
  if self.features is not None:
  if not _is_elevated_user(user) and self.sentry_app.status == SentryAppStatus.PUBLISHED:
  raise APIError("Cannot update features on a published integration.")
@@ -141,7 +141,7 @@ def _update_author(self) -> None:
  if self.author is not None:
  self.sentry_app.author = self.author
 
- def _update_status(self, user: User) -> None:
+ def _update_status(self, user: User | RpcUser) -> None:
  if self.status is not None:
  if _is_elevated_user(user):
  if self.status == SentryAppStatus.PUBLISHED_STR:
@@ -239,7 +239,7 @@ def _update_allowed_origins(self) -> None:
  self.sentry_app.application.allowed_origins = "\n".join(self.allowed_origins)
  self.sentry_app.application.save()
 
- def _update_popularity(self, user: User) -> None:
+ def _update_popularity(self, user: User | RpcUser) -> None:
  if self.popularity is not None:
  if _is_elevated_user(user):
  self.sentry_app.popularity = self.popularity
@@ -269,7 +269,7 @@ def _create_ui_components(self) -> None:
  type=element["type"], sentry_app_id=self.sentry_app.id, schema=element
  )
 
- def record_analytics(self, user: User, new_schema_elements: set[str] | None) -> None:
+ def record_analytics(self, user: User | RpcUser, new_schema_elements: set[str] | None) -> None:
  analytics.record(
  "sentry_app.updated",
  user_id=user.id,
@@ -306,7 +306,7 @@ def __post_init__(self) -> None:
  def run(
  self,
  *,
- user: User,
+ user: User | RpcUser,
  request: HttpRequest | None = None,
  skip_default_auth_token: bool = False,
  ) -> SentryApp:
@@ -407,7 +407,7 @@ def _create_integration_feature(self, sentry_app: SentryApp) -> None:
  sentry_sdk.capture_message("IntegrityError while creating IntegrationFeature")
 
  def _install(
- self, *, slug: str, user: User, request: HttpRequest | None
+ self, *, slug: str, user: User | RpcUser, request: HttpRequest | None
  ) -> SentryAppInstallation:
  return SentryAppInstallationCreator(
  organization_id=self.organization_id,
@@ -416,7 +416,7 @@ def _install(
  ).run(user=user, request=request)
 
  def _create_access_token(
- self, user: User, install: SentryAppInstallation, request: HttpRequest | None
+ self, user: User | RpcUser, install: SentryAppInstallation, request: HttpRequest | None
  ) -> None:
  install.api_token = SentryAppInstallationTokenCreator(sentry_app_installation=install).run(
  request=request, user=user
@@ -444,7 +444,7 @@ def audit(self, request: HttpRequest | None, sentry_app: SentryApp) -> None:
  data={"name": sentry_app.name},
  )
 
- def record_analytics(self, user: User, sentry_app: SentryApp) -> None:
+ def record_analytics(self, user: User | RpcUser, sentry_app: SentryApp) -> None:
  analytics.record(
  "sentry_app.created",
  user_id=user.id,