mirror of
https://github.com/freedombox/FreedomBox.git
synced 2026-05-20 10:34:30 +00:00
notification: New API for showing better notifications
Closes: #867. Signed-off-by: Sunil Mohan Adapa <sunil@medhas.org> Reviewed-by: James Valleroy <jvalleroy@mailbox.org>
This commit is contained in:
Sunil Mohan Adapa
James Valleroy
parent
5c514e91d3
commit
8e0a94282f
217
plinth/notification.py
Normal file
217
plinth/notification.py
Normal file
@ -0,0 +1,217 @@
|
||||
#
|
||||
# This file is part of FreedomBox.
|
||||
#
|
||||
# This program is free software: you can redistribute it and/or modify
|
||||
# it under the terms of the GNU Affero General Public License as
|
||||
# published by the Free Software Foundation, either version 3 of the
|
||||
# License, or (at your option) any later version.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU Affero General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU Affero General Public License
|
||||
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
#
|
||||
"""
|
||||
Module to provide API for showing notifications.
|
||||
"""
|
||||
|
||||
import copy
|
||||
import logging
|
||||
|
||||
from django.core.exceptions import ValidationError
|
||||
from django.db.models import Q
|
||||
from django.template.exceptions import TemplateDoesNotExist
|
||||
from django.template.response import SimpleTemplateResponse
|
||||
from django.utils.translation import ugettext
|
||||
|
||||
from plinth import cfg
|
||||
|
||||
from . import models
|
||||
|
||||
severities = {'exception': 5, 'error': 4, 'warning': 3, 'info': 2, 'debug': 1}
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class Notification(models.StoredNotification):
|
||||
"""Wrapper over Notification model to provide the API."""
|
||||
class Meta: # pylint: disable=too-few-public-methods
|
||||
"""Meta properties of the Notification model."""
|
||||
proxy = True
|
||||
|
||||
@property
|
||||
def severity_value(self):
|
||||
"""Return severity as a numeric value suitable for comparison."""
|
||||
try:
|
||||
return severities[self.severity]
|
||||
except KeyError:
|
||||
return severities['info']
|
||||
|
||||
def dismiss(self, should_dismiss=True):
|
||||
"""Mark the notification as read or unread."""
|
||||
self.dismissed = should_dismiss
|
||||
super().save()
|
||||
|
||||
def clean(self):
|
||||
"""Perform additional validations on the model."""
|
||||
if self.severity not in ('exception', 'error', 'warning', 'info',
|
||||
'debug'):
|
||||
raise ValidationError('Invalid severity')
|
||||
|
||||
if self.actions:
|
||||
self._validate_actions(self.actions)
|
||||
|
||||
if (self.message or self.actions) and self.body_template:
|
||||
raise ValidationError(
|
||||
'Either body_template or message and actions must exist')
|
||||
|
||||
@staticmethod
|
||||
def _validate_actions(actions):
|
||||
"""Check that actions structure is valid."""
|
||||
if not isinstance(actions, list):
|
||||
raise ValidationError('Actions must be a list')
|
||||
|
||||
for action in actions:
|
||||
if not isinstance(action, dict):
|
||||
raise ValidationError('Action must a dictionary')
|
||||
|
||||
if 'type' not in action:
|
||||
raise ValidationError('Action must have a type')
|
||||
|
||||
if action['type'] not in ('dismiss', 'link'):
|
||||
raise ValidationError('Action type must be dismiss or link')
|
||||
|
||||
if action['type'] == 'dismiss':
|
||||
continue
|
||||
|
||||
if 'text' not in action or 'url' not in action:
|
||||
raise ValidationError('Action must have text and url')
|
||||
|
||||
if 'class' in action and action['class'] not in (
|
||||
'primary', 'default', 'warning', 'danger', 'success',
|
||||
'info'):
|
||||
raise ValidationError('Invalid action class')
|
||||
|
||||
@staticmethod
|
||||
def update_or_create(**kwargs):
|
||||
"""Update a notification or create one if necessary."""
|
||||
id = kwargs.pop('id')
|
||||
return Notification.objects.update_or_create(defaults=kwargs, id=id)[0]
|
||||
|
||||
@staticmethod
|
||||
def get(key): # pylint: disable=redefined-builtin
|
||||
"""Return a notification with matching ID."""
|
||||
# pylint: disable=no-member
|
||||
try:
|
||||
return Notification.objects.get(pk=key)
|
||||
except Notification.DoesNotExist:
|
||||
raise KeyError('No such notification')
|
||||
|
||||
@staticmethod
|
||||
def list(key=None, app_id=None, user=None, dismissed=False):
|
||||
"""Return a list of notifications for a user."""
|
||||
filters = []
|
||||
if key:
|
||||
filters.append(Q(id=key))
|
||||
|
||||
if app_id:
|
||||
filters.append(Q(app_id=app_id))
|
||||
|
||||
if user:
|
||||
# XXX: Consider implementing caching for user groups
|
||||
groups = user.groups.values_list('name', flat=True)
|
||||
filters.append(Q(user__isnull=True) | Q(user=user.username))
|
||||
filters.append(Q(group__isnull=True) | Q(group__in=groups))
|
||||
|
||||
if dismissed is not None:
|
||||
filters.append(Q(dismissed=dismissed))
|
||||
|
||||
return Notification.objects.filter(*filters)[0:10]
|
||||
|
||||
@staticmethod
|
||||
def _translate(string, data=None):
|
||||
"""Translate a string for final display using data dict."""
|
||||
if not string:
|
||||
return None
|
||||
|
||||
string = ugettext(string)
|
||||
try:
|
||||
string = str(string)
|
||||
if data:
|
||||
string = string.format(**data)
|
||||
except KeyError as error:
|
||||
logger.warning(
|
||||
'Notification missing required key during translation: %s',
|
||||
error)
|
||||
|
||||
return string
|
||||
|
||||
@staticmethod
|
||||
def _translate_dict(data_dict, data=None):
|
||||
"""Translate strings inside a data dict for display."""
|
||||
if not data_dict:
|
||||
return data_dict
|
||||
|
||||
new_dict = {}
|
||||
for key, value in data_dict.items():
|
||||
if isinstance(value, str) and value.startswith('translate:'):
|
||||
value = value.split(':', maxsplit=1)[1]
|
||||
value = Notification._translate(value, data)
|
||||
elif isinstance(value, dict):
|
||||
value = Notification._translate_dict(value, data)
|
||||
else:
|
||||
value = copy.deepcopy(value)
|
||||
|
||||
new_dict[key] = value
|
||||
|
||||
return new_dict
|
||||
|
||||
@staticmethod
|
||||
def _render(template, data):
|
||||
"""Use the template name and render it."""
|
||||
if not template:
|
||||
return None
|
||||
|
||||
context = dict(data, box_name=ugettext(cfg.box_name))
|
||||
try:
|
||||
return SimpleTemplateResponse(template, context).render()
|
||||
except TemplateDoesNotExist:
|
||||
# Developer only error, no i18n
|
||||
return {'content': f'Template {template} does not exist.'.encode()}
|
||||
|
||||
@staticmethod
|
||||
def get_display_context(user):
|
||||
"""Return a list of notifications meant for display to a user."""
|
||||
notifications = Notification.list(user=user)
|
||||
max_severity = max(notifications, default=None,
|
||||
key=lambda note: note.severity_value)
|
||||
max_severity = max_severity.severity if max_severity else None
|
||||
|
||||
notes = []
|
||||
for note in notifications:
|
||||
data = Notification._translate_dict(note.data, note.data)
|
||||
actions = copy.deepcopy(note.actions)
|
||||
for action in actions:
|
||||
if 'text' in action:
|
||||
action['text'] = Notification._translate(
|
||||
action['text'], data)
|
||||
|
||||
notes.append({
|
||||
'id': note.id,
|
||||
'app_id': note.app_id,
|
||||
'severity': note.severity,
|
||||
'title': Notification._translate(note.title, data),
|
||||
'message': Notification._translate(note.message, data),
|
||||
'body': Notification._render(note.body_template, data),
|
||||
'actions': actions,
|
||||
'data': data,
|
||||
'created_time': note.created_time,
|
||||
'last_update_time': note.last_update_time,
|
||||
'user': note.user,
|
||||
'group': note.group,
|
||||
'dismissed': note.dismissed,
|
||||
})
|
||||
|
||||
return {'notifications': notes, 'max_severity': max_severity}
|
||||
Loading…
x
Reference in New Issue
Block a user