132bdf235b
Related to https://github.com/grafana/oncall-private/issues/2826 RBAC enabled or not (OSS or cloud), it is possible to get service account permissions, enabling perm check (for service account tokens) in public API. Also allow empty value for users in sync (instead of returning a 400 response).
116 lines
4.1 KiB
Python
116 lines
4.1 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.CharField(allow_blank=True)
|
|
avatar_url = serializers.CharField(allow_blank=True)
|
|
|
|
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()
|
|
irm_enabled = serializers.BooleanField(default=False)
|
|
|
|
def validate_grafana_url(self, value):
|
|
# remove trailing slash for URL consistency
|
|
return value.rstrip("/")
|
|
|
|
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(), allow_null=True, allow_empty=True)
|
|
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
|