|
1 | 1 | from collections import defaultdict |
2 | | -from typing import Any, Dict, Iterable, Mapping, Optional, Tuple |
| 2 | +from typing import Any, Dict, Iterable, List, Mapping, Optional, Tuple |
3 | 3 |
|
| 4 | +from sentry.models.integration import ExternalProviders |
4 | 5 | from sentry.notifications.legacy_mappings import get_legacy_value |
5 | 6 | from sentry.notifications.types import ( |
6 | 7 | NotificationScopeType, |
|
9 | 10 | ) |
10 | 11 |
|
11 | 12 |
|
12 | | -def _get_setting_value_from_mapping( |
| 13 | +def _get_setting_mapping_from_mapping( |
13 | 14 | notification_settings_by_user: Mapping[ |
14 | | - Any, Mapping[NotificationScopeType, NotificationSettingOptionValues] |
| 15 | + Any, |
| 16 | + Mapping[NotificationScopeType, Mapping[ExternalProviders, NotificationSettingOptionValues]], |
15 | 17 | ], |
16 | 18 | user: Any, |
17 | 19 | type: NotificationSettingTypes, |
18 | | - default: NotificationSettingOptionValues, |
19 | | -) -> NotificationSettingOptionValues: |
| 20 | +) -> Mapping[ExternalProviders, NotificationSettingOptionValues]: |
| 21 | + # XXX(CEO): may not respect granularity of a setting for Slack a setting for email |
| 22 | + # but we'll worry about that later since we don't have a FE for it yet |
20 | 23 | specific_scope = get_scope_type(type) |
21 | | - notification_settings_option = notification_settings_by_user.get(user) |
22 | | - if notification_settings_option: |
23 | | - notification_setting_option = notification_settings_option.get( |
| 24 | + notification_settings_mapping = notification_settings_by_user.get(user) |
| 25 | + if notification_settings_mapping: |
| 26 | + notification_setting_option = notification_settings_mapping.get( |
24 | 27 | specific_scope |
25 | | - ) or notification_settings_option.get(NotificationScopeType.USER) |
| 28 | + ) or notification_settings_mapping.get(NotificationScopeType.USER) |
26 | 29 | if notification_setting_option: |
27 | 30 | return notification_setting_option |
28 | | - return default |
| 31 | + return {ExternalProviders.EMAIL: NotificationSettingOptionValues.ALWAYS} |
29 | 32 |
|
30 | 33 |
|
31 | | -def should_user_be_notified( |
| 34 | +def where_should_user_be_notified( |
32 | 35 | notification_settings_by_user: Mapping[ |
33 | | - Any, Mapping[NotificationScopeType, NotificationSettingOptionValues] |
| 36 | + Any, |
| 37 | + Mapping[NotificationScopeType, Mapping[ExternalProviders, NotificationSettingOptionValues]], |
34 | 38 | ], |
35 | 39 | user: Any, |
36 | | -) -> bool: |
| 40 | +) -> List[ExternalProviders]: |
37 | 41 | """ |
38 | 42 | Given a mapping of default and specific notification settings by user, |
39 | | - determine if a user should receive an ISSUE_ALERT notification. |
| 43 | + return the list of providers after verifying the user has opted into this notification. |
40 | 44 | """ |
41 | | - return ( |
42 | | - _get_setting_value_from_mapping( |
43 | | - notification_settings_by_user, |
44 | | - user, |
45 | | - NotificationSettingTypes.ISSUE_ALERTS, |
46 | | - NotificationSettingOptionValues.ALWAYS, |
47 | | - ) |
48 | | - == NotificationSettingOptionValues.ALWAYS |
| 45 | + mapping = _get_setting_mapping_from_mapping( |
| 46 | + notification_settings_by_user, |
| 47 | + user, |
| 48 | + NotificationSettingTypes.ISSUE_ALERTS, |
| 49 | + ) |
| 50 | + return list( |
| 51 | + filter(lambda elem: mapping[elem] == NotificationSettingOptionValues.ALWAYS, mapping) |
49 | 52 | ) |
50 | 53 |
|
51 | 54 |
|
52 | | -def should_be_participating( |
| 55 | +def where_should_be_participating( |
53 | 56 | user: Any, |
54 | 57 | subscriptions_by_user_id: Mapping[int, Any], |
55 | 58 | notification_settings_by_user: Mapping[ |
56 | | - Any, Mapping[NotificationScopeType, NotificationSettingOptionValues] |
| 59 | + Any, |
| 60 | + Mapping[NotificationScopeType, Mapping[ExternalProviders, NotificationSettingOptionValues]], |
57 | 61 | ], |
58 | | -) -> bool: |
| 62 | +) -> List[ExternalProviders]: |
59 | 63 | """ |
60 | 64 | Given a mapping of users to subscriptions and a mapping of default and |
61 | | - specific notification settings by user, determine if a user should receive |
| 65 | + specific notification settings by user, determine where a user should receive |
62 | 66 | a WORKFLOW notification. Unfortunately, this algorithm does not respect |
63 | 67 | NotificationSettingOptionValues.ALWAYS. If the user is unsubscribed from |
64 | 68 | the group, that overrides their notification preferences. |
65 | 69 | """ |
66 | | - value = _get_setting_value_from_mapping( |
| 70 | + mapping = _get_setting_mapping_from_mapping( |
67 | 71 | notification_settings_by_user, |
68 | 72 | user, |
69 | 73 | NotificationSettingTypes.WORKFLOW, |
70 | | - NotificationSettingOptionValues.SUBSCRIBE_ONLY, |
71 | 74 | ) |
72 | | - |
73 | | - if value == NotificationSettingOptionValues.NEVER: |
74 | | - return False |
75 | | - |
76 | | - subscription = subscriptions_by_user_id.get(user.id) |
77 | | - if subscription: |
78 | | - return bool(subscription.is_active) |
79 | | - |
80 | | - return value == NotificationSettingOptionValues.ALWAYS |
| 75 | + output = [] |
| 76 | + for provider, value in mapping.items(): |
| 77 | + subscription = subscriptions_by_user_id.get(user.id) |
| 78 | + if (subscription and not subscription.is_active) or ( |
| 79 | + value == NotificationSettingOptionValues.NEVER |
| 80 | + ): |
| 81 | + continue |
| 82 | + if (subscription and subscription.is_active) or ( |
| 83 | + value == NotificationSettingOptionValues.ALWAYS |
| 84 | + ): |
| 85 | + output.append(provider) |
| 86 | + return output |
81 | 87 |
|
82 | 88 |
|
83 | 89 | def transform_to_notification_settings_by_user( |
84 | 90 | notification_settings: Iterable[Any], |
85 | 91 | users: Iterable[Any], |
86 | | -) -> Mapping[Any, Mapping[NotificationScopeType, NotificationSettingOptionValues]]: |
| 92 | +) -> Mapping[ |
| 93 | + Any, Mapping[NotificationScopeType, Mapping[ExternalProviders, NotificationSettingOptionValues]] |
| 94 | +]: |
87 | 95 | """ |
88 | 96 | Given a unorganized list of notification settings, create a mapping of |
89 | 97 | users to a map of notification scopes to setting values. |
90 | 98 | """ |
91 | 99 | actor_mapping = {user.actor_id: user for user in users} |
92 | 100 | notification_settings_by_user: Dict[ |
93 | | - Any, Dict[NotificationScopeType, NotificationSettingOptionValues] |
94 | | - ] = defaultdict(dict) |
| 101 | + Any, Dict[NotificationScopeType, Dict[ExternalProviders, NotificationSettingOptionValues]] |
| 102 | + ] = defaultdict(lambda: defaultdict(dict)) |
95 | 103 | for notification_setting in notification_settings: |
96 | 104 | user = actor_mapping.get(notification_setting.target_id) |
97 | | - notification_settings_by_user[user][ |
98 | | - NotificationScopeType(notification_setting.scope_type) |
99 | | - ] = NotificationSettingOptionValues(notification_setting.value) |
| 105 | + scope_type = NotificationScopeType(notification_setting.scope_type) |
| 106 | + value = NotificationSettingOptionValues(notification_setting.value) |
| 107 | + provider = ExternalProviders(notification_setting.provider) |
| 108 | + notification_settings_by_user[user][scope_type][provider] = value |
100 | 109 | return notification_settings_by_user |
101 | 110 |
|
102 | 111 |
|
|
0 commit comments