notifications: add user preferences filter

invenio_users_resources/notifications.py

@@ -0,0 +1,60 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2023 Graz University of Technology.
+#
+# Invenio-Users-Resources is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""User specific resources for notifications."""
+
+
+from invenio_notifications.backends.email import EmailNotificationBackend
+from invenio_notifications.models import Recipient
+from invenio_notifications.services.builders import (
+    RecipientBackendGenerator,
+    RecipientGenerator,
+)
+from invenio_notifications.services.filters import RecipientFilter
+from invenio_records.dictutils import dict_lookup
+
+
+class UserPreferencesRecipientFilter(RecipientFilter):
+    """Recipient filter for notifications being enabled at all."""
+
+    def __call__(self, notification, recipients):
+        """Filter recipients."""
+        for key in list(recipients.keys()):
+            r = recipients[key]
+            if not (
+                r.data.get("preferences", {})
+                .get("notifications", {})
+                .get("enabled", False)
+            ):
+                del recipients[key]
+
+        return recipients
+
+
+class UserRecipient(RecipientGenerator):
+    """User recipient generator for a notification."""
+
+    def __init__(self, key):
+        """Ctor."""
+        self.key = key
+
+    def __call__(self, notification, recipients):
+        """Update required recipient information and add backend id."""
+        user = dict_lookup(notification.context, self.key)
+        recipients[user["id"]] = Recipient(data=user)
+        return recipients
+
+
+class UserEmailBackend(RecipientBackendGenerator):
+    """User related email backend generator for a notification."""
+
+    def __call__(self, notification, recipient, backends):
+        """Update required recipient information and add backend id."""
+        backend_id = EmailNotificationBackend.id
+        backends.append(backend_id)
+        return backend_id

invenio_users_resources/services/schemas.py

@@ -91,3 +91,9 @@ class UserGhostSchema(Schema):
         dump_only=True,
     )
     is_ghost = fields.Boolean(dump_only=True)
+
+
+class NotificationPreferences(Schema):
+    """Schema for notification preferences."""
+
+    enabled: fields.Bool()

setup.cfg

@@ -29,6 +29,7 @@ zip_safe = False
 install_requires =
     invenio-accounts>=2.2.0,<3.0.0
     invenio-i18n>=2.0.0
+    invenio-notifications>=0.1.0,<1.0.0
     invenio-oauthclient>=2.2.0,<3.0.0
     invenio-records-resources>=4.0.0,<5.0.0
 

tests/test_notifications.py

@@ -0,0 +1,43 @@
+# -*- coding: utf-8 -*-
+#
+# Copyright (C) 2023 Graz University of Technology.
+#
+# Invenio-Users-Resources is free software; you can redistribute it and/or
+# modify it under the terms of the MIT License; see LICENSE file for more
+# details.
+
+"""Notification related tests."""
+
+from copy import deepcopy
+
+from invenio_notifications.models import Notification, Recipient
+
+from invenio_users_resources.notifications import UserPreferencesRecipientFilter
+from invenio_users_resources.records.api import UserAggregate
+
+
+def test_user_recipient_filter(user_pub):
+    """Test user recipient filter for notifications."""
+    preferences_filter = UserPreferencesRecipientFilter()
+
+    u = UserAggregate.from_user(user_pub.user).dumps()
+
+    user_notifications_enabled = deepcopy(u)
+    user_notifications_enabled["preferences"]["notifications"] = {"enabled": True}
+    user_notifications_disabled = deepcopy(u)
+    user_notifications_disabled["preferences"]["notifications"] = {"enabled": False}
+
+    n = Notification(type="", context={})
+    recipient_enabled = Recipient(data=user_notifications_enabled)
+    recipient_disabled = Recipient(data=user_notifications_disabled)
+
+    filtered_recipients = preferences_filter(
+        notification=n,
+        recipients={
+            user_notifications_disabled["id"]: recipient_disabled,
+            user_notifications_enabled["id"]: recipient_enabled,
+        },
+    )
+
+    assert 1 == len(filtered_recipients)
+    assert recipient_enabled == filtered_recipients[user_notifications_enabled["id"]]