- def get_alarm(self, alarm_id) -> Alarm:
- with db.atomic():
- alarm = (Alarm.select()
- .where(Alarm.alarm_id == alarm_id)
- .get())
- return alarm
-
- def save_alarm(self, name, threshold, operation, severity, statistic, metric_name, vdur_name,
- vnf_member_index, nsr_id) -> Alarm:
- """Saves alarm."""
- # TODO: Add uuid optional param and check if exists to handle updates (see self.save_credentials)
- with db.atomic():
- alarm = Alarm()
- alarm.uuid = str(uuid.uuid4())
- alarm.name = name
- alarm.threshold = threshold
- alarm.operation = operation
- alarm.severity = severity
- alarm.statistic = statistic
- alarm.monitoring_param = metric_name
- alarm.vdur_name = vdur_name
- alarm.vnf_member_index = vnf_member_index
- alarm.nsr_id = nsr_id
- alarm.save()
- return alarm
-
- def delete_alarm(self, alarm_uuid) -> None:
- with db.atomic():
- alarm = (Alarm.select()
- .where(Alarm.uuid == alarm_uuid)
- .get())
- alarm.delete_instance()
-
- def get_vim_type(self, vim_account_id) -> str:
- """Get the vim type that is required by the message."""
- vim_type = None
- credentials = self.get_credentials(vim_account_id)
- config = json.loads(credentials.config)
- if 'vim_type' in config:
- vim_type = config['vim_type']
- return str(vim_type.lower())
+
+class AlarmRepository:
+ @staticmethod
+ def create(**query) -> Alarm:
+ return Alarm.create(**query)
+
+ @staticmethod
+ def get(*expressions) -> Alarm:
+ return Alarm.select().where(*expressions).get()
+
+ @staticmethod
+ def list(*expressions) -> Iterable[Alarm]:
+ if expressions == ():
+ return Alarm.select()