| Total Complexity | 6 |
| Total Lines | 26 |
| Duplicated Lines | 0 % |
| Changes | 0 | ||
| 1 | import pygerduty |
||
| 6 | class PagerDutyAction(Action): |
||
| 7 | def __init__(self, config): |
||
| 8 | super(PagerDutyAction, self).__init__(config) |
||
| 9 | self.pager = self._init_client() |
||
| 10 | self.trigger = [] |
||
| 11 | |||
| 12 | def _init_client(self): |
||
| 13 | api_key = self.config['api_key'] |
||
| 14 | # service_api = self.config['service_api'] |
||
| 15 | subdomain = self.config['subdomain'] |
||
| 16 | pager = pygerduty.PagerDuty(subdomain, api_token=api_key) |
||
| 17 | return pager |
||
| 18 | |||
| 19 | # get all the acknowledged incidents |
||
| 20 | def get_ack_incidents(self): |
||
| 21 | ack_alarms = [] |
||
| 22 | for incident in self.pager.incidents.list(status="acknowledged"): |
||
| 23 | ack_alarms.append(incident.incident_key) |
||
| 24 | return ack_alarms |
||
| 25 | |||
| 26 | # get all the triggered incidents |
||
| 27 | def get_triggered_incidents(self): |
||
| 28 | trigger_alarms = [] |
||
| 29 | for incident in self.pager.incidents.list(status="triggered"): |
||
| 30 | trigger_alarms.append(incident.incident_key) |
||
| 31 | return trigger_alarms |
||
| 32 |