From d6140cbe8d866861fc3593f9b4ac71c3f593c3e0 Mon Sep 17 00:00:00 2001 From: Joey Orlando Date: Thu, 3 Aug 2023 11:43:03 +0200 Subject: [PATCH] Re-enable a few mypy rules + fix existing errors (#2725) # 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) --- engine/apps/alerts/models/alert.py | 8 +- .../alerts/models/alert_group_log_record.py | 4 +- engine/apps/alerts/models/invitation.py | 8 +- engine/apps/alerts/paging.py | 32 +++++-- engine/apps/api/serializers/channel_filter.py | 4 +- engine/apps/grafana_plugin/helpers/client.py | 41 ++++++-- engine/apps/metrics_exporter/helpers.py | 20 ++-- .../metrics_exporter/metrics_cache_manager.py | 17 +++- .../tests/test_update_metics_cache.py | 4 +- engine/apps/mobile_app/tasks.py | 8 +- .../phone_notifications/phone_provider.py | 10 +- engine/apps/schedules/ical_utils.py | 4 +- .../apps/schedules/models/on_call_schedule.py | 10 +- .../slack/scenarios/alertgroup_appearance.py | 4 +- .../apps/slack/scenarios/declare_incident.py | 2 +- .../apps/slack/scenarios/distribute_alerts.py | 32 +++---- .../slack/scenarios/invited_to_channel.py | 2 +- .../apps/slack/scenarios/manage_responders.py | 16 ++-- .../apps/slack/scenarios/manual_incident.py | 34 +++---- .../scenarios/notified_user_not_in_channel.py | 2 +- engine/apps/slack/scenarios/onboarding.py | 4 +- engine/apps/slack/scenarios/paging.py | 96 ++++++++++--------- engine/apps/slack/scenarios/profile_update.py | 2 +- .../apps/slack/scenarios/resolution_note.py | 96 ++++++++++--------- engine/apps/slack/scenarios/scenario_step.py | 3 +- engine/apps/slack/scenarios/schedules.py | 14 +-- .../slack/scenarios/shift_swap_requests.py | 2 +- engine/apps/slack/scenarios/slack_channel.py | 8 +- .../scenarios/slack_channel_integration.py | 6 +- .../apps/slack/scenarios/slack_usergroup.py | 4 +- engine/apps/slack/scenarios/step_mixins.py | 16 ++-- engine/apps/slack/types/__init__.py | 9 +- engine/apps/slack/types/block_elements.py | 23 +++-- engine/apps/slack/types/blocks.py | 16 ++-- .../apps/slack/types/composition_objects.py | 42 +++----- .../types/interaction_payloads/__init__.py | 25 ++--- engine/apps/slack/types/views.py | 8 +- engine/apps/slack/views.py | 8 +- engine/apps/user_management/models/team.py | 17 ++-- engine/apps/user_management/models/user.py | 4 +- engine/apps/user_management/sync.py | 20 ++-- engine/common/ordered_model/ordered_model.py | 13 +-- engine/common/public_primary_keys.py | 50 ++-------- engine/pyproject.toml | 4 +- 44 files changed, 380 insertions(+), 372 deletions(-) diff --git a/engine/apps/alerts/models/alert.py b/engine/apps/alerts/models/alert.py index dce585a2..50efdd84 100644 --- a/engine/apps/alerts/models/alert.py +++ b/engine/apps/alerts/models/alert.py @@ -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 diff --git a/engine/apps/alerts/models/alert_group_log_record.py b/engine/apps/alerts/models/alert_group_log_record.py index fedcbf6d..194afc53 100644 --- a/engine/apps/alerts/models/alert_group_log_record.py +++ b/engine/apps/alerts/models/alert_group_log_record.py @@ -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) diff --git a/engine/apps/alerts/models/invitation.py b/engine/apps/alerts/models/invitation.py index e24455e4..afc26cce 100644 --- a/engine/apps/alerts/models/invitation.py +++ b/engine/apps/alerts/models/invitation.py @@ -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,)) diff --git a/engine/apps/alerts/paging.py b/engine/apps/alerts/paging.py index 0d275387..a4fbb27c 100644 --- a/engine/apps/alerts/paging.py +++ b/engine/apps/alerts/paging.py @@ -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, diff --git a/engine/apps/api/serializers/channel_filter.py b/engine/apps/api/serializers/channel_filter.py index 16f257a2..17e92bbe 100644 --- a/engine/apps/api/serializers/channel_filter.py +++ b/engine/apps/api/serializers/channel_filter.py @@ -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: diff --git a/engine/apps/grafana_plugin/helpers/client.py b/engine/apps/grafana_plugin/helpers/client.py index 064982c2..4058c6b3 100644 --- a/engine/apps/grafana_plugin/helpers/client.py +++ b/engine/apps/grafana_plugin/helpers/client.py @@ -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: diff --git a/engine/apps/metrics_exporter/helpers.py b/engine/apps/metrics_exporter/helpers.py index 7f0bf120..1900c37c 100644 --- a/engine/apps/metrics_exporter/helpers.py +++ b/engine/apps/metrics_exporter/helpers.py @@ -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) diff --git a/engine/apps/metrics_exporter/metrics_cache_manager.py b/engine/apps/metrics_exporter/metrics_cache_manager.py index 1b6f4a0e..42d3a0cb 100644 --- a/engine/apps/metrics_exporter/metrics_cache_manager.py +++ b/engine/apps/metrics_exporter/metrics_cache_manager.py @@ -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 diff --git a/engine/apps/metrics_exporter/tests/test_update_metics_cache.py b/engine/apps/metrics_exporter/tests/test_update_metics_cache.py index 313f08b8..f85a74eb 100644 --- a/engine/apps/metrics_exporter/tests/test_update_metics_cache.py +++ b/engine/apps/metrics_exporter/tests/test_update_metics_cache.py @@ -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) diff --git a/engine/apps/mobile_app/tasks.py b/engine/apps/mobile_app/tasks.py index a4cd9889..df619e2e 100644 --- a/engine/apps/mobile_app/tasks.py +++ b/engine/apps/mobile_app/tasks.py @@ -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", diff --git a/engine/apps/phone_notifications/phone_provider.py b/engine/apps/phone_notifications/phone_provider.py index 1bff1397..63304a60 100644 --- a/engine/apps/phone_notifications/phone_provider.py +++ b/engine/apps/phone_notifications/phone_provider.py @@ -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: diff --git a/engine/apps/schedules/ical_utils.py b/engine/apps/schedules/ical_utils.py index 0bedb986..0324a462 100644 --- a/engine/apps/schedules/ical_utils.py +++ b/engine/apps/schedules/ical_utils.py @@ -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 diff --git a/engine/apps/schedules/models/on_call_schedule.py b/engine/apps/schedules/models/on_call_schedule.py index a8fc2a49..5f5c49f7 100644 --- a/engine/apps/schedules/models/on_call_schedule.py +++ b/engine/apps/schedules/models/on_call_schedule.py @@ -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): diff --git a/engine/apps/slack/scenarios/alertgroup_appearance.py b/engine/apps/slack/scenarios/alertgroup_appearance.py index 67e7dcdd..af1899a1 100644 --- a/engine/apps/slack/scenarios/alertgroup_appearance.py +++ b/engine/apps/slack/scenarios/alertgroup_appearance.py @@ -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 diff --git a/engine/apps/slack/scenarios/declare_incident.py b/engine/apps/slack/scenarios/declare_incident.py index 0f1286aa..7a7589df 100644 --- a/engine/apps/slack/scenarios/declare_incident.py +++ b/engine/apps/slack/scenarios/declare_incident.py @@ -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. diff --git a/engine/apps/slack/scenarios/distribute_alerts.py b/engine/apps/slack/scenarios/distribute_alerts.py index 27ed6636..b23a0920 100644 --- a/engine/apps/slack/scenarios/distribute_alerts.py +++ b/engine/apps/slack/scenarios/distribute_alerts.py @@ -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 diff --git a/engine/apps/slack/scenarios/invited_to_channel.py b/engine/apps/slack/scenarios/invited_to_channel.py index d111beba..e5119bb9 100644 --- a/engine/apps/slack/scenarios/invited_to_channel.py +++ b/engine/apps/slack/scenarios/invited_to_channel.py @@ -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"] diff --git a/engine/apps/slack/scenarios/manage_responders.py b/engine/apps/slack/scenarios/manage_responders.py index fc7187ac..94dee927 100644 --- a/engine/apps/slack/scenarios/manage_responders.py +++ b/engine/apps/slack/scenarios/manage_responders.py @@ -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] diff --git a/engine/apps/slack/scenarios/manual_incident.py b/engine/apps/slack/scenarios/manual_incident.py index c7625bc2..fa78b0d5 100644 --- a/engine/apps/slack/scenarios/manual_incident.py +++ b/engine/apps/slack/scenarios/manual_incident.py @@ -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() diff --git a/engine/apps/slack/scenarios/notified_user_not_in_channel.py b/engine/apps/slack/scenarios/notified_user_not_in_channel.py index ca4a4d6d..d3f045d4 100644 --- a/engine/apps/slack/scenarios/notified_user_not_in_channel.py +++ b/engine/apps/slack/scenarios/notified_user_not_in_channel.py @@ -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 diff --git a/engine/apps/slack/scenarios/onboarding.py b/engine/apps/slack/scenarios/onboarding.py index 163ed9d2..d84dc1ef 100644 --- a/engine/apps/slack/scenarios/onboarding.py +++ b/engine/apps/slack/scenarios/onboarding.py @@ -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 diff --git a/engine/apps/slack/scenarios/paging.py b/engine/apps/slack/scenarios/paging.py index bf541c81..65440394 100644 --- a/engine/apps/slack/scenarios/paging.py +++ b/engine/apps/slack/scenarios/paging.py @@ -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") diff --git a/engine/apps/slack/scenarios/profile_update.py b/engine/apps/slack/scenarios/profile_update.py index 07c7b453..3c46d9ba 100644 --- a/engine/apps/slack/scenarios/profile_update.py +++ b/engine/apps/slack/scenarios/profile_update.py @@ -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. diff --git a/engine/apps/slack/scenarios/resolution_note.py b/engine/apps/slack/scenarios/resolution_note.py index 4e6bd23a..c43bf17d 100644 --- a/engine/apps/slack/scenarios/resolution_note.py +++ b/engine/apps/slack/scenarios/resolution_note.py @@ -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": "{} (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": "{} (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 diff --git a/engine/apps/slack/scenarios/scenario_step.py b/engine/apps/slack/scenarios/scenario_step.py index 02882807..b474d1b2 100644 --- a/engine/apps/slack/scenarios/scenario_step.py +++ b/engine/apps/slack/scenarios/scenario_step.py @@ -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)) diff --git a/engine/apps/slack/scenarios/schedules.py b/engine/apps/slack/scenarios/schedules.py index 8224135d..84306683 100644 --- a/engine/apps/slack/scenarios/schedules.py +++ b/engine/apps/slack/scenarios/schedules.py @@ -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}", diff --git a/engine/apps/slack/scenarios/shift_swap_requests.py b/engine/apps/slack/scenarios/shift_swap_requests.py index 5422f974..7aff33cd 100644 --- a/engine/apps/slack/scenarios/shift_swap_requests.py +++ b/engine/apps/slack/scenarios/shift_swap_requests.py @@ -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 diff --git a/engine/apps/slack/scenarios/slack_channel.py b/engine/apps/slack/scenarios/slack_channel.py index cd1e01bf..882815e1 100644 --- a/engine/apps/slack/scenarios/slack_channel.py +++ b/engine/apps/slack/scenarios/slack_channel.py @@ -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 diff --git a/engine/apps/slack/scenarios/slack_channel_integration.py b/engine/apps/slack/scenarios/slack_channel_integration.py index d89bbb28..8e1cb787 100644 --- a/engine/apps/slack/scenarios/slack_channel_integration.py +++ b/engine/apps/slack/scenarios/slack_channel_integration.py @@ -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 diff --git a/engine/apps/slack/scenarios/slack_usergroup.py b/engine/apps/slack/scenarios/slack_usergroup.py index 79be82c5..865ec688 100644 --- a/engine/apps/slack/scenarios/slack_usergroup.py +++ b/engine/apps/slack/scenarios/slack_usergroup.py @@ -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. diff --git a/engine/apps/slack/scenarios/step_mixins.py b/engine/apps/slack/scenarios/step_mixins.py index 85c8aa9b..fb366338 100644 --- a/engine/apps/slack/scenarios/step_mixins.py +++ b/engine/apps/slack/scenarios/step_mixins.py @@ -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. diff --git a/engine/apps/slack/types/__init__.py b/engine/apps/slack/types/__init__.py index 81ebea1d..372e3965 100644 --- a/engine/apps/slack/types/__init__.py +++ b/engine/apps/slack/types/__init__.py @@ -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 diff --git a/engine/apps/slack/types/block_elements.py b/engine/apps/slack/types/block_elements.py index 9505ac2d..a039ce97 100644 --- a/engine/apps/slack/types/block_elements.py +++ b/engine/apps/slack/types/block_elements.py @@ -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. diff --git a/engine/apps/slack/types/blocks.py b/engine/apps/slack/types/blocks.py index 409b00ae..10973bc2 100644 --- a/engine/apps/slack/types/blocks.py +++ b/engine/apps/slack/types/blocks.py @@ -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. diff --git a/engine/apps/slack/types/composition_objects.py b/engine/apps/slack/types/composition_objects.py index a0a8864f..7a439238 100644 --- a/engine/apps/slack/types/composition_objects.py +++ b/engine/apps/slack/types/composition_objects.py @@ -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", -] diff --git a/engine/apps/slack/types/interaction_payloads/__init__.py b/engine/apps/slack/types/interaction_payloads/__init__.py index 67eed4b4..f5683c55 100644 --- a/engine/apps/slack/types/interaction_payloads/__init__.py +++ b/engine/apps/slack/types/interaction_payloads/__init__.py @@ -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 +) diff --git a/engine/apps/slack/types/views.py b/engine/apps/slack/types/views.py index 7f881a22..75225980 100644 --- a/engine/apps/slack/types/views.py +++ b/engine/apps/slack/types/views.py @@ -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. diff --git a/engine/apps/slack/views.py b/engine/apps/slack/views.py index 8d722254..e63df98e 100644 --- a/engine/apps/slack/views.py +++ b/engine/apps/slack/views.py @@ -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 diff --git a/engine/apps/user_management/models/team.py b/engine/apps/user_management/models/team.py index 06099e40..4adf2988 100644 --- a/engine/apps/user_management/models/team.py +++ b/engine/apps/user_management/models/team.py @@ -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( diff --git a/engine/apps/user_management/models/user.py b/engine/apps/user_management/models/user.py index f92571e8..e3969431 100644 --- a/engine/apps/user_management/models/user.py +++ b/engine/apps/user_management/models/user.py @@ -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, diff --git a/engine/apps/user_management/sync.py b/engine/apps/user_management/sync.py index 4f5b754f..ebbcf6fd 100644 --- a/engine/apps/user_management/sync.py +++ b/engine/apps/user_management/sync.py @@ -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) diff --git a/engine/common/ordered_model/ordered_model.py b/engine/common/ordered_model/ordered_model.py index 0fd980e2..6733af2d 100644 --- a/engine/common/ordered_model/ordered_model.py +++ b/engine/common/ordered_model/ordered_model.py @@ -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. diff --git a/engine/common/public_primary_keys.py b/engine/common/public_primary_keys.py index eecf3db3..eb5a6c8f 100644 --- a/engine/common/public_primary_keys.py +++ b/engine/common/public_primary_keys.py @@ -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}!" + ) diff --git a/engine/pyproject.toml b/engine/pyproject.toml index 0cdc724a..d1b94fde 100644 --- a/engine/pyproject.toml +++ b/engine/pyproject.toml @@ -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.*",