152d5f74fc
## Which issue(s) this PR closes https://www.loom.com/share/e1858db48e8b4fa99014a997af5e3d5e Closes https://github.com/grafana/irm/issues/322 ## Checklist - [x] Unit, integration, and e2e (if applicable) tests updated - [x] Documentation added (or `pr:no public docs` PR label added if not required) - [x] Added the relevant release notes label (see labels prefixed w/ `release:`). These labels dictate how your PR will show up in the autogenerated release notes. --------- Co-authored-by: Vadim Stepanov <vadimkerr@gmail.com>
82 lines
3.1 KiB
Python
82 lines
3.1 KiB
Python
import typing
|
|
|
|
from django.core.exceptions import ObjectDoesNotExist
|
|
from rest_framework import serializers
|
|
|
|
from apps.alerts.models import AlertGroup
|
|
from apps.user_management.models import Organization
|
|
from common.api_helpers.custom_fields import TeamPrimaryKeyRelatedField
|
|
from common.api_helpers.utils import CurrentTeamDefault
|
|
|
|
|
|
class SerializerContext(typing.TypedDict):
|
|
organization: Organization
|
|
|
|
|
|
class UserReferenceSerializer(serializers.Serializer):
|
|
context: SerializerContext
|
|
|
|
id = serializers.CharField()
|
|
important = serializers.BooleanField()
|
|
instance = serializers.HiddenField(default=None) # set in UserReferenceSerializer.validate
|
|
|
|
def validate(self, attrs):
|
|
id = attrs["id"]
|
|
organization = self.context["organization"]
|
|
|
|
try:
|
|
attrs["instance"] = organization.users.get(public_primary_key=id)
|
|
except ObjectDoesNotExist:
|
|
raise serializers.ValidationError(f"User {id} does not exist")
|
|
|
|
return attrs
|
|
|
|
|
|
class BasePagingSerializer(serializers.Serializer):
|
|
context: SerializerContext
|
|
|
|
ALLOWS_GRAFANA_INCIDENT_ID = False
|
|
|
|
users = UserReferenceSerializer(many=True, required=False, default=list)
|
|
team = TeamPrimaryKeyRelatedField(allow_null=True, default=CurrentTeamDefault())
|
|
important_team_escalation = serializers.BooleanField(required=False, default=False)
|
|
|
|
alert_group_id = serializers.CharField(required=False, default=None)
|
|
alert_group = serializers.HiddenField(default=None) # set in DirectPagingSerializer.validate
|
|
|
|
title = serializers.CharField(required=False, default=None)
|
|
message = serializers.CharField(required=False, default=None, allow_null=True)
|
|
source_url = serializers.URLField(required=False, default=None, allow_null=True)
|
|
|
|
def validate(self, attrs):
|
|
organization = self.context["organization"]
|
|
alert_group_id = attrs["alert_group_id"]
|
|
title = attrs["title"]
|
|
message = attrs["message"]
|
|
source_url = attrs["source_url"]
|
|
grafana_incident_id = self.ALLOWS_GRAFANA_INCIDENT_ID and attrs.get("grafana_incident_id")
|
|
|
|
if alert_group_id and (title or message or source_url or grafana_incident_id):
|
|
raise serializers.ValidationError(
|
|
f"alert_group_id and (title, message, source_url{', grafana_incident_id' if self.ALLOWS_GRAFANA_INCIDENT_ID else ''}) "
|
|
"are mutually exclusive"
|
|
)
|
|
|
|
if attrs["users"] and attrs["team"]:
|
|
raise serializers.ValidationError("users and team are mutually exclusive")
|
|
|
|
if alert_group_id:
|
|
try:
|
|
attrs["alert_group"] = AlertGroup.objects.get(
|
|
public_primary_key=alert_group_id, channel__organization=organization
|
|
)
|
|
except ObjectDoesNotExist:
|
|
raise serializers.ValidationError("Alert group {} does not exist".format(alert_group_id))
|
|
|
|
return attrs
|
|
|
|
|
|
class DirectPagingSerializer(BasePagingSerializer):
|
|
ALLOWS_GRAFANA_INCIDENT_ID = True
|
|
|
|
grafana_incident_id = serializers.CharField(required=False, default=None, allow_null=True)
|