Improve OpenAPI schema coverage

engine/apps/alerts/models/alert_group.py

@@ -348,7 +348,7 @@ def is_silenced_for_period(self):
         return self.silenced and self.silenced_until is not None
 
     @property
-    def status(self):
+    def status(self) -> int:
         if self.resolved:
             return AlertGroup.RESOLVED
         elif self.acknowledged:

engine/apps/alerts/models/alert_receive_channel.py

@@ -412,15 +412,15 @@ def alerts_count(self):
         return Alert.objects.filter(group__channel=self).count()
 
     @property
-    def is_able_to_autoresolve(self):
+    def is_able_to_autoresolve(self) -> bool:
         return self.config.is_able_to_autoresolve
 
     @property
     def is_demo_alert_enabled(self):
         return self.config.is_demo_alert_enabled
 
     @property
-    def description(self):
+    def description(self) -> str | None:
         # TODO: AMV2: Remove this check after legacy integrations are migrated.
         if self.integration == AlertReceiveChannel.INTEGRATION_LEGACY_GRAFANA_ALERTING:
             contact_points = self.contact_points.all()
@@ -496,7 +496,7 @@ def web_link(self):
         return urljoin(self.organization.web_link, f"integrations/{self.public_primary_key}")
 
     @property
-    def integration_url(self):
+    def integration_url(self) -> str | None:
         if self.integration in [
             AlertReceiveChannel.INTEGRATION_MANUAL,
             AlertReceiveChannel.INTEGRATION_SLACK_CHANNEL,
@@ -595,7 +595,7 @@ def notify_about_maintenance_action(self, text, send_to_general_log_channel=True
 
     # Heartbeat
     @property
-    def is_available_for_integration_heartbeat(self):
+    def is_available_for_integration_heartbeat(self) -> bool:
         return self.heartbeat_module is not None
 
     @property

engine/apps/alerts/models/maintainable_object.py

@@ -151,7 +151,7 @@ def start_maintenance(self, mode, maintenance_duration, user):
             )
 
     @property
-    def till_maintenance_timestamp(self):
+    def till_maintenance_timestamp(self) -> int | None:
         if self.maintenance_started_at is not None and self.maintenance_duration is not None:
             return int((self.maintenance_started_at + self.maintenance_duration).astimezone(pytz.UTC).timestamp())
         return None

engine/apps/api/serializers/alert.py

@@ -1,3 +1,5 @@
+import typing
+
 from django.core.cache import cache
 from django.utils import timezone
 from rest_framework import serializers
@@ -8,6 +10,13 @@
 from .alerts_field_cache_buster_mixin import AlertsFieldCacheBusterMixin
 
 
+class RenderForWeb(typing.TypedDict):
+    title: str
+    message: str
+    image_url: str | None
+    source_link: str | None
+
+
 class AlertFieldsCacheSerializerMixin(AlertsFieldCacheBusterMixin):
     CACHE_KEY_FORMAT_TEMPLATE = "{field_name}_alert_{object_id}"
 
@@ -51,7 +60,7 @@ class Meta:
             "created_at",
         ]
 
-    def get_render_for_web(self, obj):
+    def get_render_for_web(self, obj) -> RenderForWeb:
         return AlertFieldsCacheSerializerMixin.get_or_set_web_template_field(
             obj,
             AlertFieldsCacheSerializerMixin.RENDER_FOR_WEB_FIELD_NAME,

engine/apps/api/serializers/alert_group.py

@@ -4,7 +4,7 @@
 
 from django.core.cache import cache
 from django.utils import timezone
-from drf_spectacular.utils import extend_schema_field, inline_serializer
+from drf_spectacular.utils import extend_schema_field
 from rest_framework import serializers
 
 from apps.alerts.incident_appearance.renderers.web_renderer import AlertGroupWebRenderer
@@ -22,6 +22,17 @@
 logger.setLevel(logging.DEBUG)
 
 
+class RenderForWeb(typing.TypedDict):
+    title: str
+    message: str
+    image_url: str | None
+    source_link: str | None
+
+
+class EmptyRenderForWeb(typing.TypedDict):
+    pass
+
+
 class AlertGroupFieldsCacheSerializerMixin(AlertsFieldCacheBusterMixin):
     CACHE_KEY_FORMAT_TEMPLATE = "{field_name}_alert_group_{object_id}"
 
@@ -80,18 +91,7 @@ class Meta:
         fields = ["pk", "render_for_web", "alert_receive_channel", "inside_organization_number"]
         read_only_fields = ["pk", "render_for_web", "alert_receive_channel", "inside_organization_number"]
 
-    @extend_schema_field(
-        inline_serializer(
-            name="render_for_web",
-            fields={
-                "title": serializers.CharField(),
-                "message": serializers.CharField(),
-                "image_url": serializers.CharField(),
-                "source_link": serializers.CharField(),
-            },
-        )
-    )
-    def get_render_for_web(self, obj: "AlertGroup"):
+    def get_render_for_web(self, obj: "AlertGroup") -> RenderForWeb | EmptyRenderForWeb:
         last_alert = obj.alerts.last()
         if last_alert is None:
             return {}
@@ -170,18 +170,7 @@ class Meta:
             "labels",
         ]
 
-    @extend_schema_field(
-        inline_serializer(
-            name="render_for_web",
-            fields={
-                "title": serializers.CharField(),
-                "message": serializers.CharField(),
-                "image_url": serializers.CharField(),
-                "source_link": serializers.CharField(),
-            },
-        )
-    )
-    def get_render_for_web(self, obj: "AlertGroup"):
+    def get_render_for_web(self, obj: "AlertGroup") -> RenderForWeb | EmptyRenderForWeb:
         if not obj.last_alert:
             return {}
         return AlertGroupFieldsCacheSerializerMixin.get_or_set_web_template_field(

engine/apps/api/serializers/alert_receive_channel.py

@@ -2,7 +2,6 @@
 from collections import OrderedDict
 
 from django.conf import settings
-from django.core.exceptions import ObjectDoesNotExist
 from django.core.exceptions import ValidationError as DjangoValidationError
 from django.db.models import Q
 from jinja2 import TemplateSyntaxError
@@ -53,17 +52,17 @@ class IntegrationAlertGroupLabels(typing.TypedDict):
 class CustomLabelSerializer(serializers.Serializer):
     """This serializer is consistent with apps.api.serializers.labels.LabelSerializer, but allows null for value ID."""
 
-    class KeySerializer(serializers.Serializer):
+    class CustomLabelKeySerializer(serializers.Serializer):
         id = serializers.CharField()
         name = serializers.CharField()
 
-    class ValueSerializer(serializers.Serializer):
+    class CustomLabelValueSerializer(serializers.Serializer):
         # ID is null for templated labels. For such labels, the "name" value is a Jinja2 template.
         id = serializers.CharField(allow_null=True)
         name = serializers.CharField()
 
-    key = KeySerializer()
-    value = ValueSerializer()
+    key = CustomLabelKeySerializer()
+    value = CustomLabelValueSerializer()
 
 
 class IntegrationAlertGroupLabelsSerializer(serializers.Serializer):
@@ -215,9 +214,9 @@ class AlertReceiveChannelSerializer(
     default_channel_filter = serializers.SerializerMethodField()
     instructions = serializers.SerializerMethodField()
     demo_alert_enabled = serializers.BooleanField(source="is_demo_alert_enabled", read_only=True)
-    is_based_on_alertmanager = serializers.BooleanField(source="has_alertmanager_payload_structure", read_only=True)
+    is_based_on_alertmanager = serializers.BooleanField(source="based_on_alertmanager", read_only=True)
     maintenance_till = serializers.ReadOnlyField(source="till_maintenance_timestamp")
-    heartbeat = serializers.SerializerMethodField()
+    heartbeat = IntegrationHeartBeatSerializer(read_only=True, allow_null=True, source="integration_heartbeat")
     allow_delete = serializers.SerializerMethodField()
     description_short = serializers.CharField(max_length=250, required=False, allow_null=True)
     demo_alert_payload = serializers.JSONField(source="config.example_payload", read_only=True)
@@ -334,15 +333,16 @@ def update(self, instance, validated_data):
         except AlertReceiveChannel.DuplicateDirectPagingError:
             raise BadRequest(detail=AlertReceiveChannel.DuplicateDirectPagingError.DETAIL)
 
-    def get_instructions(self, obj: "AlertReceiveChannel"):
+    def get_instructions(self, obj: "AlertReceiveChannel") -> str:
         # Deprecated, kept for api-backward compatibility
         return ""
 
     # MethodFields are used instead of relevant properties because of properties hit db on each instance in queryset
-    def get_default_channel_filter(self, obj: "AlertReceiveChannel"):
+    def get_default_channel_filter(self, obj: "AlertReceiveChannel") -> str | None:
         for filter in obj.channel_filters.all():
             if filter.is_default:
                 return filter.public_primary_key
+        return None
 
     @staticmethod
     def validate_integration(integration):
@@ -367,21 +367,14 @@ def validate_verbal_name(self, verbal_name):
         else:
             raise serializers.ValidationError(detail="Integration with this name already exists")
 
-    def get_heartbeat(self, obj: "AlertReceiveChannel"):
-        try:
-            heartbeat = obj.integration_heartbeat
-        except ObjectDoesNotExist:
-            return None
-        return IntegrationHeartBeatSerializer(heartbeat).data
-
-    def get_allow_delete(self, obj: "AlertReceiveChannel"):
+    def get_allow_delete(self, obj: "AlertReceiveChannel") -> bool:
         # don't allow deleting direct paging integrations
         return obj.integration != AlertReceiveChannel.INTEGRATION_DIRECT_PAGING
 
-    def get_alert_count(self, obj: "AlertReceiveChannel"):
+    def get_alert_count(self, obj: "AlertReceiveChannel") -> int:
         return 0
 
-    def get_alert_groups_count(self, obj: "AlertReceiveChannel"):
+    def get_alert_groups_count(self, obj: "AlertReceiveChannel") -> int:
         return 0
 
     def get_routes_count(self, obj: "AlertReceiveChannel") -> int:
@@ -428,10 +421,10 @@ class Meta:
         model = AlertReceiveChannel
         fields = ["value", "display_name", "integration_url"]
 
-    def _get_value(self, obj: "AlertReceiveChannel"):
+    def _get_value(self, obj: "AlertReceiveChannel") -> str:
         return obj.public_primary_key
 
-    def get_display_name(self, obj: "AlertReceiveChannel"):
+    def get_display_name(self, obj: "AlertReceiveChannel") -> str:
         display_name = obj.verbal_name or AlertReceiveChannel.INTEGRATION_CHOICES[obj.integration][1]
         return display_name
 

engine/apps/api/serializers/integration_heartbeat.py

@@ -41,10 +41,10 @@ def validate_alert_receive_channel(self, alert_receive_channel):
                 {"alert_receive_channel": "Heartbeat is not available for this integration"}
             )
 
-    def get_last_heartbeat_time_verbal(self, obj):
+    def get_last_heartbeat_time_verbal(self, obj) -> str | None:
         return self._last_heartbeat_time_verbal(obj) if obj.last_heartbeat_time else None
 
-    def get_instruction(self, obj):
+    def get_instruction(self, obj) -> str:
         # Deprecated. Kept for API backward compatibility.
         return ""
 

engine/apps/api/serializers/slack_user_identity.py

@@ -14,5 +14,5 @@ class Meta:
         fields = ["slack_login", "slack_id", "avatar", "name", "display_name"]
         read_only_fields = ["slack_login", "slack_id", "avatar", "name", "display_name"]
 
-    def get_display_name(self, obj):
+    def get_display_name(self, obj) -> str | None:
         return obj.profile_display_name or obj.slack_verbal

engine/apps/api/serializers/user.py

@@ -3,12 +3,14 @@
 import typing
 
 from django.conf import settings
+from drf_spectacular.utils import extend_schema_field
 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
@@ -31,6 +33,31 @@ class UserPermissionSerializer(serializers.Serializer):
     action = serializers.CharField(read_only=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())
+
+
+@extend_schema_field(WorkingHoursSerializer)
+class WorkingHoursField(serializers.JSONField):
+    pass
+
+
 class UserSerializer(DynamicFieldsModelSerializer, EagerLoadingMixin):
     pk = serializers.CharField(read_only=True, source="public_primary_key")
     slack_user_identity = SlackUserIdentitySerializer(read_only=True)
@@ -47,6 +74,7 @@ class UserSerializer(DynamicFieldsModelSerializer, EagerLoadingMixin):
     avatar_full = serializers.URLField(source="avatar_full_url", read_only=True)
     notification_chain_verbal = serializers.SerializerMethodField()
     cloud_connection_status = serializers.SerializerMethodField()
+    working_hours = WorkingHoursField(required=False)
 
     SELECT_RELATED = ["telegram_verification_code", "telegram_connection", "organization", "slack_user_identity"]
 
@@ -127,18 +155,18 @@ def validate_unverified_phone_number(self, value):
         else:
             return None
 
-    def get_messaging_backends(self, obj: User):
+    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):
+    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_cloud_connection_status(self, obj: User):
+    def get_cloud_connection_status(self, obj: User) -> CloudSyncStatus | None:
         if settings.IS_OPEN_SOURCE and live_settings.GRAFANA_CLOUD_NOTIFICATIONS_ENABLED:
             connector = self.context.get("connector", None)
             identities = self.context.get("cloud_identities", {})