06d19bf6e9
# What this PR does
New OnCall plugin initialization process
## 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: Michael Derynck <michael.derynck@grafana.com>
Co-authored-by: Matias Bordese <mbordese@gmail.com>
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()
|
|
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
|