Source code for kadi.lib.notifications.core

# Copyright 2020 Karlsruhe Institute of Technology
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from flask import current_app
from flask_babel import gettext as _

from kadi.ext.db import db
from kadi.lib.plugins.core import run_hook
from kadi.lib.tasks.models import Task
from kadi.lib.tasks.models import TaskState

from .models import NotificationName


[docs] def create_notification_data(notification): """Create notification data suitable for presenting it to a client. Uses the :func:`kadi.plugins.spec.kadi_get_background_task_notification` plugin hook for handling task status notifications. :param notification: A :class:`.Notification` object to use for creating the notification data. :return: A tuple containing the title and the HTML body of the notification. """ title = body = notification.name # Task status notifications. if notification.name == NotificationName.TASK_STATUS: title = _("Task status") task = Task.query.filter(Task.id == notification.data["task_id"]).first() if task is None: return title, _("Task no longer exists.") if task.state == TaskState.PENDING: body = _("Waiting for available resources...") elif task.state == TaskState.RUNNING: body = _("Task running...") elif task.state == TaskState.SUCCESS: body = _("Task succeeded.") elif task.state == TaskState.FAILURE: body = _("Task failed.") elif task.state == TaskState.REVOKED: body = _("Task revoked.") try: data = run_hook("kadi_get_background_task_notification", task=task) if isinstance(data, tuple) and len(data) == 2: title = data[0] if data[0] is not None else title body = data[1] if data[1] is not None else body except Exception as e: current_app.logger.exception(e) title = f"{title} ({task.pretty_state})" return title, body
[docs] def dismiss_notification(notification): """Dismiss a notification. If the notification is of type ``"task_status"``, the referenced task will be revoked as well. :param notification: The :class:`.Notification` to dismiss. """ if notification.name == NotificationName.TASK_STATUS: task = Task.query.filter(Task.id == notification.data["task_id"]).first() if task is not None: task.revoke() db.session.delete(notification)