oncall-engine/engine/apps/api/serializers/alert_receive_channel.py

import typing
from collections import OrderedDict

from django.conf import settings
from django.core.exceptions import ValidationError as DjangoValidationError
from drf_spectacular.utils import PolymorphicProxySerializer, extend_schema_field
from jinja2 import TemplateSyntaxError
from rest_framework import serializers
from rest_framework.exceptions import ValidationError
from rest_framework.fields import SerializerMethodField

from apps.alerts.grafana_alerting_sync_manager.grafana_alerting_sync import GrafanaAlertingSyncManager
from apps.alerts.models import AlertReceiveChannel
from apps.base.messaging import get_messaging_backends
from apps.integrations.legacy_prefix import has_legacy_prefix
from apps.labels.models import LabelKeyCache, LabelValueCache
from apps.labels.types import LabelKey
from apps.user_management.models import Organization
from common.api_helpers.custom_fields import TeamPrimaryKeyRelatedField
from common.api_helpers.exceptions import BadRequest
from common.api_helpers.mixins import APPEARANCE_TEMPLATE_NAMES, EagerLoadingMixin
from common.jinja_templater import jinja_template_env

from .integration_heartbeat import IntegrationHeartBeatSerializer
from .labels import LabelsSerializerMixin


def _additional_settings_serializer_from_type(integration_type: str) -> serializers.Serializer:
    """Return serializer class for given integration_type additional settings."""
    cls = None
    config = AlertReceiveChannel.get_config_from_type(integration_type)
    cls = getattr(config, "additional_settings_serializer", None) if config else None
    return cls


# TODO: refactor this types as we no longer support storing static labels in this field.
# AlertGroupCustomLabelValue represents custom alert group label value for API requests
# It handles two types of label's value:
# 1. Just Label Value from a label repo for a static label
# 2. Templated Label value which is actually a jinja template for a dynamic label.
class AlertGroupCustomLabelValueAPI(typing.TypedDict):
    id: str | None  # None for templated labels, label value ID for plain labels
    name: str  # Jinja template for templated labels, label value name for plain labels
    prescribed: bool  # Indicates of selected label value is prescribed. Not applicable for templated values.


# AlertGroupCustomLabel represents Alert group custom label for API requests
# Key is just a LabelKey from label repo, while value could be value from repo or a jinja template.
class AlertGroupCustomLabelAPI(typing.TypedDict):
    key: LabelKey
    value: AlertGroupCustomLabelValueAPI


AlertGroupCustomLabelsAPI = list[AlertGroupCustomLabelAPI]


class IntegrationAlertGroupLabels(typing.TypedDict):
    inheritable: dict[str, bool] | None  # Deprecated
    custom: AlertGroupCustomLabelsAPI
    template: str | None


AlertReceiveChannelAdditionalSettingsSerializers = (
    i.additional_settings_serializer
    for i in AlertReceiveChannel._config
    if getattr(i, "additional_settings_serializer", None)
)


@extend_schema_field(
    PolymorphicProxySerializer(
        component_name="AdditionalSettingsField",
        serializers=list(AlertReceiveChannelAdditionalSettingsSerializers),
        resource_type_field_name=None,
    )
)
class AdditionalSettingsField(serializers.DictField):
    pass


class CustomLabelSerializer(serializers.Serializer):
    """
    This serializer is consistent with apps.api.serializers.labels.LabelPairSerializer,
    but allows null for value ID to support templated labels.
    """

    class CustomLabelKeySerializer(serializers.Serializer):
        id = serializers.CharField()
        name = serializers.CharField()
        prescribed = serializers.BooleanField(default=False)

    class CustomLabelValueSerializer(serializers.Serializer):
        # ID is null for templated labels. For such labels, the "name" value is a Jinja2 template.
        id = serializers.CharField(allow_null=True)
        name = serializers.CharField()
        prescribed = serializers.BooleanField(default=False)

    key = CustomLabelKeySerializer()
    value = CustomLabelValueSerializer()


class IntegrationAlertGroupLabelsSerializer(serializers.Serializer):
    # todo: inheritable field is deprecated. Remove in a future release
    inheritable = serializers.DictField(child=serializers.BooleanField(), required=False)
    custom = CustomLabelSerializer(many=True)
    template = serializers.CharField(allow_null=True)

    def to_representation(self, instance: AlertReceiveChannel) -> IntegrationAlertGroupLabels:
        """
        The API representation of alert group labels is very different from the underlying model.

        "inheritable" field is deprecated. Kept for api-backward compatibility. Will be removed in a future release
        "custom" is based on AlertReceiveChannel.alert_group_labels_custom, a JSONField with a different schema.
        "template" is based on AlertReceiveChannel.alert_group_labels_template, this one is straightforward.
        """

        return {
            # todo: "inheritable" field is deprecated, remove in a future release.
            "inheritable": {label.key_id: True for label in instance.labels.all()},
            "custom": self._custom_labels_to_representation(instance.alert_group_labels_custom),
            "template": instance.alert_group_labels_template,
        }

    def to_internal_value(self, validated_data: dict) -> dict:
        """
        to_internal_value converts dynamic labels from API format to internal format and updates labels cache
        """
        alert_group_labels = self._pop_alert_group_labels(validated_data)
        if alert_group_labels is None:
            return validated_data

        organization = self.context["request"].auth.organization
        self._create_custom_labels(organization, alert_group_labels["custom"] if alert_group_labels else [])

        custom_labels = (
            self._custom_labels_to_internal_value(alert_group_labels["custom"]) if alert_group_labels else []
        )
        validated_data["alert_group_labels_custom"] = custom_labels or None
        validated_data["alert_group_labels_template"] = alert_group_labels["template"] if alert_group_labels else None

        return validated_data

    @staticmethod
    def _custom_labels_to_representation(
        custom_labels: AlertReceiveChannel.DynamicLabelsConfigDB,
    ) -> AlertGroupCustomLabelsAPI:
        """
        Inverse of the _custom_labels_to_internal_value method above.
        Fetches label names from DB cache, so the API response schema is consistent with other label endpoints.
        """

        from apps.labels.models import LabelKeyCache, LabelValueCache

        if custom_labels is None:
            return []

        # build index of keys id to name and prescribed flag
        label_key_index = {
            k.id: {"name": k.name, "prescribed": k.prescribed}
            for k in LabelKeyCache.objects.filter(id__in=[label[0] for label in custom_labels]).only(
                "id", "name", "prescribed"
            )
        }

        # build index of values id to name and prescribed flag
        label_value_index = {
            v.id: {"name": v.name, "prescribed": v.prescribed}
            for v in LabelValueCache.objects.filter(id__in=[label[1] for label in custom_labels if label[1]]).only(
                "id", "name", "prescribed"
            )
        }

        return [
            {
                "key": {
                    "id": key_id,
                    "name": label_key_index[key_id]["name"],
                    "prescribed": label_key_index[key_id]["prescribed"],
                },
                "value": {
                    "id": value_id if value_id else None,
                    "name": label_value_index[value_id]["name"] if value_id else typing.cast(str, template),
                    "prescribed": label_value_index[value_id]["prescribed"] if value_id else False,
                },
            }
            for key_id, value_id, template in custom_labels
            if key_id in label_key_index and (value_id in label_value_index or not value_id)
        ]

    @staticmethod
    def _custom_labels_to_internal_value(
        custom_labels: AlertGroupCustomLabelsAPI,
    ) -> AlertReceiveChannel.DynamicLabelsConfigDB:
        """
        Convert dynamic labels from API representation to the schema used by the JSONField on the model:
        [[key.id, None, template(stored in value.name here)]].
        """

        return [
            [label["key"]["id"], None, label["value"]["name"]]
            for label in custom_labels
            if label["value"]["id"] is None
            # value.id is not None for deprecated static labels, for dynamic labels it's always None
        ]

    @staticmethod
    def _pop_alert_group_labels(validated_data: dict) -> IntegrationAlertGroupLabels | None:
        # the "alert_group_labels" field is optional, so either all 2 fields (custom and template) are present or none
        # "inheritable" field is deprecated
        if "custom" not in validated_data:
            return None

        return {
            "inheritable": validated_data.pop("inheritable", None),  # deprecated
            "custom": validated_data.pop("custom"),
            "template": validated_data.pop("template"),
        }

    @staticmethod
    def _create_custom_labels(organization: Organization, labels: AlertGroupCustomLabelsAPI) -> None:
        """Create LabelKeyCache and LabelValueCache objects for labels used in labelsSchema"""

        label_keys = [
            LabelKeyCache(
                id=label["key"]["id"],
                name=label["key"]["name"],
                prescribed=label["key"]["prescribed"],
                organization=organization,
            )
            for label in labels
        ]

        label_values = [
            LabelValueCache(
                id=label["value"]["id"],
                name=label["value"]["name"],
                prescribed=label["value"]["prescribed"],
                key_id=label["key"]["id"],
            )
            for label in labels
            if label["value"]["id"]  # don't create LabelValueCache objects for templated labels
        ]

        LabelKeyCache.objects.bulk_create(label_keys, ignore_conflicts=True, batch_size=5000)
        LabelValueCache.objects.bulk_create(label_values, ignore_conflicts=True, batch_size=5000)


class AlertReceiveChannelSerializer(
    EagerLoadingMixin, LabelsSerializerMixin, serializers.ModelSerializer[AlertReceiveChannel]
):
    id = serializers.CharField(read_only=True, source="public_primary_key")
    integration_url = serializers.ReadOnlyField()
    alert_count = serializers.SerializerMethodField()
    alert_groups_count = serializers.SerializerMethodField()
    author = serializers.CharField(read_only=True, source="author.public_primary_key")
    organization = serializers.CharField(read_only=True, source="organization.public_primary_key")
    team = TeamPrimaryKeyRelatedField(allow_null=True, required=False)
    is_able_to_autoresolve = serializers.ReadOnlyField()
    default_channel_filter = serializers.SerializerMethodField()
    instructions = serializers.SerializerMethodField()
    demo_alert_enabled = serializers.BooleanField(source="is_demo_alert_enabled", read_only=True)
    is_based_on_alertmanager = serializers.BooleanField(source="based_on_alertmanager", read_only=True)
    maintenance_till = serializers.ReadOnlyField(source="till_maintenance_timestamp")
    heartbeat = IntegrationHeartBeatSerializer(read_only=True, allow_null=True, source="integration_heartbeat")
    allow_delete = serializers.SerializerMethodField()
    description_short = serializers.CharField(max_length=250, required=False, allow_null=True)
    demo_alert_payload = serializers.JSONField(source="config.example_payload", read_only=True)
    routes_count = serializers.SerializerMethodField()
    connected_escalations_chains_count = serializers.SerializerMethodField()
    inbound_email = serializers.CharField(required=False, read_only=True)
    is_legacy = serializers.SerializerMethodField()
    alert_group_labels = IntegrationAlertGroupLabelsSerializer(source="*", required=False)
    additional_settings = AdditionalSettingsField(allow_null=True, allow_empty=False, required=False, default=None)

    # integration heartbeat is in PREFETCH_RELATED not by mistake.
    # With using of select_related ORM builds strange join
    # which leads to incorrect heartbeat-alert_receive_channel binding in result
    PREFETCH_RELATED = ["channel_filters", "integration_heartbeat", "labels", "labels__key", "labels__value"]
    SELECT_RELATED = ["organization", "author", "team"]

    class Meta:
        model = AlertReceiveChannel
        fields = [
            "id",
            "description",
            "description_short",
            "integration",
            "smile_code",
            "verbal_name",
            "author",
            "organization",
            "team",
            "created_at",
            "integration_url",
            "alert_count",
            "alert_groups_count",
            "allow_source_based_resolving",
            "instructions",
            "is_able_to_autoresolve",
            "default_channel_filter",
            "demo_alert_enabled",
            "maintenance_mode",
            "maintenance_till",
            "heartbeat",
            "is_available_for_integration_heartbeat",
            "allow_delete",
            "demo_alert_payload",
            "routes_count",
            "connected_escalations_chains_count",
            "is_based_on_alertmanager",
            "inbound_email",
            "is_legacy",
            "labels",
            "alert_group_labels",
            "alertmanager_v2_migrated_at",
            "additional_settings",
        ]
        read_only_fields = [
            "created_at",
            "author",
            "organization",
            "smile_code",
            "integration_url",
            "instructions",
            "demo_alert_enabled",
            "maintenance_mode",
            "demo_alert_payload",
            "routes_count",
            "connected_escalations_chains_count",
            "is_based_on_alertmanager",
            "inbound_email",
            "is_legacy",
            "alertmanager_v2_migrated_at",
        ]
        extra_kwargs = {"integration": {"required": True}}

    def to_internal_value(self, data):
        settings_serializer_cls = (
            _additional_settings_serializer_from_type(self.instance.config.slug) if self.instance else None
        )
        if settings_serializer_cls:
            additional_settings_data = data.get("additional_settings", self.instance.additional_settings)
            settings_serializer = settings_serializer_cls(self.instance, data=additional_settings_data)
            settings_serializer.is_valid()
            if settings_serializer.errors:
                raise ValidationError({"additional_settings": settings_serializer.errors})
            data["additional_settings"] = settings_serializer.to_internal_value(additional_settings_data)
        return super().to_internal_value(data)

    def to_representation(self, instance):
        result = super().to_representation(instance)
        if instance.additional_settings:
            settings_serializer_cls = _additional_settings_serializer_from_type(instance.config.slug)
            if settings_serializer_cls:
                settings_serializer = settings_serializer_cls(instance)
                result["additional_settings"] = settings_serializer.to_representation(instance)
        return result

    def validate(self, data):
        validated_data = super().validate(data)
        self.validate_name_uniqueness(validated_data)
        return validated_data

    def validate_name_uniqueness(self, validated_data):
        organization = self.context["request"].auth.organization
        verbal_name = validated_data.get("verbal_name")
        team = validated_data.get("team", self.instance.team) if self.instance else validated_data.get("team")
        try:
            obj = AlertReceiveChannel.objects.get(organization=organization, team=team, verbal_name=verbal_name)
        except AlertReceiveChannel.DoesNotExist:
            pass
        except AlertReceiveChannel.MultipleObjectsReturned:
            raise serializers.ValidationError(
                {"verbal_name": "An integration with this name already exists for this team"}
            )
        else:
            if self.instance is None or obj.id != self.instance.id:
                raise serializers.ValidationError(
                    {"verbal_name": "An integration with this name already exists for this team"}
                )

    def create(self, validated_data):
        organization = self.context["request"].auth.organization
        integration = validated_data.get("integration")
        create_default_webhooks = validated_data.pop("create_default_webhooks", True)
        if has_legacy_prefix(integration):
            raise BadRequest(detail="This integration is deprecated")
        if integration == AlertReceiveChannel.INTEGRATION_GRAFANA_ALERTING:
            connection_error = GrafanaAlertingSyncManager.check_for_connection_errors(organization)
            if connection_error:
                raise BadRequest(detail=connection_error)
        for _integration in AlertReceiveChannel._config:
            if _integration.slug == integration:
                is_able_to_autoresolve = _integration.is_able_to_autoresolve

        # pop associated labels, so they are not passed to AlertReceiveChannel.create. They will be created later.
        labels = validated_data.pop("labels", None)

        try:
            instance = AlertReceiveChannel.create(
                **validated_data,
                organization=organization,
                author=self.context["request"].user,
                allow_source_based_resolving=is_able_to_autoresolve,
            )
        except AlertReceiveChannel.DuplicateDirectPagingError:
            raise BadRequest(detail=AlertReceiveChannel.DuplicateDirectPagingError.DETAIL)

        # Create label associations
        self.update_labels_association_if_needed(labels, instance, organization)

        # Create default webhooks if needed
        if create_default_webhooks and hasattr(instance.config, "create_default_webhooks"):
            instance.config.create_default_webhooks(instance)

        # Create default service_name label
        instance.create_service_name_dynamic_label()

        return instance

    def update(self, instance, validated_data):
        # update associated labels
        labels = validated_data.pop("labels", None)
        self.update_labels_association_if_needed(labels, instance, self.context["request"].auth.organization)

        try:
            updated_instance = super().update(instance, validated_data)
        except AlertReceiveChannel.DuplicateDirectPagingError:
            raise BadRequest(detail=AlertReceiveChannel.DuplicateDirectPagingError.DETAIL)

        # update webhooks if needed, using updated instance
        if hasattr(instance.config, "update_default_webhooks"):
            instance.config.update_default_webhooks(updated_instance)

        return updated_instance

    def get_instructions(self, obj: "AlertReceiveChannel") -> str:
        # Deprecated, kept for api-backward compatibility
        return ""

    # MethodFields are used instead of relevant properties because of properties hit db on each instance in queryset
    def get_default_channel_filter(self, obj: "AlertReceiveChannel") -> str | None:
        for filter in obj.channel_filters.all():
            if filter.is_default:
                return filter.public_primary_key
        return None

    @staticmethod
    def validate_integration(integration):
        if integration is None or integration not in AlertReceiveChannel.WEB_INTEGRATION_CHOICES:
            raise BadRequest(detail="invalid integration")

        if integration == AlertReceiveChannel.INTEGRATION_DIRECT_PAGING:
            raise BadRequest(detail="Direct paging integrations can't be created")

        return integration

    def validate_additional_settings(self, data):
        integration = self.instance.integration if self.instance else self.initial_data.get("integration")
        settings_serializer_cls = _additional_settings_serializer_from_type(integration)
        if settings_serializer_cls:
            if not data:
                raise ValidationError(["This field is required for this integration."])
            serializer = settings_serializer_cls(data=data)
            serializer.is_valid(raise_exception=True)
            data = serializer.validated_data
        elif data is not None:
            raise ValidationError(["Invalid data"])
        return data

    def get_allow_delete(self, obj: "AlertReceiveChannel") -> bool:
        # don't allow deleting direct paging integrations
        return obj.integration != AlertReceiveChannel.INTEGRATION_DIRECT_PAGING

    def get_alert_count(self, obj: "AlertReceiveChannel") -> int:
        return 0

    def get_alert_groups_count(self, obj: "AlertReceiveChannel") -> int:
        return 0

    def get_routes_count(self, obj: "AlertReceiveChannel") -> int:
        return obj.channel_filters.count()

    def get_is_legacy(self, obj: "AlertReceiveChannel") -> bool:
        return has_legacy_prefix(obj.integration)

    def get_connected_escalations_chains_count(self, obj: "AlertReceiveChannel") -> int:
        return len(
            set(
                channel_filter.escalation_chain_id
                for channel_filter in obj.channel_filters.all()
                if channel_filter.escalation_chain_id is not None
            )
        )


class AlertReceiveChannelCreateSerializer(AlertReceiveChannelSerializer):
    create_default_webhooks = serializers.BooleanField(required=False, default=True)

    class Meta(AlertReceiveChannelSerializer.Meta):
        fields = [
            *AlertReceiveChannelSerializer.Meta.fields,
            "create_default_webhooks",
        ]


class AlertReceiveChannelUpdateSerializer(AlertReceiveChannelSerializer):
    class Meta(AlertReceiveChannelSerializer.Meta):
        read_only_fields = [*AlertReceiveChannelSerializer.Meta.read_only_fields, "integration"]


class FastAlertReceiveChannelSerializer(serializers.ModelSerializer[AlertReceiveChannel]):
    id = serializers.CharField(read_only=True, source="public_primary_key")
    integration = serializers.CharField(read_only=True)
    deleted = serializers.SerializerMethodField()

    class Meta:
        model = AlertReceiveChannel
        fields = ["id", "integration", "verbal_name", "deleted"]

    def get_deleted(self, obj: "AlertReceiveChannel") -> bool:
        return obj.deleted_at is not None


class FilterAlertReceiveChannelSerializer(serializers.ModelSerializer[AlertReceiveChannel]):
    # don't use get_value as the method name, otherwise this will override the get_value method on
    # serializers.ModelSerializer, which may cause unexpected behavior (+ this violates the "Lisov substition
    # principle" which mypy complains about)
    value = serializers.SerializerMethodField(method_name="_get_value")
    display_name = serializers.SerializerMethodField()

    class Meta:
        model = AlertReceiveChannel
        fields = ["value", "display_name", "integration_url"]

    def _get_value(self, obj: "AlertReceiveChannel") -> str:
        return obj.public_primary_key

    def get_display_name(self, obj: "AlertReceiveChannel") -> str:
        display_name = obj.verbal_name or AlertReceiveChannel.INTEGRATION_CHOICES[obj.integration][1]
        return display_name


class AlertReceiveChannelTemplatesSerializer(EagerLoadingMixin, serializers.ModelSerializer[AlertReceiveChannel]):
    id = serializers.CharField(read_only=True, source="public_primary_key")

    payload_example = SerializerMethodField()
    is_based_on_alertmanager = SerializerMethodField()

    class Meta:
        model = AlertReceiveChannel
        fields = [
            "id",
            "verbal_name",
            "payload_example",
            "is_based_on_alertmanager",
        ]
        extra_kwargs = {"integration": {"required": True}}

    def get_payload_example(self, obj: "AlertReceiveChannel"):
        from apps.alerts.models import AlertGroup

        if "alert_group_id" in self.context["request"].query_params:
            alert_group_id = self.context["request"].query_params.get("alert_group_id")
            try:
                return obj.alert_groups.get(public_primary_key=alert_group_id).alerts.first().raw_request_data
            except AlertGroup.DoesNotExist:
                raise serializers.ValidationError("Alert group doesn't exist for this integration")
            except AttributeError:
                raise serializers.ValidationError("Unable to retrieve example payload for this alert group")
        else:
            try:
                return obj.alert_groups.only("id").last().alerts.first().raw_request_data
            except AttributeError:
                return None

    def get_is_based_on_alertmanager(self, obj: "AlertReceiveChannel"):
        return obj.based_on_alertmanager

    # Override method to pass field_name directly in set_value to handle None values for WritableSerializerField
    def to_internal_value(self, data):
        """
        Dict of native values <- Dict of primitive datatypes.
        """
        # First validate and save data from serializer fields
        ret = super().to_internal_value(data)

        # Separately validate and save template fields we generate dynamically
        errors = OrderedDict()

        # handle updates for core templates
        core_template_errors = self._handle_core_template_updates(data, ret)
        errors.update(core_template_errors)

        # handle updates for messaging backend templates
        messaging_backend_errors = self._handle_messaging_backend_updates(data, ret)
        errors.update(messaging_backend_errors)

        if errors:
            raise ValidationError(errors)
        return ret

    def _handle_messaging_backend_updates(self, data, ret):
        """Update additional messaging backend templates if needed."""
        errors = {}
        for backend_id, backend in get_messaging_backends():
            if not backend.customizable_templates:
                continue
            # fetch existing templates if any
            backend_templates = {}
            if self.instance.messaging_backends_templates is not None:
                backend_templates = self.instance.messaging_backends_templates.get(backend_id, {})
            # validate updated templates if any
            backend_updates = {}
            for field in APPEARANCE_TEMPLATE_NAMES:
                field_name = f"{backend.slug}_{field}_template"
                value = data.get(field_name)
                validator = jinja_template_env.from_string
                if value is not None:
                    try:
                        if value:
                            validator(value)
                    except TemplateSyntaxError:
                        errors[field_name] = "invalid template"
                    except DjangoValidationError:
                        errors[field_name] = "invalid URL"
                    else:
                        backend_updates[field] = value
            # update backend templates
            backend_templates.update(backend_updates)
            self.set_value(ret, ["messaging_backends_templates", backend_id], backend_templates)

        return errors

    def _handle_core_template_updates(self, data, ret):
        """Update core templates if needed."""
        errors = {}

        for field_name in self.core_templates_names:
            value = data.get(field_name)
            validator = jinja_template_env.from_string
            if value is not None:
                try:
                    if value:
                        validator(value)
                except TemplateSyntaxError:
                    errors[field_name] = "invalid template"
                except DjangoValidationError:
                    errors[field_name] = "invalid URL"
                self.set_value(ret, [field_name], value)
        return errors

    def to_representation(self, obj: "AlertReceiveChannel"):
        ret = super().to_representation(obj)

        core_templates = self._get_core_templates(obj)
        ret.update(core_templates)

        # include messaging backend templates
        additional_templates = self._get_messaging_backend_templates(obj)
        ret.update(additional_templates)

        return ret

    def _get_messaging_backend_templates(self, obj: "AlertReceiveChannel"):
        """Return additional messaging backend templates if any."""
        templates = {}
        for backend_id, backend in get_messaging_backends():
            if not backend.customizable_templates:
                continue
            for field in backend.template_fields:
                value = None
                is_default = False
                if obj.messaging_backends_templates:
                    value = obj.messaging_backends_templates.get(backend_id, {}).get(field)
                if not value and not backend.skip_default_template_fields:
                    value = obj.get_default_template_attribute(backend_id, field)
                    is_default = True
                field_name = f"{backend.slug}_{field}_template"
                templates[field_name] = value
                templates[f"{field_name}_is_default"] = is_default
        return templates

    def _get_core_templates(self, obj: "AlertReceiveChannel"):
        core_templates = {}

        for template_name in self.core_templates_names:
            template_value = getattr(obj, template_name)
            defaults = getattr(obj, f"INTEGRATION_TO_DEFAULT_{template_name.upper()}", {})
            default_template_value = defaults.get(obj.integration)
            core_templates[template_name] = template_value or default_template_value
            core_templates[f"{template_name}_is_default"] = not bool(template_value)

        return core_templates

    @property
    def core_templates_names(self) -> typing.List[str]:
        """
        returns names of templates introduced before messaging backends system with respect to enabled integrations.
        """
        core_templates = [
            "web_title_template",
            "web_message_template",
            "web_image_url_template",
            "sms_title_template",
            "phone_call_title_template",
            "source_link_template",
            "grouping_id_template",
            "resolve_condition_template",
            "acknowledge_condition_template",
        ]

        if settings.FEATURE_SLACK_INTEGRATION_ENABLED:
            core_templates += [
                "slack_title_template",
                "slack_message_template",
                "slack_image_url_template",
            ]
        if settings.FEATURE_TELEGRAM_INTEGRATION_ENABLED:
            core_templates += [
                "telegram_title_template",
                "telegram_message_template",
                "telegram_image_url_template",
            ]
        return core_templates