diff --git a/src/sentry/notifications/notificationcontroller.py b/src/sentry/notifications/notificationcontroller.py index 65774ab3f9d0..10a02dee3e12 100644 --- a/src/sentry/notifications/notificationcontroller.py +++ b/src/sentry/notifications/notificationcontroller.py @@ -2,6 +2,7 @@ from collections import defaultdict from collections.abc import Iterable, Mapping, MutableMapping +from functools import cached_property from typing import Union from django.db.models import Q @@ -50,6 +51,39 @@ def sort_settings_by_scope(setting: NotificationSettingOption | NotificationSett return 0 +def _group_by_recipient[T: (NotificationSettingOption, NotificationSettingProvider)]( + settings: Iterable[T], +) -> Mapping[tuple[str, int], list[T]]: + by_recipient: defaultdict[tuple[str, int], list[T]] = defaultdict(list) + for setting in settings: + if setting.user_id is not None: + by_recipient[("user_id", setting.user_id)].append(setting) + if setting.team_id is not None: + by_recipient[("team_id", setting.team_id)].append(setting) + return by_recipient + + +def _filter_settings[T: (NotificationSettingOption, NotificationSettingProvider)]( + settings: Iterable[T], + by_recipient: Mapping[tuple[str, int], list[T]], + kwargs: Mapping[str, object], +) -> list[T]: + # Filters are applied once per recipient, so narrow to that recipient's + # settings first instead of scanning every recipient's settings each time. + candidates: Iterable[T] = settings + for key in ("user_id", "team_id"): + recipient_id = kwargs.get(key) + if isinstance(recipient_id, int): + candidates = by_recipient.get((key, recipient_id), []) + break + + return [ + setting + for setting in candidates + if all(getattr(setting, arg) == kwargs[arg] for arg in kwargs) + ] + + class NotificationController: _setting_options: Iterable[NotificationSettingOption] = [] _setting_providers: Iterable[NotificationSettingProvider] = [] @@ -154,19 +188,27 @@ def _get_query(self) -> Q: return project_settings | org_settings | user_settings | team_settings + @cached_property + def _setting_options_by_recipient( + self, + ) -> Mapping[tuple[str, int], list[NotificationSettingOption]]: + return _group_by_recipient(self.get_all_setting_options) + + @cached_property + def _setting_providers_by_recipient( + self, + ) -> Mapping[tuple[str, int], list[NotificationSettingProvider]]: + return _group_by_recipient(self.get_all_setting_providers) + def _filter_options(self, **kwargs: object) -> list[NotificationSettingOption]: - return [ - setting - for setting in self.get_all_setting_options - if all(getattr(setting, arg) == kwargs[arg] for arg in kwargs) - ] + return _filter_settings( + self.get_all_setting_options, self._setting_options_by_recipient, kwargs + ) def _filter_providers(self, **kwargs: object) -> list[NotificationSettingProvider]: - return [ - setting - for setting in self.get_all_setting_providers - if all(getattr(setting, arg) == kwargs[arg] for arg in kwargs) - ] + return _filter_settings( + self.get_all_setting_providers, self._setting_providers_by_recipient, kwargs + ) def _get_layered_setting_options( self, diff --git a/tests/sentry/notifications/test_notificationcontroller.py b/tests/sentry/notifications/test_notificationcontroller.py index 4e8a21034d64..a6b8cdfa3cc7 100644 --- a/tests/sentry/notifications/test_notificationcontroller.py +++ b/tests/sentry/notifications/test_notificationcontroller.py @@ -214,6 +214,49 @@ def test_filter_setting_providers(self) -> None: ) assert filtered_providers == [self.setting_providers[0]] + def test_filter_settings_by_recipient(self) -> None: + other_user = self.create_user() + self.create_member(user=other_user, organization=self.organization, teams=[self.team]) + other_user_option = add_notification_setting_option( + scope_type=NotificationScopeEnum.USER, + scope_identifier=other_user.id, + type=NotificationSettingEnum.DEPLOY, + value=NotificationSettingsOptionEnum.NEVER, + user_id=other_user.id, + ) + team_option = add_notification_setting_option( + scope_type=NotificationScopeEnum.TEAM, + scope_identifier=self.team.id, + type=NotificationSettingEnum.DEPLOY, + value=NotificationSettingsOptionEnum.ALWAYS, + team_id=self.team.id, + ) + team_provider = add_notification_setting_provider( + scope_type=NotificationScopeEnum.TEAM, + scope_identifier=self.team.id, + provider=ExternalProviderEnum.SLACK, + type=NotificationSettingEnum.DEPLOY, + value=NotificationSettingsOptionEnum.ALWAYS, + team_id=self.team.id, + ) + controller = NotificationController( + recipients=[self.user, other_user, self.team], + project_ids=[self.project.id], + organization_id=self.organization.id, + ) + + assert controller._filter_options(user_id=self.user.id) == self.setting_options + assert controller._filter_options(user_id=other_user.id) == [other_user_option] + assert controller._filter_options(team_id=self.team.id) == [team_option] + assert controller._filter_options( + user_id=self.user.id, type=NotificationSettingEnum.DEPLOY.value + ) == [self.setting_options[0]] + assert controller._filter_options(user_id=self.create_user().id) == [] + + assert controller._filter_providers(user_id=self.user.id) == self.setting_providers + assert controller._filter_providers(team_id=self.team.id) == [team_provider] + assert controller._filter_providers(user_id=other_user.id) == [] + def test_layering(self) -> None: NotificationSettingOption.objects.all().delete() top_level_option = add_notification_setting_option(