chore: run pre-comit

alerta/database/backends/postgres/base.py

@@ -462,7 +462,7 @@ def get_escalate(self):
             FROM alerts
             WHERE status='open' AND last_receive_time < %(etime)s
         """
-        return self._fetchall(select, {"etime": datetime.utcnow() - timedelta(minutes=current_app.config['ESCALATE_TIME'])}, limit=1000)
+        return self._fetchall(select, {'etime': datetime.utcnow() - timedelta(minutes=current_app.config['ESCALATE_TIME'])}, limit=1000)
 
     def get_alert_history(self, alert, page=None, page_size=None):
         select = """
@@ -1076,7 +1076,7 @@ def get_notification_rules_reactivate(self, time):
             AND reactivate IS NOT NULL
             AND reactivate < %(time)s
         """
-        return self._fetchall(select, {"time": time})
+        return self._fetchall(select, {'time': time})
 
     def get_notification_rules_active_status(self, alert, status):
         select = """
@@ -1097,7 +1097,7 @@ def get_notification_rules_active_status(self, alert, status):
         """
         if current_app.config['CUSTOMER_VIEWS']:
             select += ' AND (customer IS NULL OR customer=%(customer)s)'
-        return self._fetchall(select, {**vars(alert), "status": status})
+        return self._fetchall(select, {**vars(alert), 'status': status})
 
     def update_notification_rule(self, id, **kwargs):
         update = """
@@ -1344,7 +1344,7 @@ def get_escalation_alerts(self):
                 AND (e.use_advanced_severity=TRUE OR e.severity='{}' OR ARRAY[a.severity] <@ e.severity)
                 AND (e.use_advanced_severity=FALSE OR ((e.advanced_severity[s].from_='{}' OR ARRAY[a.previous_severity] <@ e.advanced_severity[s].from_) AND (e.advanced_severity[s].to='{}' OR ARRAY[a.severity] <@ e.advanced_severity[s].to)))
         """
-        return self._fetchall(select, {"now": datetime.utcnow()}, limit='ALL')
+        return self._fetchall(select, {'now': datetime.utcnow()}, limit='ALL')
 
     def update_escalation_rule(self, id, **kwargs):
         update = """

alerta/database/base.py

@@ -363,7 +363,6 @@ def update_escalation_rule(self, id, **kwargs):
     def delete_escalation_rule(self, id):
         raise NotImplementedError
 
-
     # ON CALLS
 
     def create_on_call(self, on_call):

alerta/models/alarms/alerta_isa_18_2.py

@@ -11,7 +11,7 @@
 
 from alerta.exceptions import InvalidAction
 from alerta.models.alarms import AlarmModel
-from alerta.models.enums import Action, Severity, Status
+from alerta.models.enums import Severity, Status
 
 SEVERITY_MAP = {
     Severity.Security: 10,

alerta/models/alarms/isa_18_2.py

@@ -73,7 +73,7 @@
 class StateMachine(AlarmModel):
 
     def register(self, app):
-        import alerta.management.views
+        # import alerta.management.views
         self.name = 'ANSI/ISA 18.2'
 
         StateMachine.Severity = app.config['SEVERITY_MAP'] or SEVERITY_MAP

alerta/models/escalation_rule.py

@@ -4,24 +4,24 @@
 
 from alerta.app import db
 from alerta.database.base import Query
-from alerta.utils.response import absolute_url
-from alerta.models.notification_rule import AdvancedSeverity
 from alerta.models.alert import Alert
+from alerta.models.notification_rule import AdvancedSeverity
+from alerta.utils.response import absolute_url
 
 JSON = Dict[str, Any]
 
 
 def alert_from_record(rec) -> 'dict':
     return {
-        "id": rec.id,
-        "resource": rec.resource,
-        "event": rec.event,
-        "severity": rec.severity,
-        "environment": rec.environment,
-        "service": rec.service,
-        "timeout": rec.timeout,
-        "value": rec.value,
-        "text": rec.text,
+        'id': rec.id,
+        'resource': rec.resource,
+        'event': rec.event,
+        'severity': rec.severity,
+        'environment': rec.environment,
+        'service': rec.service,
+        'timeout': rec.timeout,
+        'value': rec.value,
+        'text': rec.text,
     }
 
 
@@ -229,8 +229,6 @@ def from_record(cls, rec) -> 'EscalationRule':
             end_time=rec.end_time,
             days=rec.days,
         )
-
-
 
     @classmethod
     def from_db(cls, r: Union[Dict, Tuple]) -> 'EscalationRule':

alerta/models/notification_group.py

@@ -1,12 +1,9 @@
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union
+from typing import Any, Dict, List, Optional, Tuple, Union
 from uuid import uuid4
 
 from alerta.app import db
 from alerta.database.base import Query
 
-if TYPE_CHECKING:
-    from alerta.models.alert import Alert
-
 JSON = Dict[str, Any]
 
 
@@ -21,7 +18,7 @@ def __init__(self, **kwargs) -> None:
     def parse(cls, json: JSON) -> 'NotificationGroup':
         if not isinstance(json.get('users', []), list):
             raise ValueError('users must be a list')
-        if "name" not in json:
+        if 'name' not in json:
             raise ValueError('Missing required key: "name"')
 
         notification_group = NotificationGroup(

alerta/models/notification_history.py

@@ -1,11 +1,10 @@
-from typing import Any, Dict, List, Optional, Tuple, Union
 from datetime import datetime
+from typing import Any, Dict, List, Optional, Tuple, Union
 from uuid import uuid4
 
 from alerta.app import db
 from alerta.database.base import Query
 
-
 JSON = Dict[str, Any]
 
 
@@ -21,9 +20,9 @@ def __init__(self, sent: bool, message: str, channel: str, rule: str, alert: str
         self.alert = alert
         self.receiver = receiver
         self.sender = sender
-        self.error = kwargs.get("error", None)
-        self.confirmed = kwargs.get("confirmed", None)
-        self.confirmed_time = kwargs.get("confirmed_time", None)
+        self.error = kwargs.get('error', None)
+        self.confirmed = kwargs.get('confirmed', None)
+        self.confirmed_time = kwargs.get('confirmed_time', None)
         # confirmed boolean,
         # confirmed_time timestamp without time zone
 
@@ -45,17 +44,17 @@ def parse(cls, json: JSON) -> 'NotificationHistory':
     def serialize(self) -> Dict[str, Any]:
         return {
             'id': self.id,
-            "sent": self.sent,
-            "message": self.message,
-            "channel": self.channel,
-            "rule": self.rule,
-            "alert": self.alert,
-            "receiver": self.receiver,
-            "sender": self.sender,
-            "sent_time": self.sent_time,
-            "confirmed": self.confirmed,
-            "confirmed_time": self.confirmed_time,
-            "error": self.error
+            'sent': self.sent,
+            'message': self.message,
+            'channel': self.channel,
+            'rule': self.rule,
+            'alert': self.alert,
+            'receiver': self.receiver,
+            'sender': self.sender,
+            'sent_time': self.sent_time,
+            'confirmed': self.confirmed,
+            'confirmed_time': self.confirmed_time,
+            'error': self.error
         }
 
     def __repr__(self) -> str:
@@ -65,17 +64,17 @@ def __repr__(self) -> str:
     def from_document(cls, doc: Dict[str, Any]) -> 'NotificationHistory':
         return NotificationHistory(
             id=doc.get('id', None) or doc.get('_id'),
-            sent=doc["sent"],
-            message=doc["message"],
-            channel=doc["channel"],
-            rule=doc["rule"],
-            alert=doc["alert"],
-            receiver=doc["receiver"],
-            sender=doc["sender"],
-            sent_time=doc["sent_time"],
-            confirmed=doc["confirmed"],
-            confirmed_time=doc["confirmed_time"],
-            error=doc.get("error", None)
+            sent=doc['sent'],
+            message=doc['message'],
+            channel=doc['channel'],
+            rule=doc['rule'],
+            alert=doc['alert'],
+            receiver=doc['receiver'],
+            sender=doc['sender'],
+            sent_time=doc['sent_time'],
+            confirmed=doc['confirmed'],
+            confirmed_time=doc['confirmed_time'],
+            error=doc.get('error', None)
         )
 
     @ classmethod

alerta/plugins/notification_rule.py

@@ -1,18 +1,18 @@
 import json
 import logging
 import smtplib
+from datetime import datetime, timedelta
 from threading import Thread
 
 import requests
 from cryptography.fernet import Fernet, InvalidToken
+from flask import current_app
 from sendgrid import SendGridAPIClient
 from sendgrid.helpers.mail import Mail
 from twilio.base.exceptions import TwilioRestException
 from twilio.rest import Client
-from datetime import datetime, timedelta
 
 from alerta.models.alert import Alert
-from flask import current_app
 from alerta.models.notification_history import NotificationHistory
 from alerta.models.notification_rule import (NotificationChannel,
                                              NotificationRule)
@@ -101,10 +101,10 @@ def mylink_bearer_request(self, channel: NotificationChannel, fernet: Fernet):
             return
 
     def send_mylink_sms(self, message: str, channel: NotificationChannel, receivers: 'list[str]', fernet: Fernet, **kwargs):
-        bearer = channel.bearer + "d"
-        data = json.dumps([{"recipient": receiver, 'content': {'text': message, 'options': {'sms.sender': channel.sender}}} for receiver in receivers])
+        bearer = channel.bearer + 'd'
+        data = json.dumps([{'recipient': receiver, 'content': {'text': message, 'options': {'sms.sender': channel.sender}}} for receiver in receivers])
         headers = {'Content-Type': 'application/json', 'Authorization': f'Bearer {bearer}'}
-        return requests.post("https://api.linkmobility.com/sms/v1", data=data, headers=headers)
+        return requests.post('https://api.linkmobility.com/sms/v1', data=data, headers=headers)
 
     def send_link_mobility_sms(self, message: str, channel: NotificationChannel, receivers: 'list[str]', fernet: Fernet, **kwargs):
         numberOfReceivers = len(receivers)
@@ -247,11 +247,11 @@ def handle_channel(self, message: str, channel: NotificationChannel, notificatio
         elif notification_type == 'my_link':
             response = self.send_mylink_sms(message, channel, list({*notification_rule.receivers, *[f'{user.country_code}{user.phone_number}' for user in users]}), fernet)
             if response.status_code != 202:
-                LOG.error(f"Failed to send myLink message with response: {response.content}")
+                LOG.error(f'Failed to send myLink message with response: {response.content}')
                 for number in list({*notification_rule.receivers, *[f'{user.country_code}{user.phone_number}' for user in users]}):
                     self.log_notification(False, message, channel, notification_rule.id, alert, [number], error=response.content)
             else:
-                LOG.info(f"Successfully Sent message to myLink with response: {response.content}")
+                LOG.info(f'Successfully Sent message to myLink with response: {response.content}')
                 print(response.json())
                 for msg in response.json()['messages']:
                     self.log_notification(True, message, channel, notification_rule.id, alert, [msg['recipient']], id=msg['messageId'])
@@ -260,7 +260,7 @@ def handle_test(self, channel: NotificationChannel, info: NotificationRule, conf
         message = info.text if info.text != '' else 'this is a test message for testing a notification_channel in alerta'
         self.handle_channel(message, channel, info, [], Fernet(config['NOTIFICATION_KEY']), 'Test Notification Channel')
 
-    def handle_notifications(self, alert: 'Alert', notifications: 'list[tuple[NotificationRule,NotificationChannel, list[set[User or None]]]]', on_users: 'list[set[User or None]]', fernet: Fernet, app_context, status: str = ""):
+    def handle_notifications(self, alert: 'Alert', notifications: 'list[tuple[NotificationRule,NotificationChannel, list[set[User or None]]]]', on_users: 'list[set[User or None]]', fernet: Fernet, app_context, status: str = ''):
         app_context.push()
         standard_message = '%(environment)s: %(severity)s alert for %(service)s - %(resource)s is %(event)s'
         for notification_rule, channel, users in notifications:
@@ -269,7 +269,7 @@ def handle_notifications(self, alert: 'Alert', notifications: 'list[tuple[Notifi
 
             if notification_rule.use_oncall:
                 users.update(on_users)
-            msg_obj = {**alert.serialize, "status": status} if status != "" else alert.serialize
+            msg_obj = {**alert.serialize, 'status': status} if status != '' else alert.serialize
             message = (
                 notification_rule.text if notification_rule.text != '' and notification_rule.text is not None else standard_message
             ) % self.get_message_obj(msg_obj)
@@ -289,9 +289,9 @@ def update_bearer(self, channel: NotificationChannel, fernet):
                     bearer = data['access_token']
                     timeout = now + timedelta(0, data['expires_in'])
                     channel = channel.update_bearer(bearer, timeout)
-                    LOG.info(f"Updated access_token for myLink channel {channel.id}")
+                    LOG.info(f'Updated access_token for myLink channel {channel.id}')
                 else:
-                    LOG.error(f"Failed to update access token for myLink channel {channel.id} with response: {response.status_code} {response.content}")
+                    LOG.error(f'Failed to update access token for myLink channel {channel.id} with response: {response.status_code} {response.content}')
         return channel
 
     def post_receive(self, alert: 'Alert', **kwargs):

alerta/utils/config.py

@@ -110,7 +110,7 @@ def get_user_config():
         config['NOTIFICATION_KEY'] = get_config('NOTIFICATION_KEY', default=random_generated_key, type=str, config=config)
 
         config['ESCALATE_TIME'] = get_config('ESCALATE_TIME', default=60, type=int, config=config)
-        config['ESCALATE_SEVERITIES'] = get_config('ESCALATE_SEVERITIES', default=["critical", "major", "minor", "warning"], type=list, config=config)
+        config['ESCALATE_SEVERITIES'] = get_config('ESCALATE_SEVERITIES', default=['critical', 'major', 'minor', 'warning'], type=list, config=config)
 
         # Runtime config check
         if config['CUSTOMER_VIEWS'] and not config['AUTH_REQUIRED']:

alerta/views/escalation_rules.py

@@ -1,18 +1,16 @@
 from flask import current_app, g, jsonify, request
 from flask_cors import cross_origin
 
-from alerta.app import qb, alarm_model
+from alerta.app import alarm_model, qb
 from alerta.auth.decorators import permission
 from alerta.exceptions import ApiError
 from alerta.models.alert import Alert
-from alerta.models.enums import Scope
+from alerta.models.enums import Scope, TrendIndication
 from alerta.models.escalation_rule import EscalationRule
-from alerta.utils.api import assign_customer
+from alerta.utils.api import assign_customer, process_alert
 from alerta.utils.audit import write_audit_trail
 from alerta.utils.paging import Page
 from alerta.utils.response import absolute_url, jsonp
-from alerta.models.enums import TrendIndication
-from alerta.utils.api import process_alert
 
 from . import api
 

tests/test_notification_rule.py

@@ -877,11 +877,11 @@ def test_advanced_severity(self):
         }
         to_critical_major_from_all = {
             **all,
-            'advancedSeverity': [{'from': [], 'to':['major', 'critical']}],
+            'advancedSeverity': [{'from': [], 'to': ['major', 'critical']}],
         }
         to_normal_from_critical_major = {
             **all,
-            'advancedSeverity': [{'from': ['major', 'critical'], 'to':['normal']}],
+            'advancedSeverity': [{'from': ['major', 'critical'], 'to': ['normal']}],
         }
         alert_base = {
             'environment': 'Production',