centralcloud/oncall-engine
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Re-enable a few mypy rules + fix existing errors (#2725)

Browse source 
# What this PR does

Related to https://github.com/grafana/oncall/issues/2392

- Re-enable the following `mypy` rules + fix their pre-existing errors:
  - `no-redef`
  - `valid-type`
  - `var-annotated`
- Add stronger return typing to the `GrafanaAPIClient` by use of
generics + add some links to documentation in the method docstrings

## 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] `CHANGELOG.md` updated (or `pr:no changelog` PR label added if not
required)
This commit is contained in:
Joey Orlando 2023-08-03 11:43:03 +02:00 • committed by GitHub
parent d7e2f7053d
commit d6140cbe8d
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
44 changed files with 380 additions and 372 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
engine/apps/alerts/models/alert.py
View file

@ -8,9 +8,9 @@ from django.core.validators import MinLengthValidator
from django.db import models
from django.db.models import JSONField
from apps.alerts import tasks
from apps.alerts.constants import TASK_DELAY_SECONDS
from apps.alerts.incident_appearance.templaters import TemplateLoader
from apps.alerts.tasks import distribute_alert, send_alert_group_signal
from common.jinja_templater import apply_jinja_template
from common.jinja_templater.apply_jinja_template import JinjaTemplateError, JinjaTemplateWarning
from common.public_primary_keys import generate_public_primary_key, increase_public_primary_key_length
@ -139,9 +139,9 @@ class Alert(models.Model):
group.save(update_fields=["resolved_by_alert"])
if settings.DEBUG:
distribute_alert(alert.pk)
tasks.distribute_alert(alert.pk)
else:
distribute_alert.apply_async((alert.pk,), countdown=TASK_DELAY_SECONDS)
tasks.distribute_alert.apply_async((alert.pk,), countdown=TASK_DELAY_SECONDS)
if group_created:
# all code below related to maintenance mode
@ -163,7 +163,7 @@ class Alert(models.Model):
f"log record {log_record_for_root_incident.pk} with type "
f"'{log_record_for_root_incident.get_type_display()}'"
)
send_alert_group_signal.apply_async((log_record_for_root_incident.pk,))
tasks.send_alert_group_signal.apply_async((log_record_for_root_incident.pk,))
except AlertGroup.DoesNotExist:
pass

4
engine/apps/alerts/models/alert_group_log_record.py
View file

@ -9,7 +9,7 @@ from django.db.models.signals import post_save
from django.dispatch import receiver
from rest_framework.fields import DateTimeField
from apps.alerts.tasks import send_update_log_report_signal
from apps.alerts import tasks
from apps.alerts.utils import render_relative_timeline
from apps.slack.slack_formatter import SlackFormatter
from common.utils import clean_markup
@ -587,4 +587,4 @@ def listen_for_alertgrouplogrecord(sender, instance, created, *args, **kwargs):
f"send_update_log_report_signal for alert_group {alert_group_pk}, "
f"alert group event: {instance.get_type_display()}"
)
send_update_log_report_signal.apply_async(kwargs={"alert_group_pk": alert_group_pk}, countdown=8)
tasks.send_update_log_report_signal.apply_async(kwargs={"alert_group_pk": alert_group_pk}, countdown=8)

8
engine/apps/alerts/models/invitation.py
View file

@ -3,7 +3,7 @@ import logging
from django.db import models, transaction
from apps.alerts.tasks import invite_user_to_join_incident, send_alert_group_signal
from apps.alerts import tasks
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)
@ -91,9 +91,9 @@ class Invitation(models.Model):
f"call send_alert_group_signal for alert_group {alert_group.pk}, "
f"log record {log_record.pk} with type '{log_record.get_type_display()}'"
)
send_alert_group_signal.apply_async((log_record.pk,))
invite_user_to_join_incident.apply_async((invitation.pk,))
tasks.send_alert_group_signal.apply_async((log_record.pk,))
tasks.invite_user_to_join_incident.apply_async((invitation.pk,))
@staticmethod
def stop_invitation(invitation_pk, user):
@ -119,4 +119,4 @@ class Invitation(models.Model):
f"call send_alert_group_signal for alert_group {invitation.alert_group.pk}, "
f"log record {log_record.pk} with type '{log_record.get_type_display()}'"
)
send_alert_group_signal.apply_async((log_record.pk,))
tasks.send_alert_group_signal.apply_async((log_record.pk,))

32
engine/apps/alerts/paging.py
View file

@ -56,6 +56,18 @@ class DirectPagingAlertGroupResolvedError(Exception):
DETAIL = "Cannot add responders for a resolved alert group" # Returned in BadRequest responses and Slack warnings
class _OnCall(typing.TypedDict):
title: str
message: str
uid: str
author_username: str
permalink: str
class DirectPagingAlertPayload(typing.TypedDict):
oncall: _OnCall
def _trigger_alert(
organization: Organization,
team: Team | None,
@ -98,15 +110,17 @@ def _trigger_alert(
if not title:
title = "Message from {}".format(from_user.username)
payload = {}
# Custom oncall property in payload to simplify rendering
payload["oncall"] = {}
payload["oncall"]["title"] = title
payload["oncall"]["message"] = message
# avoid grouping
payload["oncall"]["uid"] = str(uuid4())
payload["oncall"]["author_username"] = from_user.username
payload["oncall"]["permalink"] = permalink
payload: DirectPagingAlertPayload = {
# Custom oncall property in payload to simplify rendering
"oncall": {
"title": title,
"message": message,
"uid": str(uuid4()), # avoid grouping
"author_username": from_user.username,
"permalink": permalink,
},
}
alert = Alert.create(
title=title,
message=message,

4
engine/apps/api/serializers/channel_filter.py
View file

@ -1,3 +1,5 @@
import typing
from rest_framework import serializers
from apps.alerts.models import AlertReceiveChannel, ChannelFilter, EscalationChain
@ -81,7 +83,7 @@ class ChannelFilterSerializer(EagerLoadingMixin, serializers.ModelSerializer):
"id": obj.slack_channel_pk,
}
def get_telegram_channel_details(self, obj) -> dict[str, any] | None:
def get_telegram_channel_details(self, obj) -> dict[str, typing.Any] | None:
if obj.telegram_channel_id is None:
return None
try:

41
engine/apps/grafana_plugin/helpers/client.py
View file

@ -50,7 +50,7 @@ class GCOMInstanceInfo(typing.TypedDict):
url: str
status: str
clusterSlug: str
config: GCOMInstanceInfoConfig | None
config: typing.NotRequired[GCOMInstanceInfoConfig]
class ApiClientResponseCallStatus(typing.TypedDict):
@ -60,10 +60,11 @@ class ApiClientResponseCallStatus(typing.TypedDict):
message: str
# TODO: come back and make the typing.Dict strongly typed once we switch to Python 3.12
# which has better support for generics
_APIClientResponse = typing.Optional[typing.Dict | typing.List]
APIClientResponse = typing.Tuple[_APIClientResponse, ApiClientResponseCallStatus]
_RT = typing.TypeVar("_RT")
class APIClientResponse(typing.Generic[_RT], typing.Tuple[typing.Optional[_RT], ApiClientResponseCallStatus]):
pass
# can't define this using class syntax because one of the keys contains a dash
@ -96,18 +97,18 @@ class APIClient:
self.api_url = api_url
self.api_token = api_token
def api_head(self, endpoint: str, body: typing.Optional[typing.Dict] = None, **kwargs) -> APIClientResponse:
def api_head(self, endpoint: str, body: typing.Optional[typing.Dict] = None, **kwargs) -> APIClientResponse[_RT]:
return self.call_api(endpoint, requests.head, body, **kwargs)
def api_get(self, endpoint: str, **kwargs) -> APIClientResponse:
def api_get(self, endpoint: str, **kwargs) -> APIClientResponse[_RT]:
return self.call_api(endpoint, requests.get, **kwargs)
def api_post(self, endpoint: str, body: typing.Optional[typing.Dict] = None, **kwargs) -> APIClientResponse:
def api_post(self, endpoint: str, body: typing.Optional[typing.Dict] = None, **kwargs) -> APIClientResponse[_RT]:
return self.call_api(endpoint, requests.post, body, **kwargs)
def call_api(
self, endpoint: str, http_method: HttpMethod, body: typing.Optional[typing.Dict] = None, **kwargs
) -> APIClientResponse:
) -> APIClientResponse[_RT]:
request_start = time.perf_counter()
call_status: ApiClientResponseCallStatus = {
"url": urljoin(self.api_url, endpoint),
@ -158,6 +159,23 @@ class APIClient:
class GrafanaAPIClient(APIClient):
USER_PERMISSION_ENDPOINT = f"api/access-control/users/permissions/search?actionPrefix={ACTION_PREFIX}"
class Types:
class _BaseGrafanaAPIResponse(typing.TypedDict):
totalCount: int
page: int
perPage: int
class GrafanaTeam(typing.TypedDict):
id: int
orgId: int
name: str
email: str
avatarUrl: str
memberCount: int
class TeamsResponse(_BaseGrafanaAPIResponse):
teams: typing.List["GrafanaAPIClient.Types.GrafanaTeam"]
def __init__(self, api_url: str, api_token: str) -> None:
super().__init__(api_url, api_token)
@ -219,7 +237,10 @@ class GrafanaAPIClient(APIClient):
user["permissions"] = user_permissions.get(str(user["userId"]), [])
return users
def get_teams(self, **kwargs) -> APIClientResponse:
def get_teams(self, **kwargs) -> APIClientResponse["GrafanaAPIClient.Types.TeamsResponse"]:
"""
[Grafana API Docs](https://grafana.com/docs/grafana/latest/developers/http_api/team/#team-search-with-paging)
"""
return self.api_get("api/teams/search?perpage=1000000", **kwargs)
def get_team_members(self, team_id: int) -> APIClientResponse:

20
engine/apps/metrics_exporter/helpers.py
View file

@ -53,17 +53,19 @@ def is_allowed_to_start_metrics_calculation(organization_id, force=False) -> boo
"""Check if metrics_cache_timer doesn't exist or if recalculation was started by force."""
recalculate_timeout = get_metrics_recalculation_timeout()
metrics_cache_timer_key = get_metrics_cache_timer_key(organization_id)
metrics_cache_timer = cache.get(metrics_cache_timer_key)
if metrics_cache_timer:
if not force or metrics_cache_timer.get("forced_started", False):
return False
else:
metrics_cache_timer["forced_started"] = True
else:
metrics_cache_timer: RecalculateMetricsTimer = {
metrics_cache_timer: RecalculateMetricsTimer = cache.get(
metrics_cache_timer_key,
{
"recalculate_timeout": recalculate_timeout,
"forced_started": force,
}
},
)
if not force or metrics_cache_timer.get("forced_started", False):
return False
else:
metrics_cache_timer["forced_started"] = True
metrics_cache_timer["recalculate_timeout"] = recalculate_timeout
cache.set(metrics_cache_timer_key, metrics_cache_timer, timeout=recalculate_timeout)

17
engine/apps/metrics_exporter/metrics_cache_manager.py
View file

@ -1,3 +1,5 @@
import typing
from apps.alerts.constants import AlertGroupState
from apps.metrics_exporter.helpers import (
get_response_time_period,
@ -7,16 +9,23 @@ from apps.metrics_exporter.helpers import (
class MetricsCacheManager:
class _TeamsDiff(typing.TypedDict):
team_name: str | None
deleted: bool
TeamsDiffMap = typing.Dict[int, _TeamsDiff]
@staticmethod
def get_default_teams_diff_dict():
default_dict = {
def get_default_teams_diff_dict() -> _TeamsDiff:
return {
"team_name": None,
"deleted": False,
}
return default_dict
@staticmethod
def update_team_diff(teams_diff, team_id, new_name=None, deleted=False):
def update_team_diff(
teams_diff: TeamsDiffMap, team_id: int, new_name: str | None = None, deleted: bool = False
) -> TeamsDiffMap:
teams_diff.setdefault(team_id, MetricsCacheManager.get_default_teams_diff_dict())
teams_diff[team_id]["team_name"] = new_name
teams_diff[team_id]["deleted"] = deleted

4
engine/apps/metrics_exporter/tests/test_update_metics_cache.py
View file

@ -22,7 +22,7 @@ from apps.metrics_exporter.tests.conftest import (
)
@patch("apps.alerts.models.alert_group_log_record.send_update_log_report_signal.apply_async")
@patch("apps.alerts.models.alert_group_log_record.tasks.send_update_log_report_signal.apply_async")
@patch("apps.alerts.models.alert_group.alert_group_action_triggered_signal.send")
@pytest.mark.django_db
@override_settings(CELERY_TASK_ALWAYS_EAGER=True)
@ -130,7 +130,7 @@ def test_update_metric_alert_groups_total_cache_on_action(
get_called_arg_index_and_compare_results(expected_result_firing)
@patch("apps.alerts.models.alert_group_log_record.send_update_log_report_signal.apply_async")
@patch("apps.alerts.models.alert_group_log_record.tasks.send_update_log_report_signal.apply_async")
@patch("apps.alerts.models.alert_group.alert_group_action_triggered_signal.send")
@pytest.mark.django_db
@override_settings(CELERY_TASK_ALWAYS_EAGER=True)

8
engine/apps/mobile_app/tasks.py
View file

@ -268,6 +268,7 @@ def _get_youre_going_oncall_fcm_message(
thread_id = f"{schedule.public_primary_key}:{user.public_primary_key}:going-oncall"
mobile_app_user_settings, _ = MobileAppUserSettings.objects.get_or_create(user=user)
info_notification_sound_name = mobile_app_user_settings.info_notification_sound_name
notification_title = _get_youre_going_oncall_notification_title(seconds_until_going_oncall)
notification_subtitle = _get_youre_going_oncall_notification_subtitle(
@ -277,9 +278,7 @@ def _get_youre_going_oncall_fcm_message(
data: FCMMessageData = {
"title": notification_title,
"subtitle": notification_subtitle,
"info_notification_sound_name": (
mobile_app_user_settings.info_notification_sound_name + MobileAppUserSettings.ANDROID_SOUND_NAME_EXTENSION
),
"info_notification_sound_name": f"{info_notification_sound_name}{MobileAppUserSettings.ANDROID_SOUND_NAME_EXTENSION}",
"info_notification_volume_type": mobile_app_user_settings.info_notification_volume_type,
"info_notification_volume": str(mobile_app_user_settings.info_notification_volume),
"info_notification_volume_override": json.dumps(mobile_app_user_settings.info_notification_volume_override),
@ -291,8 +290,7 @@ def _get_youre_going_oncall_fcm_message(
alert=ApsAlert(title=notification_title, subtitle=notification_subtitle),
sound=CriticalSound(
critical=False,
name=mobile_app_user_settings.info_notification_sound_name
+ MobileAppUserSettings.IOS_SOUND_NAME_EXTENSION,
name=f"{info_notification_sound_name}{MobileAppUserSettings.IOS_SOUND_NAME_EXTENSION}",
),
custom_data={
"interruption-level": "time-sensitive",

10
engine/apps/phone_notifications/phone_provider.py
View file

@ -1,6 +1,6 @@
import typing
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional
from django.conf import settings
from django.utils.module_loading import import_string
@ -46,7 +46,7 @@ class PhoneProvider(ABC):
TwilioPhoneProvider as example of complicated phone provider which supports status callbacks and gather actions.
"""
def make_notification_call(self, number: str, text: str) -> Optional[ProviderPhoneCall]:
def make_notification_call(self, number: str, text: str) -> typing.Optional[ProviderPhoneCall]:
"""
make_notification_call makes a call to notify about alert group and optionally returns unsaved ProviderPhoneCall
instance. If returned, instance will be linked to PhoneCallRecord and saved by PhoneBackend.
@ -68,7 +68,7 @@ class PhoneProvider(ABC):
"""
raise ProviderNotSupports
def send_notification_sms(self, number: str, message: str) -> Optional[ProviderSMS]:
def send_notification_sms(self, number: str, message: str) -> typing.Optional[ProviderSMS]:
"""
send_notification_sms sends a sms to notify about alert group.
@ -147,7 +147,7 @@ class PhoneProvider(ABC):
"""
raise ProviderNotSupports
def finish_verification(self, number: str, code: str) -> Optional[str]:
def finish_verification(self, number: str, code: str) -> typing.Optional[str]:
"""
finish_verification validates the verification code.
@ -172,7 +172,7 @@ class PhoneProvider(ABC):
raise NotImplementedError
_providers = {}
_providers: typing.Dict[str, PhoneProvider] = {}
def get_phone_provider() -> PhoneProvider:

4
engine/apps/schedules/ical_utils.py
View file

@ -589,8 +589,8 @@ def _get_ical_data_final_schedule(schedule: "OnCallSchedule") -> str | None:
ical_data = schedule.cached_ical_final_schedule
if ical_data is None:
schedule.refresh_ical_final_schedule()
# typing is safe here. cached_ical_final_schedule is updated inside of refresh_ical_final_schedule
ical_data: str = schedule.cached_ical_final_schedule
# casting is safe here. cached_ical_final_schedule is updated inside of refresh_ical_final_schedule
return typing.cast(str, schedule.cached_ical_final_schedule)
return ical_data

10
engine/apps/schedules/models/on_call_schedule.py
View file

@ -258,14 +258,18 @@ class OnCallSchedule(PolymorphicModel):
def get_icalendars(self) -> typing.Tuple[typing.Optional[icalendar.Calendar], typing.Optional[icalendar.Calendar]]:
"""Returns list of calendars. Primary calendar should always be the first"""
calendar_primary: typing.Optional[icalendar.Calendar] = None
calendar_overrides: typing.Optional[icalendar.Calendar] = None
# if self._ical_file_(primary|overrides) is None -> no cache, will trigger a refresh
# if self._ical_file_(primary|overrides) == "" -> cached value for an empty schedule
if self._ical_file_primary:
calendar_primary: icalendar.Calendar = icalendar.Calendar.from_ical(self._ical_file_primary)
else:
calendar_primary = None
if self._ical_file_overrides:
calendar_overrides = icalendar.Calendar.from_ical(self._ical_file_overrides)
calendar_overrides: icalendar.Calendar = icalendar.Calendar.from_ical(self._ical_file_overrides)
else:
calendar_overrides = None
return calendar_primary, calendar_overrides
def get_prev_and_current_ical_files(self):

4
engine/apps/slack/scenarios/alertgroup_appearance.py
View file

@ -26,7 +26,7 @@ class OpenAlertAppearanceDialogStep(AlertGroupActionsMixin, scenario_step.Scenar
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = self.get_alert_group(slack_team_identity, payload)
if not self.is_authorized(alert_group):
@ -78,7 +78,7 @@ class UpdateAppearanceStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
from apps.alerts.models import AlertGroup

2
engine/apps/slack/scenarios/declare_incident.py
View file

@ -12,7 +12,7 @@ class DeclareIncidentStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
"""
Slack sends a POST request to the backend upon clicking a button with a redirect link to Incident.

32
engine/apps/slack/scenarios/distribute_alerts.py
View file

@ -35,7 +35,7 @@ from apps.slack.tasks import (
from apps.slack.types import (
Block,
BlockActionType,
CompositionObjects,
CompositionObjectOption,
EventPayload,
InteractiveMessageActionType,
ModalView,
@ -231,7 +231,7 @@ class AlertShootingStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
pass
@ -248,7 +248,7 @@ class InviteOtherPersonToIncident(AlertGroupActionsMixin, scenario_step.Scenario
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
from apps.user_management.models import User
@ -284,7 +284,7 @@ class SilenceGroupStep(AlertGroupActionsMixin, scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = self.get_alert_group(slack_team_identity, payload)
if not self.is_authorized(alert_group):
@ -311,7 +311,7 @@ class UnSilenceGroupStep(AlertGroupActionsMixin, scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = self.get_alert_group(slack_team_identity, payload)
if not self.is_authorized(alert_group):
@ -331,7 +331,7 @@ class SelectAttachGroupStep(AlertGroupActionsMixin, scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = self.get_alert_group(slack_team_identity, payload)
if not self.is_authorized(alert_group):
@ -407,7 +407,7 @@ class SelectAttachGroupStep(AlertGroupActionsMixin, scenario_step.ScenarioStep):
)
def get_select_incidents_blocks(self, alert_group: AlertGroup) -> Block.AnyBlocks:
collected_options: typing.List[CompositionObjects.Option] = []
collected_options: typing.List[CompositionObjectOption] = []
blocks: Block.AnyBlocks = []
alert_receive_channel_ids = AlertReceiveChannel.objects.filter(
@ -502,7 +502,7 @@ class AttachGroupStep(AlertGroupActionsMixin, scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
# submit selection in modal window
if payload["type"] == PayloadType.VIEW_SUBMISSION:
@ -532,7 +532,7 @@ class UnAttachGroupStep(AlertGroupActionsMixin, scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = self.get_alert_group(slack_team_identity, payload)
if not self.is_authorized(alert_group):
@ -557,7 +557,7 @@ class StopInvitationProcess(AlertGroupActionsMixin, scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = self.get_alert_group(slack_team_identity, payload)
if not self.is_authorized(alert_group):
@ -584,7 +584,7 @@ class CustomButtonProcessStep(AlertGroupActionsMixin, scenario_step.ScenarioStep
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
from apps.alerts.models import CustomButton
@ -647,7 +647,7 @@ class ResolveGroupStep(AlertGroupActionsMixin, scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
ResolutionNoteModalStep = scenario_step.ScenarioStep.get_step("resolution_note", "ResolutionNoteModalStep")
@ -688,7 +688,7 @@ class UnResolveGroupStep(AlertGroupActionsMixin, scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = self.get_alert_group(slack_team_identity, payload)
if not self.is_authorized(alert_group):
@ -708,7 +708,7 @@ class AcknowledgeGroupStep(AlertGroupActionsMixin, scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = self.get_alert_group(slack_team_identity, payload)
if not self.is_authorized(alert_group):
@ -728,7 +728,7 @@ class UnAcknowledgeGroupStep(AlertGroupActionsMixin, scenario_step.ScenarioStep)
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = self.get_alert_group(slack_team_identity, payload)
if not self.is_authorized(alert_group):
@ -795,7 +795,7 @@ class AcknowledgeConfirmationStep(AcknowledgeGroupStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
from apps.alerts.models import AlertGroup

2
engine/apps/slack/scenarios/invited_to_channel.py
View file

@ -19,7 +19,7 @@ class InvitedToChannelStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
if payload["event"]["user"] == slack_team_identity.bot_user_id:
channel_id = payload["event"]["channel"]

16
engine/apps/slack/scenarios/manage_responders.py
View file

@ -38,7 +38,7 @@ class StartManageResponders(AlertGroupActionsMixin, scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = self.get_alert_group(slack_team_identity, payload)
if not self.is_authorized(alert_group):
@ -60,7 +60,7 @@ class ManageRespondersUserChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = _get_alert_group_from_payload(payload)
selected_user = _get_selected_user_from_payload(payload)
@ -111,7 +111,7 @@ class ManageRespondersConfirmUserChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = _get_alert_group_from_payload(payload)
selected_user = _get_selected_user_from_payload(payload)
@ -144,7 +144,7 @@ class ManageRespondersScheduleChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = _get_alert_group_from_payload(payload)
selected_schedule = _get_selected_schedule_from_payload(payload)
@ -177,7 +177,7 @@ class ManageRespondersRemoveUser(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
alert_group = _get_alert_group_from_payload(payload)
selected_user = _get_selected_user_from_payload(payload)
@ -255,7 +255,7 @@ def render_dialog(alert_group: "AlertGroup", alert_group_resolved_warning=False)
return view
def _get_selected_user_from_payload(payload: EventPayload.Any) -> "User":
def _get_selected_user_from_payload(payload: EventPayload) -> "User":
from apps.user_management.models import User
try:
@ -274,7 +274,7 @@ def _get_selected_user_from_payload(payload: EventPayload.Any) -> "User":
return User.objects.get(pk=selected_user_id)
def _get_selected_schedule_from_payload(payload: EventPayload.Any) -> "OnCallSchedule":
def _get_selected_schedule_from_payload(payload: EventPayload) -> "OnCallSchedule":
from apps.schedules.models import OnCallSchedule
input_id_prefix = json.loads(payload["view"]["private_metadata"])["input_id_prefix"]
@ -285,7 +285,7 @@ def _get_selected_schedule_from_payload(payload: EventPayload.Any) -> "OnCallSch
return OnCallSchedule.objects.get(pk=selected_schedule_id)
def _get_alert_group_from_payload(payload: EventPayload.Any) -> "AlertGroup":
def _get_alert_group_from_payload(payload: EventPayload) -> "AlertGroup":
from apps.alerts.models import AlertGroup
alert_group_pk = json.loads(payload["view"]["private_metadata"])[ALERT_GROUP_DATA_KEY]

34
engine/apps/slack/scenarios/manual_incident.py
View file

@ -11,7 +11,7 @@ from apps.slack.slack_client.exceptions import SlackAPIException
from apps.slack.types import (
Block,
BlockActionType,
CompositionObjects,
CompositionObjectOption,
EventPayload,
ModalView,
PayloadType,
@ -45,7 +45,7 @@ class StartCreateIncidentFromSlashCommand(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
input_id_prefix = _generate_input_id_prefix()
@ -84,7 +84,7 @@ class FinishCreateIncidentFromSlashCommand(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
from apps.alerts.models import Alert
@ -165,7 +165,7 @@ class OnOrgChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
private_metadata = json.loads(payload["view"]["private_metadata"])
with_title_and_message_inputs = private_metadata.get("with_title_and_message_inputs", False)
@ -214,7 +214,7 @@ class OnTeamChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
private_metadata = json.loads(payload["view"]["private_metadata"])
with_title_and_message_inputs = private_metadata.get("with_title_and_message_inputs", False)
@ -266,7 +266,7 @@ class OnRouteChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
pass
@ -314,7 +314,7 @@ def _get_manual_incident_initial_form_fields(
slack_team_identity: "SlackTeamIdentity",
slack_user_identity: "SlackUserIdentity",
input_id_prefix: str,
payload: EventPayload.Any,
payload: EventPayload,
with_title_and_message_inputs=False,
) -> Block.AnyBlocks:
initial_organization = (
@ -363,7 +363,7 @@ def _get_organization_select(
organizations = slack_team_identity.organizations.filter(
users__slack_user_identity=slack_user_identity,
).distinct()
organizations_options: typing.List[CompositionObjects.Option] = []
organizations_options: typing.List[CompositionObjectOption] = []
initial_option_idx = 0
for idx, org in enumerate(organizations):
if org == value:
@ -395,7 +395,7 @@ def _get_organization_select(
return organization_select
def _get_selected_org_from_payload(payload: EventPayload.Any, input_id_prefix: str) -> typing.Optional["Organization"]:
def _get_selected_org_from_payload(payload: EventPayload, input_id_prefix: str) -> typing.Optional["Organization"]:
from apps.user_management.models import Organization
selected_org_id = payload["view"]["state"]["values"][input_id_prefix + MANUAL_INCIDENT_ORG_SELECT_ID][
@ -410,7 +410,7 @@ def _get_team_select(
teams = organization.teams.filter(
users__slack_user_identity=slack_user_identity,
).distinct()
team_options: typing.List[CompositionObjects.Option] = []
team_options: typing.List[CompositionObjectOption] = []
# Adding pseudo option for default team
initial_option_idx = 0
team_options.append(
@ -453,7 +453,7 @@ def _get_team_select(
return team_select
def _get_selected_team_from_payload(payload: EventPayload.Any, input_id_prefix: str) -> typing.Optional["Team"]:
def _get_selected_team_from_payload(payload: EventPayload, input_id_prefix: str) -> typing.Optional["Team"]:
from apps.user_management.models import Team
selected_team_id = payload["view"]["state"]["values"][input_id_prefix + MANUAL_INCIDENT_TEAM_SELECT_ID][
@ -465,7 +465,7 @@ def _get_selected_team_from_payload(payload: EventPayload.Any, input_id_prefix:
def _get_route_select(integration: AlertReceiveChannel, value, input_id_prefix: str) -> Block.Section:
route_options: typing.List[CompositionObjects.Option] = []
route_options: typing.List[CompositionObjectOption] = []
initial_option_idx = 0
for idx, route in enumerate(integration.channel_filters.all()):
filtering_term = f'"{route.filtering_term}"'
@ -498,7 +498,7 @@ def _get_route_select(integration: AlertReceiveChannel, value, input_id_prefix:
return route_select
def _get_selected_route_from_payload(payload: EventPayload.Any, input_id_prefix: str) -> ChannelFilter | None:
def _get_selected_route_from_payload(payload: EventPayload, input_id_prefix: str) -> ChannelFilter | None:
from apps.alerts.models import ChannelFilter
selected_org_id = payload["view"]["state"]["values"][input_id_prefix + MANUAL_INCIDENT_ROUTE_SELECT_ID][
@ -516,7 +516,7 @@ def _get_and_change_input_id_prefix_from_metadata(
return old_input_id_prefix, new_input_id_prefix, metadata
def _get_title_input(payload: EventPayload.Any) -> Block.Input:
def _get_title_input(payload: EventPayload) -> Block.Input:
title_input_block: Block.Input = {
"type": "input",
"block_id": MANUAL_INCIDENT_TITLE_INPUT_ID,
@ -538,14 +538,14 @@ def _get_title_input(payload: EventPayload.Any) -> Block.Input:
return title_input_block
def _get_title_from_payload(payload: EventPayload.Any) -> str:
def _get_title_from_payload(payload: EventPayload) -> str:
title = payload["view"]["state"]["values"][MANUAL_INCIDENT_TITLE_INPUT_ID][
FinishCreateIncidentFromSlashCommand.routing_uid()
]["value"]
return title
def _get_message_input(payload: EventPayload.Any) -> Block.Input:
def _get_message_input(payload: EventPayload) -> Block.Input:
message_input_block: Block.Input = {
"type": "input",
"block_id": MANUAL_INCIDENT_MESSAGE_INPUT_ID,
@ -569,7 +569,7 @@ def _get_message_input(payload: EventPayload.Any) -> Block.Input:
return message_input_block
def _get_message_from_payload(payload: EventPayload.Any) -> str:
def _get_message_from_payload(payload: EventPayload) -> str:
return (
payload["view"]["state"]["values"][MANUAL_INCIDENT_MESSAGE_INPUT_ID][
FinishCreateIncidentFromSlashCommand.routing_uid()

2
engine/apps/slack/scenarios/notified_user_not_in_channel.py
View file

@ -20,7 +20,7 @@ class NotifiedUserNotInChannelStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
logger.info("Gracefully handle NotifiedUserNotInChannelStep. Do nothing.")
pass

4
engine/apps/slack/scenarios/onboarding.py
View file

@ -19,7 +19,7 @@ class ImOpenStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
logger.info("InOpenStep, doing nothing.")
@ -29,7 +29,7 @@ class AppHomeOpenedStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
pass

96
engine/apps/slack/scenarios/paging.py
View file

@ -4,7 +4,7 @@ import typing
from uuid import uuid4
from django.conf import settings
from django.db.models import Model
from django.db.models import Model, QuerySet
from apps.alerts.models import AlertReceiveChannel, EscalationChain
from apps.alerts.paging import (
@ -21,7 +21,8 @@ from apps.slack.slack_client.exceptions import SlackAPIException
from apps.slack.types import (
Block,
BlockActionType,
CompositionObjects,
CompositionObjectOption,
CompositionObjectOptionGroup,
EventPayload,
ModalView,
PayloadType,
@ -76,7 +77,7 @@ class DataKey(enum.StrEnum):
MAX_STATIC_SELECT_OPTIONS = 100
def add_or_update_item(payload: EventPayload.Any, key: DataKey, item_pk: str, policy: Policy) -> EventPayload:
def add_or_update_item(payload: EventPayload, key: DataKey, item_pk: str, policy: Policy) -> EventPayload:
metadata = json.loads(payload["view"]["private_metadata"])
metadata[key][item_pk] = policy
updated_metadata = json.dumps(metadata)
@ -86,7 +87,7 @@ def add_or_update_item(payload: EventPayload.Any, key: DataKey, item_pk: str, po
return payload
def remove_item(payload: EventPayload.Any, key: DataKey, item_pk: str) -> EventPayload:
def remove_item(payload: EventPayload, key: DataKey, item_pk: str) -> EventPayload:
metadata = json.loads(payload["view"]["private_metadata"])
if item_pk in metadata[key]:
del metadata[key][item_pk]
@ -94,7 +95,7 @@ def remove_item(payload: EventPayload.Any, key: DataKey, item_pk: str) -> EventP
return payload
def reset_items(payload: EventPayload.Any) -> EventPayload:
def reset_items(payload: EventPayload) -> EventPayload:
metadata = json.loads(payload["view"]["private_metadata"])
for key in (DataKey.USERS, DataKey.SCHEDULES):
metadata[key] = {}
@ -106,10 +107,10 @@ T = typing.TypeVar("T", bound=Model)
def get_current_items(
payload: EventPayload.Any, key: DataKey, qs: "RelatedManager['T']"
payload: EventPayload, key: DataKey, qs: "RelatedManager['T']"
) -> typing.List[typing.Tuple[T, Policy]]:
metadata = json.loads(payload["view"]["private_metadata"])
items: typing.List[T] = []
items: typing.List[typing.Tuple[T, Policy]] = []
for u, p in metadata[key].items():
item = qs.filter(pk=u).first()
items.append((item, p))
@ -128,7 +129,7 @@ class StartDirectPaging(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
input_id_prefix = _generate_input_id_prefix()
@ -160,7 +161,7 @@ class FinishDirectPaging(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
title = _get_title_from_payload(payload)
message = _get_message_from_payload(payload)
@ -230,7 +231,7 @@ class OnPagingOrgChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
updated_payload = reset_items(payload)
view = render_dialog(slack_user_identity, slack_team_identity, updated_payload)
@ -249,7 +250,7 @@ class OnPagingTeamChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
view = render_dialog(slack_user_identity, slack_team_identity, payload)
self._slack_client.api_call(
@ -274,7 +275,7 @@ class OnPagingUserChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
private_metadata = json.loads(payload["view"]["private_metadata"])
selected_organization = _get_selected_org_from_payload(
@ -314,7 +315,7 @@ class OnPagingUserChange(scenario_step.ScenarioStep):
class OnPagingItemActionChange(scenario_step.ScenarioStep):
"""Reload form with updated user details."""
def _parse_action(self, payload: EventPayload.Any) -> typing.Tuple[Policy, str, str]:
def _parse_action(self, payload: EventPayload) -> typing.Tuple[Policy, str, str]:
value = payload["actions"][0]["selected_option"]["value"]
return value.split("|")
@ -322,7 +323,7 @@ class OnPagingItemActionChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
policy, key, user_pk = self._parse_action(payload)
@ -352,7 +353,7 @@ class OnPagingConfirmUserChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
metadata = json.loads(payload["view"]["private_metadata"])
@ -397,7 +398,7 @@ class OnPagingScheduleChange(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
private_metadata = json.loads(payload["view"]["private_metadata"])
selected_schedule = _get_selected_schedule_from_payload(payload, private_metadata["input_id_prefix"])
@ -425,7 +426,7 @@ class OnPagingScheduleChange(scenario_step.ScenarioStep):
def render_dialog(
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
initial=False,
error_msg=None,
) -> ModalView:
@ -503,9 +504,9 @@ def _get_form_view(routing_uid: str, blocks: Block.AnyBlocks, private_metadata:
def _get_organization_select(
organizations: "RelatedManager['Organization']", value: "Organization", input_id_prefix: str
organizations: QuerySet["Organization"], value: "Organization", input_id_prefix: str
) -> Block.Input:
organizations_options: typing.List[CompositionObjects.Option] = []
organizations_options: typing.List[CompositionObjectOption] = []
initial_option_idx = 0
for idx, org in enumerate(organizations):
if org == value:
@ -541,7 +542,7 @@ def _get_organization_select(
return organization_select
def _get_select_field_value(payload: EventPayload.Any, prefix_id: str, routing_uid: str, field_id: str) -> str | None:
def _get_select_field_value(payload: EventPayload, prefix_id: str, routing_uid: str, field_id: str) -> str | None:
try:
field = payload["view"]["state"]["values"][prefix_id + field_id][routing_uid]["selected_option"]
except KeyError:
@ -550,7 +551,7 @@ def _get_select_field_value(payload: EventPayload.Any, prefix_id: str, routing_u
def _get_selected_org_from_payload(
payload: EventPayload.Any,
payload: EventPayload,
input_id_prefix: str,
slack_team_identity: "SlackTeamIdentity",
slack_user_identity: "SlackUserIdentity",
@ -575,7 +576,7 @@ def _get_team_select_blocks(
user = slack_user_identity.get_user(organization) # TODO: handle None
teams = user.available_teams
team_options: typing.List[CompositionObjects.Option] = []
team_options: typing.List[CompositionObjectOption] = []
# Adding pseudo option for default team
initial_option_idx = 0
team_options.append(
@ -668,13 +669,13 @@ def _get_team_select_context(organization: "Organization", team: "Team") -> Bloc
def _get_additional_responders_blocks(
payload: EventPayload.Any,
payload: EventPayload,
organization: "Organization",
input_id_prefix,
is_additional_responders_checked: bool,
error_msg: str | None,
) -> Block.AnyBlocks:
checkbox_option: CompositionObjects.Option = {
checkbox_option: CompositionObjectOption = {
"text": {
"type": "plain_text",
"text": "Notify additional responders",
@ -743,7 +744,7 @@ def _get_users_select(
) -> Block.Context | Block.Section:
users = organization.users.all()
user_options: typing.List[CompositionObjects.Option] = [
user_options: typing.List[CompositionObjectOption] = [
{
"text": {
"type": "plain_text",
@ -756,8 +757,9 @@ def _get_users_select(
]
if not user_options:
user_select: Block.Context = {"type": "context", "elements": [{"type": "mrkdwn", "text": "No users available"}]}
return user_select
return typing.cast(
Block.Context, {"type": "context", "elements": [{"type": "mrkdwn", "text": "No users available"}]}
)
user_select: Block.Section = {
"type": "section",
@ -783,7 +785,7 @@ def _get_schedules_select(
) -> Block.Context | Block.Section:
schedules = organization.oncall_schedules.all()
schedule_options: typing.List[CompositionObjects.Option] = [
schedule_options: typing.List[CompositionObjectOption] = [
{
"text": {
"type": "plain_text",
@ -796,11 +798,13 @@ def _get_schedules_select(
]
if not schedule_options:
schedule_select: Block.Context = {
"type": "context",
"elements": [{"type": "mrkdwn", "text": "No schedules available"}],
}
return schedule_select
return typing.cast(
Block.Context,
{
"type": "context",
"elements": [{"type": "mrkdwn", "text": "No schedules available"}],
},
)
schedule_select: Block.Section = {
"type": "section",
@ -822,11 +826,11 @@ def _get_schedules_select(
def _get_option_groups(
options: typing.List[CompositionObjects.Option], max_options_per_group: int
) -> typing.List[CompositionObjects.OptionGroup]:
options: typing.List[CompositionObjectOption], max_options_per_group: int
) -> typing.List[CompositionObjectOptionGroup]:
chunks = [options[x : x + max_options_per_group] for x in range(0, len(options), max_options_per_group)]
option_groups: typing.List[CompositionObjects.OptionGroup] = []
option_groups: typing.List[CompositionObjectOptionGroup] = []
for idx, group in enumerate(chunks):
start = idx * max_options_per_group + 1
end = idx * max_options_per_group + max_options_per_group
@ -876,7 +880,7 @@ def _get_selected_entries_list(
def _display_availability_warnings(
payload: EventPayload.Any, warnings: typing.List[AvailabilityWarning], organization: "Organization", user: "User"
payload: EventPayload, warnings: typing.List[AvailabilityWarning], organization: "Organization", user: "User"
) -> ModalView:
metadata = json.loads(payload["view"]["private_metadata"])
return _get_availability_warnings_view(
@ -941,7 +945,7 @@ def _get_availability_warnings_view(
def _get_selected_team_from_payload(
payload: EventPayload.Any, input_id_prefix: str
payload: EventPayload, input_id_prefix: str
) -> typing.Tuple[str | None, typing.Optional["Team"]]:
from apps.user_management.models import Team
@ -959,7 +963,7 @@ def _get_selected_team_from_payload(
return selected_team_id, team
def _get_additional_responders_checked_from_payload(payload: EventPayload.Any, input_id_prefix: str) -> bool:
def _get_additional_responders_checked_from_payload(payload: EventPayload, input_id_prefix: str) -> bool:
try:
selected_options = payload["view"]["state"]["values"][
input_id_prefix + DIRECT_PAGING_ADDITIONAL_RESPONDERS_INPUT_ID
@ -970,7 +974,7 @@ def _get_additional_responders_checked_from_payload(payload: EventPayload.Any, i
return len(selected_options) > 0
def _get_selected_user_from_payload(payload: EventPayload.Any, input_id_prefix: str) -> typing.Optional["User"]:
def _get_selected_user_from_payload(payload: EventPayload, input_id_prefix: str) -> typing.Optional["User"]:
from apps.user_management.models import User
selected_user_id = _get_select_field_value(
@ -983,7 +987,7 @@ def _get_selected_user_from_payload(payload: EventPayload.Any, input_id_prefix:
def _get_selected_schedule_from_payload(
payload: EventPayload.Any, input_id_prefix: str
payload: EventPayload, input_id_prefix: str
) -> typing.Optional["OnCallSchedule"]:
from apps.schedules.models import OnCallSchedule
@ -1004,7 +1008,7 @@ def _get_and_change_input_id_prefix_from_metadata(
return old_input_id_prefix, new_input_id_prefix, metadata
def _get_title_input(payload: EventPayload.Any) -> Block.Input:
def _get_title_input(payload: EventPayload) -> Block.Input:
title_input_block: Block.Input = {
"type": "input",
"block_id": DIRECT_PAGING_TITLE_INPUT_ID,
@ -1026,12 +1030,12 @@ def _get_title_input(payload: EventPayload.Any) -> Block.Input:
return title_input_block
def _get_title_from_payload(payload: EventPayload.Any) -> str:
def _get_title_from_payload(payload: EventPayload) -> str:
title = payload["view"]["state"]["values"][DIRECT_PAGING_TITLE_INPUT_ID][FinishDirectPaging.routing_uid()]["value"]
return title
def _get_message_input(payload: EventPayload.Any) -> Block.Input:
def _get_message_input(payload: EventPayload) -> Block.Input:
message_input_block: Block.Input = {
"type": "input",
"block_id": DIRECT_PAGING_MESSAGE_INPUT_ID,
@ -1055,7 +1059,7 @@ def _get_message_input(payload: EventPayload.Any) -> Block.Input:
return message_input_block
def _get_message_from_payload(payload: EventPayload.Any) -> str:
def _get_message_from_payload(payload: EventPayload) -> str:
return (
payload["view"]["state"]["values"][DIRECT_PAGING_MESSAGE_INPUT_ID][FinishDirectPaging.routing_uid()]["value"]
or ""
@ -1064,7 +1068,7 @@ def _get_message_from_payload(payload: EventPayload.Any) -> str:
def _get_available_organizations(
slack_team_identity: "SlackTeamIdentity", slack_user_identity: "SlackUserIdentity"
) -> "RelatedManager['Organization']":
) -> QuerySet["Organization"]:
return (
slack_team_identity.organizations.filter(users__slack_user_identity=slack_user_identity)
.order_by("pk")

2
engine/apps/slack/scenarios/profile_update.py
View file

@ -13,7 +13,7 @@ class ProfileUpdateStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
"""
Triggered by action: Any update in Slack Profile.

96
engine/apps/slack/scenarios/resolution_note.py
View file

@ -41,7 +41,7 @@ class AddToResolutionNoteStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
from apps.alerts.models import ResolutionNote, ResolutionNoteSlackMessage
from apps.slack.models import SlackMessage, SlackUserIdentity
@ -401,7 +401,7 @@ class ResolutionNoteModalStep(AlertGroupActionsMixin, scenario_step.ScenarioStep
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
data: ScenarioData | None = None,
) -> None:
if data:
@ -599,54 +599,58 @@ class ResolutionNoteModalStep(AlertGroupActionsMixin, scenario_step.ScenarioStep
message_timestamp = datetime.datetime.timestamp(resolution_note.created_at)
blocks.append(DIVIDER)
source = "web" if resolution_note.source == ResolutionNote.Source.WEB else "slack"
message_block: Block.Section = {
"type": "section",
"text": {
"type": "mrkdwn",
"text": "{} <!date^{:.0f}^{{date_num}} {{time_secs}}|note_created_at> (from {})\n{}".format(
user_verbal,
float(message_timestamp),
source,
resolution_note.message_text,
),
},
"accessory": {
"type": "button",
"style": "danger",
"text": {
"type": "plain_text",
"text": "Remove",
"emoji": True,
},
"action_id": AddRemoveThreadMessageStep.routing_uid(),
"value": json.dumps(
{
"resolution_note_window_action": "edit",
"msg_value": "remove",
"message_pk": None
if not resolution_note_slack_message
else resolution_note_slack_message.pk,
"resolution_note_pk": resolution_note.pk,
"alert_group_pk": alert_group.pk,
}
),
"confirm": {
"title": {"type": "plain_text", "text": "Are you sure?"},
blocks.append(
typing.cast(
Block.Section,
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": "This operation will permanently delete this Resolution Note.",
"text": "{} <!date^{:.0f}^{{date_num}} {{time_secs}}|note_created_at> (from {})\n{}".format(
user_verbal,
float(message_timestamp),
source,
resolution_note.message_text,
),
},
"confirm": {"type": "plain_text", "text": "Delete"},
"deny": {
"type": "plain_text",
"text": "Stop, I've changed my mind!",
"accessory": {
"type": "button",
"style": "danger",
"text": {
"type": "plain_text",
"text": "Remove",
"emoji": True,
},
"action_id": AddRemoveThreadMessageStep.routing_uid(),
"value": json.dumps(
{
"resolution_note_window_action": "edit",
"msg_value": "remove",
"message_pk": None
if not resolution_note_slack_message
else resolution_note_slack_message.pk,
"resolution_note_pk": resolution_note.pk,
"alert_group_pk": alert_group.pk,
}
),
"confirm": {
"title": {"type": "plain_text", "text": "Are you sure?"},
"text": {
"type": "mrkdwn",
"text": "This operation will permanently delete this Resolution Note.",
},
"confirm": {"type": "plain_text", "text": "Delete"},
"deny": {
"type": "plain_text",
"text": "Stop, I've changed my mind!",
},
"style": "danger",
},
},
"style": "danger",
},
},
}
blocks.append(message_block)
)
)
if not blocks:
# there aren't any resolution notes yet, display a hint instead
@ -711,7 +715,7 @@ class AddRemoveThreadMessageStep(UpdateResolutionNoteStep, scenario_step.Scenari
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
from apps.alerts.models import AlertGroup, ResolutionNote, ResolutionNoteSlackMessage

3
engine/apps/slack/scenarios/scenario_step.py
View file

@ -48,8 +48,7 @@ class ScenarioStep(object):
# https://stackoverflow.com/posts/36442015/revisions
try:
module = importlib.import_module("apps.slack.scenarios." + scenario)
step = getattr(module, step)
return step
return getattr(module, step)
except ImportError as e:
raise Exception("Check import spelling! Scenario: {}, Step:{}, Error: {}".format(scenario, step, e))

14
engine/apps/slack/scenarios/schedules.py
View file

@ -8,7 +8,7 @@ from apps.slack.scenarios import scenario_step
from apps.slack.types import (
Block,
BlockActionType,
CompositionObjects,
CompositionObjectOption,
EventPayload,
ModalView,
PayloadType,
@ -31,14 +31,14 @@ class EditScheduleShiftNotifyStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
if payload["actions"][0].get("value", None) and payload["actions"][0]["value"].startswith("edit"):
self.open_settings_modal(payload)
elif payload["actions"][0].get("type", None) and payload["actions"][0]["type"] == "static_select":
self.set_selected_value(slack_user_identity, payload)
def open_settings_modal(self, payload: EventPayload.Any) -> None:
def open_settings_modal(self, payload: EventPayload) -> None:
schedule_id = payload["actions"][0]["value"].split("_")[1]
try:
_ = OnCallSchedule.objects.get(pk=schedule_id) # noqa
@ -67,7 +67,7 @@ class EditScheduleShiftNotifyStep(scenario_step.ScenarioStep):
view=view,
)
def set_selected_value(self, slack_user_identity: "SlackUserIdentity", payload: EventPayload.Any) -> None:
def set_selected_value(self, slack_user_identity: "SlackUserIdentity", payload: EventPayload) -> None:
action = payload["actions"][0]
private_metadata = json.loads(payload["view"]["private_metadata"])
schedule_id = private_metadata["schedule_id"]
@ -138,20 +138,20 @@ class EditScheduleShiftNotifyStep(scenario_step.ScenarioStep):
return blocks
def get_options(self, select_name: str) -> typing.List[CompositionObjects.Option]:
def get_options(self, select_name: str) -> typing.List[CompositionObjectOption]:
select_options = getattr(self, f"{select_name}_options")
return [
{"text": {"type": "plain_text", "text": select_options[option]}, "value": str(option)}
for option in select_options
]
def get_initial_option(self, schedule_id: str, select_name: str) -> CompositionObjects.Option:
def get_initial_option(self, schedule_id: str, select_name: str) -> CompositionObjectOption:
schedule = OnCallSchedule.objects.get(pk=schedule_id)
current_value = getattr(schedule, select_name)
text = getattr(self, f"{select_name}_options")[current_value]
initial_option: CompositionObjects.Option = {
initial_option: CompositionObjectOption = {
"text": {
"type": "plain_text",
"text": f"{text}",

2
engine/apps/slack/scenarios/shift_swap_requests.py
View file

@ -184,7 +184,7 @@ class AcceptShiftSwapRequestStep(BaseShiftSwapRequestStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
from apps.schedules import exceptions
from apps.schedules.models import ShiftSwapRequest

8
engine/apps/slack/scenarios/slack_channel.py
View file

@ -16,7 +16,7 @@ class SlackChannelCreatedOrRenamedEventStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
"""
Triggered by action: Create or rename channel
@ -41,7 +41,7 @@ class SlackChannelDeletedEventStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
"""
Triggered by action: Delete channel
@ -63,7 +63,7 @@ class SlackChannelArchivedEventStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
"""
Triggered by action: Archive channel
@ -84,7 +84,7 @@ class SlackChannelUnArchivedEventStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
"""
Triggered by action: UnArchive channel

6
engine/apps/slack/scenarios/slack_channel_integration.py
View file

@ -16,7 +16,7 @@ class SlackChannelMessageEventStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
"""
Triggered by action: Any new message in channel.
@ -38,7 +38,7 @@ class SlackChannelMessageEventStep(scenario_step.ScenarioStep):
self.delete_thread_message_from_resolution_note(slack_user_identity, payload)
def save_thread_message_for_resolution_note(
self, slack_user_identity: "SlackUserIdentity", payload: EventPayload.Any
self, slack_user_identity: "SlackUserIdentity", payload: EventPayload
) -> None:
from apps.alerts.models import ResolutionNoteSlackMessage
from apps.slack.models import SlackMessage
@ -125,7 +125,7 @@ class SlackChannelMessageEventStep(scenario_step.ScenarioStep):
slack_thread_message.save()
def delete_thread_message_from_resolution_note(
self, slack_user_identity: "SlackUserIdentity", payload: EventPayload.Any
self, slack_user_identity: "SlackUserIdentity", payload: EventPayload
) -> None:
from apps.alerts.models import ResolutionNoteSlackMessage

4
engine/apps/slack/scenarios/slack_usergroup.py
View file

@ -14,7 +14,7 @@ class SlackUserGroupEventStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
"""
Triggered by action: creation user groups or changes in user groups except its members.
@ -45,7 +45,7 @@ class SlackUserGroupMembersChangedEventStep(scenario_step.ScenarioStep):
self,
slack_user_identity: "SlackUserIdentity",
slack_team_identity: "SlackTeamIdentity",
payload: EventPayload.Any,
payload: EventPayload,
) -> None:
"""
Triggered by action: changed members in user group.

16
engine/apps/slack/scenarios/step_mixins.py
View file

@ -2,7 +2,7 @@ import json
import logging
from apps.alerts.models import AlertGroup
from apps.api.permissions import user_is_authorized
from apps.api.permissions import LegacyAccessControlCompatiblePermissions, user_is_authorized
from apps.slack.models import SlackMessage, SlackTeamIdentity
from apps.slack.types import EventPayload
from apps.user_management.models import User
@ -17,9 +17,9 @@ class AlertGroupActionsMixin:
user: User | None
REQUIRED_PERMISSIONS = []
REQUIRED_PERMISSIONS: LegacyAccessControlCompatiblePermissions = []
def get_alert_group(self, slack_team_identity: SlackTeamIdentity, payload: EventPayload.Any) -> AlertGroup:
def get_alert_group(self, slack_team_identity: SlackTeamIdentity, payload: EventPayload) -> AlertGroup:
"""
Get AlertGroup instance on Slack message button click or select menu change.
"""
@ -47,7 +47,7 @@ class AlertGroupActionsMixin:
and user_is_authorized(self.user, self.REQUIRED_PERMISSIONS)
)
def open_unauthorized_warning(self, payload: EventPayload.Any) -> None:
def open_unauthorized_warning(self, payload: EventPayload) -> None:
self.open_warning_window(
payload,
warning_text="You do not have permission to perform this action. Ask an admin to upgrade your permissions.",
@ -55,7 +55,7 @@ class AlertGroupActionsMixin:
)
def _repair_alert_group(
self, slack_team_identity: SlackTeamIdentity, alert_group: AlertGroup, payload: EventPayload.Any
self, slack_team_identity: SlackTeamIdentity, alert_group: AlertGroup, payload: EventPayload
) -> None:
"""
There's a possibility that OnCall failed to create a SlackMessage instance for an AlertGroup, but the message
@ -79,7 +79,7 @@ class AlertGroupActionsMixin:
alert_group.slack_message = slack_message
alert_group.save(update_fields=["slack_message"])
def _get_alert_group_from_action(self, payload: EventPayload.Any) -> AlertGroup | None:
def _get_alert_group_from_action(self, payload: EventPayload) -> AlertGroup | None:
"""
Get AlertGroup instance from action data in payload. Action data is data encoded into buttons and select
menus in apps.alerts.incident_appearance.renderers.slack_renderer.AlertGroupSlackRenderer._get_buttons_blocks.
@ -107,7 +107,7 @@ class AlertGroupActionsMixin:
return AlertGroup.objects.get(pk=alert_group_pk)
def _get_alert_group_from_message(self, payload: EventPayload.Any) -> AlertGroup | None:
def _get_alert_group_from_message(self, payload: EventPayload) -> AlertGroup | None:
"""
Get AlertGroup instance from message data in payload. It's similar to _get_alert_group_from_action,
but it tries to get alert_group_pk from ANY button in the message, not just the one that was clicked.
@ -139,7 +139,7 @@ class AlertGroupActionsMixin:
return None
def _get_alert_group_from_slack_message_in_db(
self, slack_team_identity: SlackTeamIdentity, payload: EventPayload.Any
self, slack_team_identity: SlackTeamIdentity, payload: EventPayload
) -> AlertGroup:
"""
Get AlertGroup instance from SlackMessage instance.

9
engine/apps/slack/types/__init__.py
View file

@ -1,6 +1,13 @@
from .blocks import Block # noqa: F401
from .common import EventType, MessageEventSubtype, PayloadType # noqa: F401
from .composition_objects import CompositionObjects # noqa: F401
from .composition_objects import ( # noqa: F401
CompositionObjectConfirm,
CompositionObjectMrkdwnText,
CompositionObjectOption,
CompositionObjectOptionGroup,
CompositionObjectPlainText,
CompositionObjectText,
)
from .interaction_payloads import EventPayload # noqa: F401
from .interaction_payloads.block_actions import BlockActionType # noqa: F401
from .interaction_payloads.interactive_messages import InteractiveMessageActionType # noqa: F401

23
engine/apps/slack/types/block_elements.py
View file

@ -5,7 +5,12 @@
import typing
from .common import Style
from .composition_objects import CompositionObjects
from .composition_objects import (
CompositionObjectConfirm,
CompositionObjectOption,
CompositionObjectPlainText,
CompositionObjectText,
)
class _BaseBlockElement(typing.TypedDict):
@ -31,7 +36,7 @@ class BlockElement:
The type of element. In this case `type` is always `button`.
"""
text: CompositionObjects.Text
text: CompositionObjectText
"""
A [text object](https://api.slack.com/reference/block-kit/composition-objects#text) that defines the button's text.
@ -63,20 +68,20 @@ class BlockElement:
The type of element. In this case `type` is always `checkboxes`.
"""
options: typing.List[CompositionObjects.Option]
options: typing.List[CompositionObjectOption]
"""
An array of [option objects](https://api.slack.com/reference/block-kit/composition-objects#option).
A maximum of 10 options are allowed.
"""
initial_options: typing.Optional[typing.List[CompositionObjects.Option]]
initial_options: typing.Optional[typing.List[CompositionObjectOption]]
"""
An array of [option objects](https://api.slack.com/reference/block-kit/composition-objects#option) that exactly
matches one or more of the options within `options`. These options will be selected when the checkbox group
initially loads.
"""
confirm: typing.Optional[CompositionObjects.Confirm]
confirm: typing.Optional[CompositionObjectConfirm]
"""
A [confirm object](https://api.slack.com/reference/block-kit/composition-objects#confirm) that defines an optional
confirmation dialog that appears after clicking one of the checkboxes in this element.
@ -106,7 +111,7 @@ class BlockElement:
The initial date that is selected when the element is loaded. This should be in the format `YYYY-MM-DD`.
"""
confirm: CompositionObjects.Confirm
confirm: CompositionObjectConfirm
"""
A [confirm object](https://api.slack.com/reference/block-kit/composition-objects#confirm) that defines an
optional confirmation dialog that appears after a menu item is selected.
@ -120,7 +125,7 @@ class BlockElement:
Only one element can be set to `true`. Defaults to `false`.
"""
placeholder: CompositionObjects.PlainText
placeholder: CompositionObjectPlainText
"""
A [plain_text only text object](https://api.slack.com/reference/block-kit/composition-objects#text) that
defines the placeholder text shown on the datepicker.
@ -171,13 +176,13 @@ class BlockElement:
The type of element. In this case `type` is always `overflow`.
"""
options: typing.List[CompositionObjects.Option]
options: typing.List[CompositionObjectOption]
"""
An array of up to five [option objects](https://api.slack.com/reference/block-kit/composition-objects#option)
to display in the menu.
"""
confirm: CompositionObjects.Confirm
confirm: CompositionObjectConfirm
"""
A [confirm object](https://api.slack.com/reference/block-kit/composition-objects#confirm) that defines an
optional confirmation dialog that appears after a menu item is selected.

16
engine/apps/slack/types/blocks.py
View file

@ -5,7 +5,7 @@
import typing
from .block_elements import BlockElement
from .composition_objects import CompositionObjects
from .composition_objects import CompositionObjectPlainText, CompositionObjectText
class Block:
@ -57,7 +57,7 @@ class Block:
The type of block. For a context block, `type` is always `context`.
"""
elements: typing.List[CompositionObjects.Text | BlockElement.Image]
elements: typing.List[CompositionObjectText | BlockElement.Image]
"""
An array of [image elements](https://api.slack.com/reference/messaging/block-elements#image) and
[text objects](https://api.slack.com/reference/messaging/composition-objects#text).
@ -91,7 +91,7 @@ class Block:
The type of block. For a header block, `type` is always `header`.
"""
text: CompositionObjects.Text
text: CompositionObjectText
"""
The text for the block, in the form of a [text object](https://api.slack.com/reference/block-kit/composition-objects#text).
@ -124,7 +124,7 @@ class Block:
Maximum length for this field is 2000 characters.
"""
title: CompositionObjects.PlainText
title: CompositionObjectPlainText
"""
An optional title for the image in the form of a
[text object](https://api.slack.com/reference/messaging/composition-objects#text) that can only be of
@ -146,7 +146,7 @@ class Block:
The type of block. For an input block, `type` is always `input`.
"""
label: CompositionObjects.PlainText
label: CompositionObjectPlainText
"""
A label that appears above an input element in the form of a
[text object](https://api.slack.com/reference/messaging/composition-objects#text) that must
@ -169,7 +169,7 @@ class Block:
Defaults to `false`.
"""
hint: CompositionObjects.PlainText
hint: CompositionObjectPlainText
"""
An optional hint that appears below an input element in a lighter grey.
@ -197,7 +197,7 @@ class Block:
The type of block. For a section block, `type` will always be `section`.
"""
text: CompositionObjects.Text
text: CompositionObjectText
"""
The text for the block, in the form of a [text object](https://api.slack.com/reference/block-kit/composition-objects#text).
@ -205,7 +205,7 @@ class Block:
This field is not required if a valid array of fields objects is provided instead.
"""
fields: typing.List[CompositionObjects.Text]
fields: typing.List[CompositionObjectText]
"""
Required if no `text` is provided.

42
engine/apps/slack/types/composition_objects.py
View file

@ -44,31 +44,31 @@ class _TextBase(typing.TypedDict):
"""
class _PlainText(_TextBase):
class CompositionObjectPlainText(_TextBase):
type: typing.Literal["plain_text"]
"""
The formatting to use for this text object.
"""
class _MrkdwnText(_TextBase):
class CompositionObjectMrkdwnText(_TextBase):
type: typing.Literal["mrkdwn"]
"""
The formatting to use for this text object.
"""
_Text = _PlainText | _MrkdwnText
CompositionObjectText = CompositionObjectPlainText | CompositionObjectMrkdwnText
class _Option(typing.TypedDict):
class CompositionObjectOption(typing.TypedDict):
"""
An object that represents a single selectable item in a select menu, multi-select menu, checkbox group, radio button group, or overflow menu.
[Documentation](https://api.slack.com/reference/block-kit/composition-objects#option)
"""
text: _Text
text: CompositionObjectText
"""
A [text object](https://api.slack.com/reference/block-kit/composition-objects#text) that defines the text shown in
the option on the menu.
@ -84,7 +84,7 @@ class _Option(typing.TypedDict):
Maximum length for this field is 75 characters.
"""
description: typing.Optional[_PlainText]
description: typing.Optional[CompositionObjectPlainText]
"""
A [plain_text-only text object](https://api.slack.com/reference/block-kit/composition-objects#confirm:~:text=A-,plain_text,%2Donly%20text%20object,-that%20defines%20the)
that defines a line of descriptive text shown below the `text` field beside the radio button.
@ -104,7 +104,7 @@ class _Option(typing.TypedDict):
"""
class _OptionGroup(typing.TypedDict):
class CompositionObjectOptionGroup(typing.TypedDict):
"""
Provides a way to group options in a [select menu](https://api.slack.com/reference/block-kit/block-elements#select)
or [multi-select menu](https://api.slack.com/reference/block-kit/block-elements#multi_select).
@ -112,7 +112,7 @@ class _OptionGroup(typing.TypedDict):
[Documentation](https://api.slack.com/reference/block-kit/composition-objects#option_group)
"""
label: _PlainText
label: CompositionObjectPlainText
"""
A [plain_text only text object](https://api.slack.com/reference/block-kit/composition-objects#text) that defines
the label shown above this group of options.
@ -120,7 +120,7 @@ class _OptionGroup(typing.TypedDict):
Maximum length for the `text` in this field is 75 characters.
"""
options: typing.List[_Option]
options: typing.List[CompositionObjectOption]
"""
An array of [option objects](https://api.slack.com/reference/block-kit/composition-objects#option) that belong to
this specific group.
@ -129,7 +129,7 @@ class _OptionGroup(typing.TypedDict):
"""
class _Confirm(typing.TypedDict):
class CompositionObjectConfirm(typing.TypedDict):
"""
An object that defines a dialog that provides a confirmation step to any interactive element.
This dialog will ask the user to confirm their action by offering a confirm and deny buttons.
@ -137,7 +137,7 @@ class _Confirm(typing.TypedDict):
[Documentation](https://api.slack.com/reference/block-kit/composition-objects#confirm)
"""
title: _PlainText
title: CompositionObjectPlainText
"""
A [plain_text-only text object](https://api.slack.com/reference/block-kit/composition-objects#confirm:~:text=A-,plain_text,%2Donly%20text%20object,-that%20defines%20the)
that defines the dialog's title.
@ -145,7 +145,7 @@ class _Confirm(typing.TypedDict):
Maximum length for this field is 100 characters.
"""
text: _PlainText
text: CompositionObjectPlainText
"""
A [plain_text-only text object](https://api.slack.com/reference/block-kit/composition-objects#confirm:~:text=A-,plain_text,%2Donly%20text%20object,-that%20defines%20the)
that defines the explanatory text that appears in the confirm dialog.
@ -153,7 +153,7 @@ class _Confirm(typing.TypedDict):
Maximum length for the text in this field is 300 characters.
"""
confirm: _PlainText
confirm: CompositionObjectPlainText
"""
A [plain_text-only text object](https://api.slack.com/reference/block-kit/composition-objects#confirm:~:text=A-,plain_text,%2Donly%20text%20object,-that%20defines%20the)
to define the text of the button that confirms the action.
@ -161,7 +161,7 @@ class _Confirm(typing.TypedDict):
Maximum length for the text in this field is 30 characters.
"""
deny: _PlainText
deny: CompositionObjectPlainText
"""
A [plain_text-only text object](https://api.slack.com/reference/block-kit/composition-objects#confirm:~:text=A-,plain_text,%2Donly%20text%20object,-that%20defines%20the)
to define the text of the button that cancels the action.
@ -178,17 +178,3 @@ class _Confirm(typing.TypedDict):
If this field is not provided, the default value will be `primary`.
"""
class CompositionObjects:
Confirm = _Confirm
MrkdwnText = _MrkdwnText
Option = _Option
OptionGroup = _OptionGroup
PlainText = _PlainText
Text = _Text
__all__ = [
"CompositionObjects",
]

25
engine/apps/slack/types/interaction_payloads/__init__.py
View file

@ -5,20 +5,11 @@ from .shortcuts import MessageActionPayload
from .slash_command import SlashCommandPayload
from .view_submission import ViewSubmissionPayload
class EventPayload:
BlockActionsPayload = BlockActionsPayload
DialogSubmissionPayload = DialogSubmissionPayload
InteractiveMessagesPayload = InteractiveMessagesPayload
MessageActionPayload = MessageActionPayload
SlashCommandPayload = SlashCommandPayload
ViewSubmissionPayload = ViewSubmissionPayload
Any = (
BlockActionsPayload
| DialogSubmissionPayload
| InteractiveMessagesPayload
| MessageActionPayload
| SlashCommandPayload
| ViewSubmissionPayload
)
EventPayload = (
BlockActionsPayload
| DialogSubmissionPayload
| InteractiveMessagesPayload
| MessageActionPayload
| SlashCommandPayload
| ViewSubmissionPayload
)

8
engine/apps/slack/types/views.py
View file

@ -1,7 +1,7 @@
import typing
from .blocks import Block
from .composition_objects import CompositionObjects
from .composition_objects import CompositionObjectPlainText
class ModalView(typing.TypedDict):
@ -14,7 +14,7 @@ class ModalView(typing.TypedDict):
Required. The type of view. Set to `modal` for modals.
"""
title: CompositionObjects.PlainText
title: CompositionObjectPlainText
"""
Required. The title that appears in the top-left of the modal.
@ -30,7 +30,7 @@ class ModalView(typing.TypedDict):
Max of 100 blocks.
"""
close: CompositionObjects.PlainText
close: CompositionObjectPlainText
"""
An optional [plain_text text element](https://api.slack.com/reference/block-kit/composition-objects#text) that
defines the text displayed in the close button at the bottom-right of the view.
@ -38,7 +38,7 @@ class ModalView(typing.TypedDict):
Max length of 24 characters.
"""
submit: CompositionObjects.PlainText
submit: CompositionObjectPlainText
"""
An optional [plain_text text element](https://api.slack.com/reference/block-kit/composition-objects#text) that
defines the text displayed in the submit button at the bottom-right of the view.

8
engine/apps/slack/views.py
View file

@ -405,7 +405,7 @@ class SlackEventApiEndpointView(APIView):
return Response(status=200)
@staticmethod
def _get_slack_team_identity_from_payload(payload: EventPayload.Any) -> SlackTeamIdentity | None:
def _get_slack_team_identity_from_payload(payload: EventPayload) -> SlackTeamIdentity | None:
def _slack_team_id() -> str | None:
with suppress(KeyError):
return payload["team"]["id"]
@ -422,7 +422,7 @@ class SlackEventApiEndpointView(APIView):
@staticmethod
def _get_organization_from_payload(
payload: EventPayload.Any, slack_team_identity: SlackTeamIdentity
payload: EventPayload, slack_team_identity: SlackTeamIdentity
) -> Organization | None:
"""
Extract organization from Slack payload.
@ -492,7 +492,7 @@ class SlackEventApiEndpointView(APIView):
return None
def _open_warning_window_if_needed(
self, payload: EventPayload.Any, slack_team_identity: SlackTeamIdentity, warning_text: str
self, payload: EventPayload, slack_team_identity: SlackTeamIdentity, warning_text: str
) -> None:
if payload.get("trigger_id") is not None:
step = ScenarioStep(slack_team_identity)
@ -504,7 +504,7 @@ class SlackEventApiEndpointView(APIView):
)
def _open_warning_for_unconnected_user(
self, slack_client: SlackClientWithErrorHandling, payload: EventPayload.Any
self, slack_client: SlackClientWithErrorHandling, payload: EventPayload
) -> None:
if payload.get("trigger_id") is None:
return

17
engine/apps/user_management/models/team.py
View file

@ -12,11 +12,12 @@ if typing.TYPE_CHECKING:
from django.db.models.manager import RelatedManager
from apps.alerts.models import AlertGroupLogRecord
from apps.grafana_plugin.helpers.client import GrafanaAPIClient
from apps.schedules.models import CustomOnCallShift
from apps.user_management.models import User
from apps.user_management.models import Organization, User
def generate_public_primary_key_for_team():
def generate_public_primary_key_for_team() -> str:
prefix = "T"
new_public_primary_key = generate_public_primary_key(prefix)
@ -30,11 +31,13 @@ def generate_public_primary_key_for_team():
return new_public_primary_key
class TeamManager(models.Manager):
class TeamManager(models.Manager["Team"]):
@staticmethod
def sync_for_organization(organization, api_teams: list[dict]):
def sync_for_organization(
organization: "Organization", api_teams: typing.List["GrafanaAPIClient.Types.GrafanaTeam"]
) -> None:
grafana_teams = {team["id"]: team for team in api_teams}
existing_team_ids = set(organization.teams.all().values_list("team_id", flat=True))
existing_team_ids: typing.Set[int] = set(organization.teams.all().values_list("team_id", flat=True))
# create missing teams
teams_to_create = tuple(
@ -55,7 +58,7 @@ class TeamManager(models.Manager):
organization.teams.filter(team_id__in=team_ids_to_delete).delete()
# collect teams diffs to update metrics cache
metrics_teams_to_update = {}
metrics_teams_to_update: MetricsCacheManager.TeamsDiffMap = {}
for team_id in team_ids_to_delete:
metrics_teams_to_update = MetricsCacheManager.update_team_diff(
metrics_teams_to_update, team_id, deleted=True
@ -97,7 +100,7 @@ class Team(models.Model):
default=generate_public_primary_key_for_team,
)
objects: models.Manager["Team"] = TeamManager()
objects = TeamManager()
team_id = models.PositiveIntegerField()
organization = models.ForeignKey(

4
engine/apps/user_management/models/user.py
View file

@ -66,7 +66,7 @@ def default_working_hours():
return working_hours
class UserManager(models.Manager):
class UserManager(models.Manager["User"]):
@staticmethod
def sync_for_team(team, api_members: list[dict]):
user_ids = tuple(member["userId"] for member in api_members)
@ -161,7 +161,7 @@ class User(models.Model):
user_schedule_export_token: "RelatedManager['UserScheduleExportAuthToken']"
wiped_alert_groups: "RelatedManager['AlertGroup']"
objects: models.Manager["User"] = UserManager.from_queryset(UserQuerySet)()
objects = UserManager.from_queryset(UserQuerySet)()
class Meta:
# For some reason there are cases when Grafana user gets deleted,

20
engine/apps/user_management/sync.py
View file

@ -12,7 +12,7 @@ logger = get_task_logger(__name__)
logger.setLevel(logging.DEBUG)
def sync_organization(organization):
def sync_organization(organization: Organization) -> None:
grafana_api_client = GrafanaAPIClient(api_url=organization.grafana_url, api_token=organization.api_token)
# NOTE: checking whether or not RBAC is enabled depends on whether we are dealing with an open-source or cloud
@ -59,7 +59,7 @@ def sync_organization(organization):
org_sync_signal.send(sender=None, organization=organization)
def _sync_instance_info(organization):
def _sync_instance_info(organization: Organization) -> None:
if organization.gcom_token:
gcom_client = GcomAPIClient(organization.gcom_token)
instance_info = gcom_client.get_instance_info(organization.stack_id)
@ -76,13 +76,13 @@ def _sync_instance_info(organization):
organization.gcom_token_org_last_time_synced = timezone.now()
def sync_users_and_teams(client: GrafanaAPIClient, organization):
def sync_users_and_teams(client: GrafanaAPIClient, organization: Organization) -> None:
sync_users(client, organization)
sync_teams(client, organization)
sync_team_members(client, organization)
def sync_users(client: GrafanaAPIClient, organization, **kwargs):
def sync_users(client: GrafanaAPIClient, organization: Organization, **kwargs) -> None:
api_users = client.get_users(organization.is_rbac_permissions_enabled, **kwargs)
# check if api_users are shaped correctly. e.g. for paused instance, the response is not a list.
if not api_users or not isinstance(api_users, (tuple, list)):
@ -90,7 +90,7 @@ def sync_users(client: GrafanaAPIClient, organization, **kwargs):
User.objects.sync_for_organization(organization=organization, api_users=api_users)
def sync_teams(client: GrafanaAPIClient, organization, **kwargs):
def sync_teams(client: GrafanaAPIClient, organization: Organization, **kwargs) -> None:
api_teams_result, _ = client.get_teams(**kwargs)
if not api_teams_result:
return
@ -98,7 +98,7 @@ def sync_teams(client: GrafanaAPIClient, organization, **kwargs):
Team.objects.sync_for_organization(organization=organization, api_teams=api_teams)
def sync_team_members(client: GrafanaAPIClient, organization):
def sync_team_members(client: GrafanaAPIClient, organization: Organization) -> None:
for team in organization.teams.all():
members, _ = client.get_team_members(team.team_id)
if not members:
@ -106,7 +106,7 @@ def sync_team_members(client: GrafanaAPIClient, organization):
User.objects.sync_for_team(team=team, api_members=members)
def sync_users_for_teams(client: GrafanaAPIClient, organization, **kwargs):
def sync_users_for_teams(client: GrafanaAPIClient, organization: Organization, **kwargs) -> None:
api_teams_result, _ = client.get_teams(**kwargs)
if not api_teams_result:
return
@ -114,7 +114,7 @@ def sync_users_for_teams(client: GrafanaAPIClient, organization, **kwargs):
Team.objects.sync_for_organization(organization=organization, api_teams=api_teams)
def check_grafana_incident_is_enabled(client):
def check_grafana_incident_is_enabled(client: GrafanaAPIClient) -> bool:
GRAFANA_INCIDENT_PLUGIN = "grafana-incident-app"
grafana_incident_settings, _ = client.get_grafana_plugin_settings(GRAFANA_INCIDENT_PLUGIN)
is_grafana_incident_enabled = False
@ -123,7 +123,7 @@ def check_grafana_incident_is_enabled(client):
return is_grafana_incident_enabled
def delete_organization_if_needed(organization):
def delete_organization_if_needed(organization: Organization) -> bool:
# Organization has a manually set API token, it will not be found within GCOM
# and would need to be deleted manually.
from apps.auth_token.models import PluginAuthToken
@ -143,7 +143,7 @@ def delete_organization_if_needed(organization):
return True
def cleanup_organization(organization_pk):
def cleanup_organization(organization_pk: int) -> None:
logger.info(f"Start cleanup Organization {organization_pk}")
try:
organization = Organization.objects.get(pk=organization_pk)

13
engine/common/ordered_model/ordered_model.py
View file

@ -29,9 +29,6 @@ def _retry(exc: typing.Type[Exception] | tuple[typing.Type[Exception], ...], max
return _retry_with_params
Self = typing.TypeVar("Self", bound="OrderedModel")
class OrderedModel(models.Model):
"""
This class is intended to be used as a mixin for models that need to be ordered.
@ -95,7 +92,7 @@ class OrderedModel(models.Model):
"""
with transaction.atomic():
instances = self._lock_ordering_queryset() # lock ordering queryset to prevent reading inconsistent data
max_order = max(instance.order for instance in instances) if instances else -1
max_order = max(typing.cast(int, instance.order) for instance in instances) if instances else -1
self.order = max_order + 1
super().save(*args, **kwargs)
@ -137,7 +134,7 @@ class OrderedModel(models.Model):
order = instances[index].order # get order of the instance at the given index
self._move_instances_to_order(instances, order)
def _move_instances_to_order(self, instances: list[Self], order: int) -> None:
def _move_instances_to_order(self, instances: list[typing.Self], order: int) -> None:
"""
Helper method for moving self to a given order, adjusting other instances' orders if necessary.
Must be called within a transaction that locks the ordering queryset.
@ -230,7 +227,7 @@ class OrderedModel(models.Model):
self.order, other.order = other.order, self.order
self._manager.filter(pk__in=[self.pk, other.pk]).bulk_update([self, other], fields=["order"])
def next(self) -> Self | None:
def next(self) -> typing.Self | None:
"""
Return the next instance in the ordering queryset, or None if there's no next instance.
Example:
@ -253,10 +250,10 @@ class OrderedModel(models.Model):
if value is None or not isinstance(value, int) or value < 0:
raise ValueError("Value must be a positive integer.")
def _get_ordering_queryset(self) -> models.QuerySet[Self]:
def _get_ordering_queryset(self) -> models.QuerySet[typing.Self]:
return self._manager.filter(**self._ordering_params)
def _lock_ordering_queryset(self) -> list[Self]:
def _lock_ordering_queryset(self) -> list[typing.Self]:
"""
Locks the ordering queryset with SELECT FOR UPDATE and returns the queryset as a list.
This allows to prevent concurrent updates from different transactions.

50
engine/common/public_primary_keys.py
View file

@ -7,48 +7,13 @@ from django.utils.crypto import get_random_string
logger = logging.getLogger(__name__)
def generate_public_primary_key(prefix, length=settings.PUBLIC_PRIMARY_KEY_MIN_LENGTH):
"""It generates random string with prefix and length
:param prefix:
"U": ("user_management", "User"),
"O": ("user_management", "Organization"),
"T": ("user_management", "Team"),
"N": ("base", "UserNotificationPolicy"),
"C": ("alerts", "AlertReceiveChannel"),
"R": ("alerts", "ChannelFilter"),
"S": ("schedules", "OnCallSchedule"),
"E": ("alerts", "EscalationPolicy"),
"F": ("alerts", "EscalationChain"),
"I": ("alerts", "AlertGroup"),
"A": ("alerts", "Alert"),
"M": ("alerts", "ResolutionNote"),
"G": ("slack", "SlackUserGroup"),
"K": ("alerts", "CustomButton"),
"O": ("schedules", "CustomOnCallShift"),
"B": ("heartbeat", "IntegrationHeartBeat"),
"H": ("slack", "SlackChannel"),
"Z": ("telegram", "TelegramToOrganizationConnector"),
"L": ("base", "LiveSetting"),
"X": ("extensions", "Other models from extensions apps"),
:param length:
:return:
"""
def generate_public_primary_key(prefix: str, length: int = settings.PUBLIC_PRIMARY_KEY_MIN_LENGTH) -> str:
return prefix + get_random_string(length=length, allowed_chars=settings.PUBLIC_PRIMARY_KEY_ALLOWED_CHARS)
def increase_public_primary_key_length(failure_counter, prefix, model_name, max_attempt_count=5):
"""
Another yet helper which generates random string with larger length
when previous public_primary_key exists
:param failure_counter:
:param prefix:
:param model_name:
:param max_attempt_count: When attempt count is more then max_attempt_count we'll get the exception
:return:
"""
def increase_public_primary_key_length(
failure_counter: int, prefix: str, model_name: str, max_attempt_count: int = 5
) -> str:
if failure_counter < max_attempt_count:
logger.warning(
f"Let's try increase a {model_name} "
@ -59,7 +24,6 @@ def increase_public_primary_key_length(failure_counter, prefix, model_name, max_
return generate_public_primary_key(
prefix=prefix, length=settings.PUBLIC_PRIMARY_KEY_MIN_LENGTH + failure_counter
)
else:
raise FieldError(
f"A count of {model_name} new_public_primary_key generation " f"attempts is more than {max_attempt_count}!"
)
raise FieldError(
f"A count of {model_name} new_public_primary_key generation " f"attempts is more than {max_attempt_count}!"
)

4
engine/pyproject.toml
View file

@ -34,13 +34,10 @@ disable_error_code = [
"index",
"misc",
"name-defined",
"no-redef",
"operator",
"return-value",
"typeddict-item",
"union-attr",
"valid-type",
"var-annotated",
]
# mypy per-module options
@ -55,6 +52,7 @@ module = [
"celery.utils.debug",
"debug_toolbar.*",
"django_deprecate_fields.*",
"django_migration_linter",
"django_sns_view.*",
"factory.*",
"fcm_django.*",