|
@@ -16,9 +16,7 @@ from sentry.notifications.types import (
|
|
|
NOTIFICATION_SETTING_TYPES,
|
|
|
NotificationScopeEnum,
|
|
|
NotificationSettingEnum,
|
|
|
- NotificationSettingOptionValues,
|
|
|
NotificationSettingsOptionEnum,
|
|
|
- NotificationSettingTypes,
|
|
|
)
|
|
|
from sentry.services.hybrid_cloud.actor import ActorType, RpcActor
|
|
|
from sentry.services.hybrid_cloud.auth.model import AuthenticationContext
|
|
@@ -35,41 +33,37 @@ from sentry.types.integrations import ExternalProviderEnum, ExternalProviders
|
|
|
|
|
|
|
|
|
class DatabaseBackedNotificationsService(NotificationsService):
|
|
|
- def update_settings(
|
|
|
- self,
|
|
|
- *,
|
|
|
- external_provider: ExternalProviders,
|
|
|
- notification_type: NotificationSettingTypes,
|
|
|
- setting_option: NotificationSettingOptionValues,
|
|
|
- actor: RpcActor,
|
|
|
- project_id: Optional[int] = None,
|
|
|
- organization_id: Optional[int] = None,
|
|
|
- skip_provider_updates: bool = False,
|
|
|
- organization_id_for_team: Optional[int] = None,
|
|
|
- ) -> None:
|
|
|
- NotificationSetting.objects.update_settings(
|
|
|
- provider=external_provider,
|
|
|
- type=notification_type,
|
|
|
- value=setting_option,
|
|
|
- project=project_id,
|
|
|
- organization=organization_id,
|
|
|
- actor=actor,
|
|
|
- skip_provider_updates=skip_provider_updates,
|
|
|
- )
|
|
|
-
|
|
|
def enable_all_settings_for_provider(
|
|
|
self,
|
|
|
*,
|
|
|
external_provider: ExternalProviderEnum,
|
|
|
- user_id: int,
|
|
|
+ user_id: Optional[int] = None,
|
|
|
+ team_id: Optional[int] = None,
|
|
|
+ types: Optional[List[NotificationSettingEnum]] = None,
|
|
|
) -> None:
|
|
|
+ assert (team_id and not user_id) or (
|
|
|
+ user_id and not team_id
|
|
|
+ ), "Can only enable settings for team or user"
|
|
|
+
|
|
|
+ kwargs: MutableMapping[str, str | int] = {}
|
|
|
+ if user_id:
|
|
|
+ kwargs["user_id"] = user_id
|
|
|
+ kwargs["scope_type"] = NotificationScopeEnum.USER.value
|
|
|
+ kwargs["scope_identifier"] = user_id
|
|
|
+ elif team_id:
|
|
|
+ kwargs["team_id"] = team_id
|
|
|
+ kwargs["scope_type"] = NotificationScopeEnum.TEAM.value
|
|
|
+ kwargs["scope_identifier"] = team_id
|
|
|
+
|
|
|
+ type_str_list = list(map(lambda t: t.value, types)) if types else None
|
|
|
with transaction.atomic(router.db_for_write(NotificationSettingProvider)):
|
|
|
for type_str in NOTIFICATION_SETTING_TYPES.values():
|
|
|
+ # check the type if it's an input
|
|
|
+ if type_str_list and type_str not in type_str_list:
|
|
|
+ continue
|
|
|
NotificationSettingProvider.objects.create_or_update(
|
|
|
+ **kwargs,
|
|
|
provider=external_provider.value,
|
|
|
- user_id=user_id,
|
|
|
- scope_type=NotificationScopeEnum.USER.value,
|
|
|
- scope_identifier=user_id,
|
|
|
type=type_str,
|
|
|
values={
|
|
|
"value": NotificationSettingsOptionEnum.ALWAYS.value,
|