Coverage for app / backend / src / couchers / notifications / settings.py: 90%
47 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-03-19 14:14 +0000
1import logging
3from sqlalchemy import select
4from sqlalchemy.orm import Session
6from couchers.db import session_scope
7from couchers.i18n import LocalizationContext
8from couchers.models import (
9 NotificationDelivery,
10 NotificationDeliveryType,
11 NotificationPreference,
12 NotificationTopicAction,
13)
14from couchers.notifications.utils import get_topic_action_description
15from couchers.proto import notifications_pb2
17logger = logging.getLogger(__name__)
20def get_preference(
21 session: Session, user_id: int, topic_action: NotificationTopicAction
22) -> list[NotificationDeliveryType]:
23 """
24 Gets the user's preference from the DB or otherwise falls back to defaults
26 Must be done in session scope
28 Returns list of delivery types
29 """
30 overrides = {
31 res.delivery_type: res.deliver
32 for res in session.execute(
33 select(NotificationPreference)
34 .where(NotificationPreference.user_id == user_id)
35 .where(NotificationPreference.topic_action == topic_action)
36 )
37 .scalars()
38 .all()
39 }
40 return [dt for dt in NotificationDeliveryType if overrides.get(dt, dt in topic_action.defaults)]
43def get_topic_actions_by_delivery_type(
44 session: Session, user_id: int, delivery_type: NotificationDeliveryType
45) -> set[NotificationTopicAction]:
46 """
47 Given push/email/digest, returns notifications that this user has enabled for that type.
48 """
49 overrides: dict[NotificationTopicAction, bool] = {}
50 for topic_action, deliver in session.execute(
51 select(NotificationPreference.topic_action, NotificationPreference.deliver)
52 .where(NotificationPreference.user_id == user_id)
53 .where(NotificationPreference.delivery_type == delivery_type)
54 ).all():
55 overrides[topic_action] = deliver
56 return {t for t in NotificationTopicAction if overrides.get(t, delivery_type in t.defaults)}
59def reset_preference(
60 session: Session,
61 user_id: int,
62 topic_action: NotificationTopicAction,
63 delivery_type: NotificationDeliveryType,
64) -> None:
65 current_pref = session.execute(
66 select(NotificationPreference)
67 .where(NotificationPreference.user_id == user_id)
68 .where(NotificationPreference.topic_action == topic_action)
69 .where(NotificationDelivery.delivery_type == delivery_type)
70 ).scalar_one_or_none()
71 if current_pref:
72 session.delete(current_pref)
73 session.flush()
76class PreferenceNotUserEditableError(Exception):
77 pass
80def set_preference(
81 session: Session,
82 user_id: int,
83 topic_action: NotificationTopicAction,
84 delivery_type: NotificationDeliveryType,
85 deliver: bool,
86) -> None:
87 if topic_action.is_critical:
88 raise PreferenceNotUserEditableError()
89 current_pref = session.execute(
90 select(NotificationPreference)
91 .where(NotificationPreference.user_id == user_id)
92 .where(NotificationPreference.topic_action == topic_action)
93 .where(NotificationPreference.delivery_type == delivery_type)
94 ).scalar_one_or_none()
95 if current_pref:
96 current_pref.deliver = deliver
97 else:
98 session.add(
99 NotificationPreference(
100 user_id=user_id,
101 topic_action=topic_action,
102 delivery_type=delivery_type,
103 deliver=deliver,
104 )
105 )
106 session.flush()
109settings_layout = [
110 (
111 "Core Features",
112 [
113 (
114 "host_request",
115 "Host requests",
116 [
117 NotificationTopicAction.host_request__create,
118 NotificationTopicAction.host_request__accept,
119 NotificationTopicAction.host_request__confirm,
120 NotificationTopicAction.host_request__reject,
121 NotificationTopicAction.host_request__cancel,
122 NotificationTopicAction.host_request__message,
123 NotificationTopicAction.host_request__missed_messages,
124 NotificationTopicAction.host_request__reminder,
125 ],
126 ),
127 (
128 "activeness",
129 "Activity Check-in",
130 [
131 NotificationTopicAction.activeness__probe,
132 ],
133 ),
134 (
135 "chat",
136 "Messaging",
137 [
138 NotificationTopicAction.chat__message,
139 NotificationTopicAction.chat__missed_messages,
140 ],
141 ),
142 (
143 "reference",
144 "References",
145 [
146 NotificationTopicAction.reference__receive_hosted,
147 NotificationTopicAction.reference__receive_surfed,
148 NotificationTopicAction.reference__receive_friend,
149 NotificationTopicAction.reference__reminder_hosted,
150 NotificationTopicAction.reference__reminder_surfed,
151 ],
152 ),
153 ],
154 ),
155 (
156 "Community Features",
157 [
158 (
159 "friend_request",
160 "Friend requests",
161 [
162 NotificationTopicAction.friend_request__create,
163 NotificationTopicAction.friend_request__accept,
164 ],
165 ),
166 (
167 "event",
168 "Events",
169 [
170 NotificationTopicAction.event__create_approved,
171 NotificationTopicAction.event__create_any,
172 NotificationTopicAction.event__update,
173 NotificationTopicAction.event__comment,
174 NotificationTopicAction.event__cancel,
175 NotificationTopicAction.event__delete,
176 NotificationTopicAction.event__invite_organizer,
177 NotificationTopicAction.event__reminder,
178 ],
179 ),
180 (
181 "discussion",
182 "Community discussions",
183 [
184 NotificationTopicAction.discussion__create,
185 NotificationTopicAction.discussion__comment,
186 ],
187 ),
188 (
189 "thread",
190 "Threads, Comments, & Replies",
191 [
192 NotificationTopicAction.thread__reply,
193 ],
194 ),
195 ],
196 ),
197 (
198 "Account Settings",
199 [
200 (
201 "onboarding",
202 "Onboarding",
203 [
204 NotificationTopicAction.onboarding__reminder,
205 ],
206 ),
207 (
208 "badge",
209 "Updates to Badges on your profile",
210 [
211 NotificationTopicAction.badge__add,
212 NotificationTopicAction.badge__remove,
213 ],
214 ),
215 (
216 "donation",
217 "Donations",
218 [
219 NotificationTopicAction.donation__received,
220 ],
221 ),
222 ],
223 ),
224 (
225 "Account Security",
226 [
227 (
228 "password",
229 "Password change",
230 [
231 NotificationTopicAction.password__change,
232 ],
233 ),
234 (
235 "password_reset",
236 "Password reset",
237 [
238 NotificationTopicAction.password_reset__start,
239 NotificationTopicAction.password_reset__complete,
240 ],
241 ),
242 (
243 "email_address",
244 "Email address change",
245 [
246 NotificationTopicAction.email_address__change,
247 NotificationTopicAction.email_address__verify,
248 ],
249 ),
250 (
251 "account_deletion",
252 "Account deletion",
253 [
254 NotificationTopicAction.account_deletion__start,
255 NotificationTopicAction.account_deletion__complete,
256 NotificationTopicAction.account_deletion__recovered,
257 ],
258 ),
259 (
260 "api_key",
261 "API keys",
262 [
263 NotificationTopicAction.api_key__create,
264 ],
265 ),
266 (
267 "phone_number",
268 "Phone number change",
269 [
270 NotificationTopicAction.phone_number__change,
271 NotificationTopicAction.phone_number__verify,
272 ],
273 ),
274 (
275 "birthdate",
276 "Birthdate change",
277 [
278 NotificationTopicAction.birthdate__change,
279 ],
280 ),
281 (
282 "gender",
283 "Displayed gender change",
284 [
285 NotificationTopicAction.gender__change,
286 ],
287 ),
288 (
289 "modnote",
290 "Moderator notes",
291 [
292 NotificationTopicAction.modnote__create,
293 ],
294 ),
295 (
296 "verification",
297 "Verification",
298 [
299 NotificationTopicAction.verification__sv_fail,
300 NotificationTopicAction.verification__sv_success,
301 ],
302 ),
303 (
304 "postal_verification",
305 "Postal Verification",
306 [
307 NotificationTopicAction.postal_verification__postcard_sent,
308 NotificationTopicAction.postal_verification__success,
309 NotificationTopicAction.postal_verification__failed,
310 ],
311 ),
312 ],
313 ),
314 (
315 "Other Notifications",
316 [
317 (
318 "general",
319 "General",
320 [
321 NotificationTopicAction.general__new_blog_post,
322 ],
323 ),
324 ],
325 ),
326]
329def get_user_setting_groups(
330 user_id: int, loc_context: LocalizationContext
331) -> list[notifications_pb2.NotificationGroup]:
332 with session_scope() as session:
333 groups = []
334 for heading, group in settings_layout:
335 topics = []
336 for topic, name, items in group:
337 actions = []
338 for topic_action in items:
339 delivery_types = get_preference(session, user_id, topic_action)
340 description = get_topic_action_description(topic_action, locale=loc_context.locale)
341 actions.append(
342 notifications_pb2.NotificationItem(
343 action=topic_action.action,
344 description=description,
345 user_editable=not topic_action.is_critical,
346 push=NotificationDeliveryType.push in delivery_types,
347 email=NotificationDeliveryType.email in delivery_types,
348 digest=NotificationDeliveryType.digest in delivery_types,
349 )
350 )
351 topics.append(
352 notifications_pb2.NotificationTopic(
353 topic=topic,
354 name=name,
355 items=actions,
356 )
357 )
358 groups.append(
359 notifications_pb2.NotificationGroup(
360 heading=heading,
361 topics=topics,
362 )
363 )
364 return groups