20
20
)
21
21
from apps .alerts .utils import is_declare_incident_step_enabled
22
22
from apps .schedules .ical_utils import list_users_to_notify_from_ical
23
- from apps .user_management .models import User
24
23
25
24
if typing .TYPE_CHECKING :
26
25
from apps .alerts .models .alert_group import AlertGroup
26
+ from apps .schedules .models import OnCallSchedule
27
+ from apps .slack .models import SlackUserGroup
28
+ from apps .user_management .models import Team , User
29
+ from apps .webhooks .models import Webhook
27
30
28
31
29
32
class EscalationPolicySnapshot :
@@ -57,24 +60,24 @@ class EscalationPolicySnapshot:
57
60
58
61
def __init__ (
59
62
self ,
60
- id ,
61
- order ,
62
- step ,
63
- wait_delay ,
64
- notify_to_users_queue ,
65
- last_notified_user ,
66
- from_time ,
67
- to_time ,
68
- num_alerts_in_window ,
69
- num_minutes_in_window ,
70
- custom_webhook ,
71
- notify_schedule ,
72
- notify_to_group ,
73
- escalation_counter ,
74
- passed_last_time ,
75
- pause_escalation ,
76
- notify_to_team_members = None ,
77
- severity = None ,
63
+ id : int ,
64
+ order : int ,
65
+ step : int ,
66
+ wait_delay : typing . Optional [ datetime . timedelta ] ,
67
+ notify_to_users_queue : typing . Optional [ typing . Sequence [ "User" ]] ,
68
+ last_notified_user : typing . Optional [ "User" ] ,
69
+ from_time : typing . Optional [ datetime . time ] ,
70
+ to_time : typing . Optional [ datetime . time ] ,
71
+ num_alerts_in_window : typing . Optional [ int ] ,
72
+ num_minutes_in_window : typing . Optional [ int ] ,
73
+ custom_webhook : typing . Optional [ "Webhook" ] ,
74
+ notify_schedule : typing . Optional [ "OnCallSchedule" ] ,
75
+ notify_to_group : typing . Optional [ "SlackUserGroup" ] ,
76
+ escalation_counter : int ,
77
+ passed_last_time : typing . Optional [ datetime . datetime ] ,
78
+ pause_escalation : bool ,
79
+ notify_to_team_members : typing . Optional [ "Team" ] = None ,
80
+ severity : typing . Optional [ str ] = None ,
78
81
):
79
82
self .id = id
80
83
self .order = order
@@ -107,11 +110,11 @@ def escalation_policy(self) -> typing.Optional[EscalationPolicy]:
107
110
return EscalationPolicy .objects .filter (pk = self .id ).first ()
108
111
109
112
@property
110
- def sorted_users_queue (self ) -> typing .List [User ]:
113
+ def sorted_users_queue (self ) -> typing .List [" User" ]:
111
114
return sorted (self .notify_to_users_queue , key = lambda user : (user .username or "" , user .pk ))
112
115
113
116
@property
114
- def next_user_in_sorted_queue (self ) -> User :
117
+ def next_user_in_sorted_queue (self ) -> " User" :
115
118
users_queue = self .sorted_users_queue
116
119
try :
117
120
last_user_index = users_queue .index (self .last_notified_user )
@@ -120,7 +123,7 @@ def next_user_in_sorted_queue(self) -> User:
120
123
next_user = users_queue [(last_user_index + 1 ) % len (users_queue )]
121
124
return next_user
122
125
123
- def execute (self , alert_group : "AlertGroup" , reason ) -> StepExecutionResultData :
126
+ def execute (self , alert_group : "AlertGroup" , reason : str ) -> StepExecutionResultData :
124
127
action_map : typing .Dict [typing .Optional [int ], EscalationPolicySnapshot .StepExecutionFunc ] = {
125
128
EscalationPolicy .STEP_WAIT : self ._escalation_step_wait ,
126
129
EscalationPolicy .STEP_FINAL_NOTIFYALL : self ._escalation_step_notify_all ,
0 commit comments