116 lines
4 KiB
Python
116 lines
4 KiB
Python
from collections import defaultdict
|
|
|
|
from rest_framework import serializers
|
|
from rest_framework.validators import UniqueTogetherValidator
|
|
|
|
from apps.webhooks.models import Webhook, WebhookResponse
|
|
from common.api_helpers.custom_fields import TeamPrimaryKeyRelatedField
|
|
from common.api_helpers.utils import CurrentOrganizationDefault, CurrentTeamDefault, CurrentUserDefault
|
|
from common.jinja_templater import apply_jinja_template
|
|
from common.jinja_templater.apply_jinja_template import JinjaTemplateError, JinjaTemplateWarning
|
|
|
|
|
|
class WebhookResponseSerializer(serializers.ModelSerializer):
|
|
class Meta:
|
|
model = WebhookResponse
|
|
fields = [
|
|
"timestamp",
|
|
"url",
|
|
"request_trigger",
|
|
"request_headers",
|
|
"request_data",
|
|
"status_code",
|
|
"content",
|
|
]
|
|
|
|
|
|
class WebhookSerializer(serializers.ModelSerializer):
|
|
id = serializers.CharField(read_only=True, source="public_primary_key")
|
|
organization = serializers.HiddenField(default=CurrentOrganizationDefault())
|
|
team = TeamPrimaryKeyRelatedField(allow_null=True, default=CurrentTeamDefault())
|
|
user = serializers.HiddenField(default=CurrentUserDefault())
|
|
last_run = serializers.SerializerMethodField()
|
|
trigger_type = serializers.CharField(required=True)
|
|
forward_all = serializers.BooleanField(allow_null=True, required=False)
|
|
last_response_log = serializers.SerializerMethodField()
|
|
trigger_type_name = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = Webhook
|
|
fields = [
|
|
"id",
|
|
"name",
|
|
"team",
|
|
"data",
|
|
"user",
|
|
"username",
|
|
"password",
|
|
"authorization_header",
|
|
"organization",
|
|
"trigger_template",
|
|
"headers",
|
|
"url",
|
|
"data",
|
|
"forward_all",
|
|
"http_method",
|
|
"trigger_type",
|
|
"trigger_type_name",
|
|
"last_run",
|
|
"last_response_log",
|
|
]
|
|
extra_kwargs = {
|
|
"name": {"required": True, "allow_null": False, "allow_blank": False},
|
|
"url": {"required": True, "allow_null": False, "allow_blank": False},
|
|
}
|
|
|
|
validators = [UniqueTogetherValidator(queryset=Webhook.objects.all(), fields=["name", "organization"])]
|
|
|
|
def _validate_template_field(self, template):
|
|
try:
|
|
apply_jinja_template(template, alert_payload=defaultdict(str), alert_group_id="alert_group_1")
|
|
except JinjaTemplateError as e:
|
|
raise serializers.ValidationError(e.fallback_message)
|
|
except JinjaTemplateWarning:
|
|
# Suppress render exceptions since we do not have a representative payload to test with
|
|
pass
|
|
return template
|
|
|
|
def validate_trigger_template(self, trigger_template):
|
|
if not trigger_template:
|
|
return None
|
|
return self._validate_template_field(trigger_template)
|
|
|
|
def validate_headers(self, headers):
|
|
if not headers:
|
|
return None
|
|
return self._validate_template_field(headers)
|
|
|
|
def validate_url(self, url):
|
|
if not url:
|
|
return None
|
|
return self._validate_template_field(url)
|
|
|
|
def validate_data(self, data):
|
|
if not data:
|
|
return None
|
|
return self._validate_template_field(data)
|
|
|
|
def validate_forward_all(self, data):
|
|
if data is None:
|
|
return False
|
|
return data
|
|
|
|
def get_last_run(self, obj):
|
|
last_run = ""
|
|
if last_log := obj.responses.all().last():
|
|
last_run = last_log.timestamp.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
return last_run
|
|
|
|
def get_last_response_log(self, obj):
|
|
return WebhookResponseSerializer(obj.responses.all().last()).data
|
|
|
|
def get_trigger_type_name(self, obj):
|
|
trigger_type_name = ""
|
|
if obj.trigger_type is not None:
|
|
trigger_type_name = Webhook.TRIGGER_TYPES[int(obj.trigger_type)][1]
|
|
return trigger_type_name
|