64bf1e5096
# What this PR does Reduces number of calls to db for `/schedules`, `/alertgroups` and `/users` endpoints. Fixes the issue when there was an additional call to db to get organization url to build user avatar full link. ## 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.
367 lines
13 KiB
Python
367 lines
13 KiB
Python
import math
|
|
import time
|
|
import typing
|
|
|
|
from django.conf import settings
|
|
from rest_framework import serializers
|
|
|
|
from apps.api.serializers.telegram import TelegramToUserConnectorSerializer
|
|
from apps.base.messaging import get_messaging_backends
|
|
from apps.base.models import UserNotificationPolicy
|
|
from apps.base.utils import live_settings
|
|
from apps.oss_installation.constants import CloudSyncStatus
|
|
from apps.oss_installation.utils import cloud_user_identity_status
|
|
from apps.schedules.ical_utils import SchedulesOnCallUsers
|
|
from apps.user_management.models import User
|
|
from common.api_helpers.custom_fields import TeamPrimaryKeyRelatedField, TimeZoneField
|
|
from common.api_helpers.mixins import EagerLoadingMixin
|
|
from common.api_helpers.utils import check_phone_number_is_valid
|
|
|
|
from .custom_serializers import DynamicFieldsModelSerializer
|
|
from .organization import FastOrganizationSerializer
|
|
from .slack_user_identity import SlackUserIdentitySerializer
|
|
from .team import FastTeamSerializer
|
|
|
|
|
|
class UserSerializerContext(typing.TypedDict):
|
|
schedules_with_oncall_users: SchedulesOnCallUsers
|
|
|
|
|
|
class UserPermissionSerializer(serializers.Serializer):
|
|
action = serializers.CharField(read_only=True)
|
|
|
|
|
|
class GoogleCalendarSettingsSerializer(serializers.Serializer):
|
|
# # TODO: figure out how to get OrganizationFilteredPrimaryKeyRelatedField to work with many=True
|
|
# oncall_schedules_to_consider_for_shift_swaps =
|
|
# oncall_schedules_to_consider_for_shift_swaps = serializers.ListField(
|
|
# child=OrganizationFilteredPrimaryKeyRelatedField(
|
|
# queryset=OnCallSchedule.objects,
|
|
# required=False,
|
|
# allow_null=True,
|
|
# ),
|
|
# required=False,
|
|
# allow_null=True,
|
|
# )
|
|
oncall_schedules_to_consider_for_shift_swaps = serializers.ListField(
|
|
child=serializers.CharField(),
|
|
required=False,
|
|
allow_null=True,
|
|
)
|
|
|
|
|
|
class NotificationChainVerbal(typing.TypedDict):
|
|
default: str
|
|
important: str
|
|
|
|
|
|
class WorkingHoursPeriodSerializer(serializers.Serializer):
|
|
start = serializers.CharField()
|
|
end = serializers.CharField()
|
|
|
|
|
|
class WorkingHoursSerializer(serializers.Serializer):
|
|
monday = serializers.ListField(child=WorkingHoursPeriodSerializer())
|
|
tuesday = serializers.ListField(child=WorkingHoursPeriodSerializer())
|
|
wednesday = serializers.ListField(child=WorkingHoursPeriodSerializer())
|
|
thursday = serializers.ListField(child=WorkingHoursPeriodSerializer())
|
|
friday = serializers.ListField(child=WorkingHoursPeriodSerializer())
|
|
saturday = serializers.ListField(child=WorkingHoursPeriodSerializer())
|
|
sunday = serializers.ListField(child=WorkingHoursPeriodSerializer())
|
|
|
|
|
|
class ListUserSerializer(DynamicFieldsModelSerializer, EagerLoadingMixin):
|
|
pk = serializers.CharField(read_only=True, source="public_primary_key")
|
|
slack_user_identity = SlackUserIdentitySerializer(read_only=True)
|
|
|
|
telegram_configuration = TelegramToUserConnectorSerializer(source="telegram_connection", read_only=True)
|
|
|
|
messaging_backends = serializers.SerializerMethodField()
|
|
|
|
organization = FastOrganizationSerializer(read_only=True)
|
|
current_team = TeamPrimaryKeyRelatedField(allow_null=True, required=False)
|
|
|
|
timezone = TimeZoneField(allow_null=True, required=False)
|
|
avatar = serializers.URLField(source="avatar_url", read_only=True)
|
|
avatar_full = serializers.SerializerMethodField()
|
|
notification_chain_verbal = serializers.SerializerMethodField()
|
|
cloud_connection_status = serializers.SerializerMethodField()
|
|
working_hours = WorkingHoursSerializer(required=False)
|
|
|
|
SELECT_RELATED = [
|
|
"telegram_verification_code",
|
|
"telegram_connection",
|
|
"organization",
|
|
"slack_user_identity",
|
|
"mobileappauthtoken",
|
|
"google_oauth2_user",
|
|
]
|
|
PREFETCH_RELATED = ["notification_policies"]
|
|
|
|
class Meta:
|
|
model = User
|
|
fields = [
|
|
"pk",
|
|
"organization",
|
|
"current_team",
|
|
"email",
|
|
"username",
|
|
"name",
|
|
"role",
|
|
"avatar",
|
|
"avatar_full",
|
|
"timezone",
|
|
"working_hours",
|
|
"unverified_phone_number",
|
|
"verified_phone_number",
|
|
"slack_user_identity",
|
|
"telegram_configuration",
|
|
"messaging_backends",
|
|
"notification_chain_verbal",
|
|
"cloud_connection_status",
|
|
"hide_phone_number",
|
|
"has_google_oauth2_connected",
|
|
]
|
|
read_only_fields = [
|
|
"email",
|
|
"username",
|
|
"name",
|
|
"role",
|
|
"verified_phone_number",
|
|
"has_google_oauth2_connected",
|
|
]
|
|
|
|
def validate_working_hours(self, working_hours):
|
|
for day in working_hours:
|
|
for period in working_hours[day]:
|
|
try:
|
|
start = time.strptime(period["start"], "%H:%M:%S")
|
|
end = time.strptime(period["end"], "%H:%M:%S")
|
|
except ValueError:
|
|
raise serializers.ValidationError("'start' and 'end' fields must be in '%H:%M:%S' format")
|
|
|
|
if start >= end:
|
|
raise serializers.ValidationError("'start' must be less than 'end'")
|
|
return working_hours
|
|
|
|
def validate_unverified_phone_number(self, value):
|
|
if value:
|
|
if check_phone_number_is_valid(value):
|
|
return value
|
|
else:
|
|
raise serializers.ValidationError(
|
|
"Phone number must be entered in the format: '+999999999'. From 8 to 15 digits allowed."
|
|
)
|
|
else:
|
|
return None
|
|
|
|
def get_messaging_backends(self, obj: User) -> dict[str, dict]:
|
|
serialized_data = {}
|
|
supported_backends = get_messaging_backends()
|
|
for backend_id, backend in supported_backends:
|
|
serialized_data[backend_id] = backend.serialize_user(obj)
|
|
return serialized_data
|
|
|
|
def get_notification_chain_verbal(self, obj: User) -> NotificationChainVerbal:
|
|
default, important = UserNotificationPolicy.get_short_verbals_for_user(user=obj)
|
|
return {"default": " - ".join(default), "important": " - ".join(important)}
|
|
|
|
def get_avatar_full(self, obj):
|
|
organization = self.context["request"].auth.organization if self.context.get("request") else obj.organization
|
|
return obj.avatar_full_url(organization)
|
|
|
|
def get_cloud_connection_status(self, obj: User) -> CloudSyncStatus | None:
|
|
is_open_source_with_cloud_notifications = self.context.get("is_open_source_with_cloud_notifications", None)
|
|
is_open_source_with_cloud_notifications = (
|
|
is_open_source_with_cloud_notifications
|
|
if is_open_source_with_cloud_notifications is not None
|
|
else settings.IS_OPEN_SOURCE and live_settings.GRAFANA_CLOUD_NOTIFICATIONS_ENABLED
|
|
)
|
|
if is_open_source_with_cloud_notifications:
|
|
connector = self.context.get("connector", None)
|
|
identities = self.context.get("cloud_identities", {})
|
|
identity = identities.get(obj.email, None)
|
|
status, _ = cloud_user_identity_status(connector, identity)
|
|
return status
|
|
return None
|
|
|
|
def to_representation(self, instance):
|
|
result = super().to_representation(instance)
|
|
if instance.id != self.context["request"].user.id:
|
|
if instance.hide_phone_number:
|
|
if result["verified_phone_number"]:
|
|
result["verified_phone_number"] = self._hide_phone_number(result["verified_phone_number"])
|
|
if result["unverified_phone_number"]:
|
|
result["unverified_phone_number"] = self._hide_phone_number(result["unverified_phone_number"])
|
|
return result
|
|
|
|
@staticmethod
|
|
def _hide_phone_number(number: str):
|
|
HIDE_SYMBOL = "*"
|
|
SHOW_LAST_SYMBOLS = 4
|
|
if len(number) <= 4:
|
|
SHOW_LAST_SYMBOLS = math.ceil(len(number) / 2)
|
|
return f"{HIDE_SYMBOL * (len(number) - SHOW_LAST_SYMBOLS)}{number[-SHOW_LAST_SYMBOLS:]}"
|
|
|
|
|
|
class UserSerializer(ListUserSerializer):
|
|
context: UserSerializerContext
|
|
|
|
is_currently_oncall = serializers.SerializerMethodField()
|
|
google_calendar_settings = GoogleCalendarSettingsSerializer(required=False)
|
|
|
|
class Meta(ListUserSerializer.Meta):
|
|
fields = ListUserSerializer.Meta.fields + [
|
|
"is_currently_oncall",
|
|
"google_calendar_settings",
|
|
]
|
|
read_only_fields = ListUserSerializer.Meta.read_only_fields + [
|
|
"is_currently_oncall",
|
|
]
|
|
|
|
def get_is_currently_oncall(self, obj: User) -> bool:
|
|
# Serializer context is set here: apps.api.views.user.UserView.get_serializer_context.
|
|
return any(obj in users for users in self.context.get("schedules_with_oncall_users", {}).values())
|
|
|
|
|
|
class CurrentUserSerializer(UserSerializer):
|
|
rbac_permissions = UserPermissionSerializer(read_only=True, many=True, source="permissions")
|
|
|
|
class Meta(UserSerializer.Meta):
|
|
fields = UserSerializer.Meta.fields + [
|
|
"rbac_permissions",
|
|
"google_oauth2_token_is_missing_scopes",
|
|
]
|
|
read_only_fields = UserSerializer.Meta.read_only_fields + [
|
|
"google_oauth2_token_is_missing_scopes",
|
|
]
|
|
|
|
|
|
class UserHiddenFieldsSerializer(ListUserSerializer):
|
|
fields_available_for_all_users = [
|
|
"pk",
|
|
"organization",
|
|
"current_team",
|
|
"username",
|
|
"avatar",
|
|
"timezone",
|
|
"working_hours",
|
|
"notification_chain_verbal",
|
|
]
|
|
|
|
def to_representation(self, instance):
|
|
ret = super(ListUserSerializer, self).to_representation(instance)
|
|
if instance.id != self.context["request"].user.id:
|
|
for field in ret:
|
|
if field not in self.fields_available_for_all_users:
|
|
ret[field] = "******"
|
|
ret["hidden_fields"] = True
|
|
return ret
|
|
|
|
|
|
class ScheduleUserSerializer(ListUserSerializer):
|
|
fields_to_keep = [
|
|
"pk",
|
|
"organization",
|
|
"email",
|
|
"username",
|
|
"name",
|
|
"avatar",
|
|
"avatar_full",
|
|
"timezone",
|
|
"working_hours",
|
|
"slack_user_identity",
|
|
"telegram_configuration",
|
|
]
|
|
|
|
def to_representation(self, instance):
|
|
serialized = super(ListUserSerializer, self).to_representation(instance)
|
|
ret = {field: value for field, value in serialized.items() if field in self.fields_to_keep}
|
|
return ret
|
|
|
|
|
|
class FastUserSerializer(serializers.ModelSerializer):
|
|
pk = serializers.CharField(source="public_primary_key")
|
|
|
|
class Meta:
|
|
model = User
|
|
fields = [
|
|
"pk",
|
|
"username",
|
|
]
|
|
read_only_fields = [
|
|
"pk",
|
|
"username",
|
|
]
|
|
|
|
|
|
class FilterUserSerializer(EagerLoadingMixin, serializers.ModelSerializer):
|
|
value = serializers.CharField(source="public_primary_key")
|
|
display_name = serializers.CharField(source="username")
|
|
|
|
class Meta:
|
|
model = User
|
|
fields = ["value", "display_name"]
|
|
read_only_fields = [
|
|
"pk",
|
|
"username",
|
|
]
|
|
|
|
|
|
class UserShortSerializer(serializers.ModelSerializer):
|
|
username = serializers.CharField()
|
|
pk = serializers.CharField(source="public_primary_key")
|
|
avatar = serializers.CharField(source="avatar_url")
|
|
avatar_full = serializers.SerializerMethodField()
|
|
|
|
class Meta:
|
|
model = User
|
|
fields = [
|
|
"username",
|
|
"pk",
|
|
"avatar",
|
|
"avatar_full",
|
|
]
|
|
read_only_fields = [
|
|
"username",
|
|
"pk",
|
|
"avatar",
|
|
"avatar_full",
|
|
]
|
|
|
|
def get_avatar_full(self, obj):
|
|
organization = self.context["request"].auth.organization if self.context.get("request") else obj.organization
|
|
return obj.avatar_full_url(organization)
|
|
|
|
|
|
class UserIsCurrentlyOnCallSerializer(UserShortSerializer, EagerLoadingMixin):
|
|
context: UserSerializerContext
|
|
|
|
teams = FastTeamSerializer(read_only=True, many=True)
|
|
is_currently_oncall = serializers.SerializerMethodField()
|
|
|
|
SELECT_RELATED = ["organization"]
|
|
PREFETCH_RELATED = ["teams"]
|
|
|
|
class Meta(UserShortSerializer.Meta):
|
|
fields = UserShortSerializer.Meta.fields + [
|
|
"name",
|
|
"timezone",
|
|
"teams",
|
|
"is_currently_oncall",
|
|
]
|
|
|
|
def get_is_currently_oncall(self, obj: User) -> bool:
|
|
# Serializer context is set here: apps.api.views.user.UserView.get_serializer_context.
|
|
return any(obj in users for users in self.context.get("schedules_with_oncall_users", {}).values())
|
|
|
|
|
|
class PagedUserSerializer(serializers.Serializer):
|
|
class Meta:
|
|
fields = [
|
|
"username",
|
|
"pk",
|
|
"avatar",
|
|
"avatar_full",
|
|
"important",
|
|
]
|