Source code for argus.notificationprofile.media.base

from __future__ import annotations

from abc import ABC, abstractmethod
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from collections.abc import Iterable

    from types import NoneType
    from typing import Union

    from django.contrib.auth import get_user_model
    from django.db.models.query import QuerySet

    from argus.incident.models import Event
    from ..models import DestinationConfig
    from ..serializers import RequestDestinationConfigSerializer

    User = get_user_model()


__all__ = ["NotificationMedium"]


[docs] class NotificationMedium(ABC): class NotDeletableError(Exception): """ Custom exception class that is raised when a destination cannot be deleted """
[docs] @classmethod @abstractmethod def validate(cls, instance: RequestDestinationConfigSerializer, dict: dict, user: User) -> dict: """ Validates the settings of destination and returns a dict with validated and cleaned data """ pass
[docs] @classmethod @abstractmethod def has_duplicate(cls, queryset: QuerySet, settings: dict) -> bool: """ Returns True if a destination with the given settings already exists in the given queryset """ pass
[docs] @staticmethod @abstractmethod def get_label(destination: DestinationConfig) -> str: """ Returns a descriptive label for this destination. """ pass
@classmethod def get_relevant_addresses(cls, destinations: Iterable[DestinationConfig]) -> set[DestinationConfig]: """Returns a set of addresses the message should be sent to""" pass
[docs] @classmethod @abstractmethod def send(cls, event: Event, destinations: Iterable[DestinationConfig], **kwargs) -> bool: """Sends message about a given event to the given destinations""" pass
[docs] @classmethod def raise_if_not_deletable(cls, destination: DestinationConfig) -> NoneType: """ Raises a NotDeletableError if the given destination is not able to be deleted (if it is in use by any notification profiles) """ connected_profiles = destination.notification_profiles.all() if connected_profiles: profiles = ", ".join([str(profile) for profile in connected_profiles]) raise cls.NotDeletableError( f"Cannot delete this destination since it is in use in the notification profile(s): {profiles}." )
[docs] @staticmethod def update(destination: DestinationConfig, validated_data: dict) -> Union[DestinationConfig, NoneType]: """ Updates a destination in case the normal update function is not sufficient and returns the updated destination in that case, returns None otherwise """ return None