962cc34432
# What this PR does - Compresses sync data being sent to engine - Minor fix to log messages when JSON parse errors occur ## Which issue(s) this PR closes Related to [issue link here] <!-- *Note*: If you want the issue to be auto-closed once the PR is merged, change "Related to" to "Closes" in the line above. If you have more than one GitHub issue that this PR closes, be sure to preface each issue link with a [closing keyword](https://docs.github.com/en/get-started/writing-on-github/working-with-advanced-formatting/using-keywords-in-issues-and-pull-requests#linking-a-pull-request-to-an-issue). This ensures that the issue(s) are auto-closed once the PR has been merged. --> ## 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.
111 lines
3.8 KiB
Python
111 lines
3.8 KiB
Python
from dataclasses import asdict
|
|
from typing import Dict, List
|
|
|
|
from rest_framework import serializers
|
|
|
|
from apps.grafana_plugin.sync_data import SyncData, SyncPermission, SyncSettings, SyncTeam, SyncUser
|
|
|
|
|
|
class SyncPermissionSerializer(serializers.Serializer):
|
|
action = serializers.CharField()
|
|
|
|
def create(self, validated_data):
|
|
return SyncPermission(**validated_data)
|
|
|
|
def to_representation(self, instance):
|
|
return asdict(instance)
|
|
|
|
|
|
class SyncUserSerializer(serializers.Serializer):
|
|
id = serializers.IntegerField()
|
|
name = serializers.CharField(allow_blank=True)
|
|
login = serializers.CharField()
|
|
email = serializers.CharField()
|
|
role = serializers.CharField()
|
|
avatar_url = serializers.CharField(allow_blank=True)
|
|
permissions = SyncPermissionSerializer(many=True, allow_empty=True, allow_null=True)
|
|
teams = serializers.ListField(child=serializers.IntegerField(), allow_empty=True, allow_null=True)
|
|
|
|
def create(self, validated_data):
|
|
return SyncUser(**validated_data)
|
|
|
|
def to_representation(self, instance):
|
|
return asdict(instance)
|
|
|
|
|
|
class SyncTeamSerializer(serializers.Serializer):
|
|
team_id = serializers.IntegerField()
|
|
name = serializers.CharField()
|
|
email = serializers.EmailField(allow_blank=True)
|
|
avatar_url = serializers.CharField()
|
|
|
|
def create(self, validated_data):
|
|
return SyncTeam(**validated_data)
|
|
|
|
def to_representation(self, instance):
|
|
return asdict(instance)
|
|
|
|
|
|
class TeamMemberMappingField(serializers.Field):
|
|
def to_representation(self, value: Dict[int, List[int]]):
|
|
return {str(k): v for k, v in value.items()}
|
|
|
|
def to_internal_value(self, data):
|
|
if not isinstance(data, dict):
|
|
raise serializers.ValidationError("Expected a dictionary")
|
|
try:
|
|
return {int(k): v for k, v in data.items()}
|
|
except ValueError:
|
|
raise serializers.ValidationError("All keys must be convertible to integers")
|
|
|
|
|
|
class SyncOnCallSettingsSerializer(serializers.Serializer):
|
|
stack_id = serializers.IntegerField()
|
|
org_id = serializers.IntegerField()
|
|
license = serializers.CharField()
|
|
oncall_api_url = serializers.CharField()
|
|
oncall_token = serializers.CharField(allow_blank=True)
|
|
grafana_url = serializers.CharField()
|
|
grafana_token = serializers.CharField()
|
|
rbac_enabled = serializers.BooleanField()
|
|
incident_enabled = serializers.BooleanField()
|
|
incident_backend_url = serializers.CharField(allow_blank=True)
|
|
labels_enabled = serializers.BooleanField()
|
|
|
|
def create(self, validated_data):
|
|
return SyncSettings(**validated_data)
|
|
|
|
def to_representation(self, instance):
|
|
return asdict(instance)
|
|
|
|
|
|
class SyncDataSerializer(serializers.Serializer):
|
|
users = serializers.ListField(child=SyncUserSerializer())
|
|
teams = serializers.ListField(child=SyncTeamSerializer(), allow_null=True, allow_empty=True)
|
|
team_members = TeamMemberMappingField()
|
|
settings = SyncOnCallSettingsSerializer()
|
|
|
|
def create(self, validated_data):
|
|
return SyncData(**validated_data)
|
|
|
|
def to_representation(self, instance):
|
|
return asdict(instance)
|
|
|
|
def to_internal_value(self, data):
|
|
data = super().to_internal_value(data)
|
|
users = data.get("users")
|
|
if users:
|
|
|
|
def create_user(user):
|
|
permissions_data = user.pop("permissions", [])
|
|
permissions = [SyncPermission(**perm) for perm in permissions_data] if permissions_data else []
|
|
return SyncUser(permissions=permissions, **user)
|
|
|
|
data["users"] = [create_user(user) for user in users]
|
|
teams = data.get("teams")
|
|
if teams:
|
|
data["teams"] = [SyncTeam(**team) for team in teams]
|
|
settings = data.get("settings")
|
|
if settings:
|
|
data["settings"] = SyncSettings(**settings)
|
|
return data
|