o2ims/domain/alarm_repo.py

   1 # Copyright (C) 2022 Wind River Systems, Inc.
   2 #
   3 #  Licensed under the Apache License, Version 2.0 (the "License");
   4 #  you may not use this file except in compliance with the License.
   5 #  You may obtain a copy of the License at
   6 #
   7 #      http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #  Unless required by applicable law or agreed to in writing, software
  10 #  distributed under the License is distributed on an "AS IS" BASIS,
  11 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #  See the License for the specific language governing permissions and
  13 #  limitations under the License.
  14
  15 import abc
  16 from typing import List, Set, Tuple
  17 from o2ims.domain import alarm_obj as obj
  18
  19
  20 class AlarmEventRecordRepository(abc.ABC):
  21     def __init__(self):
  22         self.seen = set()  # type: Set[obj.AlarmEventRecord]
  23
  24     def add(self, alarm_event_record: obj.AlarmEventRecord):
  25         self._add(alarm_event_record)
  26         self.seen.add(alarm_event_record)
  27
  28     def get(self, alarm_event_record_id) -> obj.AlarmEventRecord:
  29         alarm_event_record = self._get(alarm_event_record_id)
  30         if alarm_event_record:
  31             self.seen.add(alarm_event_record)
  32         return alarm_event_record
  33
  34     def list(self, **kwargs) -> List[obj.AlarmEventRecord]:
  35         return self._list(*[], **kwargs)[1]
  36
  37     def list_with_count(self, *args, **kwargs) -> \
  38             Tuple[int, List[obj.AlarmEventRecord]]:
  39         return self._list(*args, **kwargs)
  40
  41     def update(self, alarm_event_record: obj.AlarmEventRecord):
  42         self._update(alarm_event_record)
  43
  44     def delete(self, alarm_event_record_id):
  45         self._delete(alarm_event_record_id)
  46
  47     @abc.abstractmethod
  48     def _add(self, alarm_event_record: obj.AlarmEventRecord):
  49         raise NotImplementedError
  50
  51     @abc.abstractmethod
  52     def _get(self, alarm_event_record_id) -> obj.AlarmEventRecord:
  53         raise NotImplementedError
  54
  55     @abc.abstractmethod
  56     def _list(self, **kwargs) -> Tuple[int, List[obj.AlarmEventRecord]]:
  57         raise NotImplementedError
  58
  59     @abc.abstractmethod
  60     def _update(self, alarm_event_record: obj.AlarmEventRecord):
  61         raise NotImplementedError
  62
  63     @abc.abstractmethod
  64     def _delete(self, alarm_event_record_id):
  65         raise NotImplementedError
  66
  67
  68 class AlarmDefinitionRepository(abc.ABC):
  69     def __init__(self):
  70         self.seen = set()  # type: Set[obj.AlarmDefinition]
  71
  72     def add(self, definition: obj.AlarmDefinition):
  73         self._add(definition)
  74         self.seen.add(definition)
  75
  76     def get(self, definition_id) -> obj.AlarmDefinition:
  77         definition = self._get(definition_id)
  78         if definition:
  79             self.seen.add(definition)
  80         return definition
  81
  82     def list(self) -> List[obj.AlarmDefinition]:
  83         return self._list()
  84
  85     def update(self, definition: obj.AlarmDefinition):
  86         self._update(definition)
  87
  88     def delete(self, definition_id):
  89         self._delete(definition_id)
  90
  91     @abc.abstractmethod
  92     def _add(self, definition: obj.AlarmDefinition):
  93         raise NotImplementedError
  94
  95     @abc.abstractmethod
  96     def _get(self, definition_id) -> obj.AlarmDefinition:
  97         raise NotImplementedError
  98
  99     @abc.abstractmethod
 100     def _list(self, **kwargs) -> List[obj.AlarmDefinition]:
 101         raise NotImplementedError
 102
 103     @abc.abstractmethod
 104     def _update(self, definition: obj.AlarmDefinition):
 105         raise NotImplementedError
 106
 107     @abc.abstractmethod
 108     def _delete(self, definition_id):
 109         raise NotImplementedError
 110
 111
 112 class AlarmDictionaryRepository(abc.ABC):
 113     def __init__(self):
 114         self.seen = set()  # type: Set[obj.AlarmDictionary]
 115
 116     def add(self, dictionary: obj.AlarmDictionary):
 117         self._add(dictionary)
 118         self.seen.add(dictionary)
 119
 120     def get(self, dictionary_id) -> obj.AlarmDictionary:
 121         dictionary = self._get(dictionary_id)
 122         if dictionary:
 123             self.seen.add(dictionary)
 124         return dictionary
 125
 126     def list(self) -> List[obj.AlarmDictionary]:
 127         return self._list()
 128
 129     def update(self, dictionary: obj.AlarmDictionary):
 130         self._update(dictionary)
 131
 132     def delete(self, dictionary_id):
 133         self._delete(dictionary_id)
 134
 135     @abc.abstractmethod
 136     def _add(self, dictionary: obj.AlarmDictionary):
 137         raise NotImplementedError
 138
 139     @abc.abstractmethod
 140     def _get(self, dictionary_id) -> obj.AlarmDictionary:
 141         raise NotImplementedError
 142
 143     @abc.abstractmethod
 144     def _list(self, **kwargs) -> List[obj.AlarmDictionary]:
 145         raise NotImplementedError
 146
 147     @abc.abstractmethod
 148     def _delete(self, dictionary_id):
 149         raise NotImplementedError
 150
 151
 152 class AlarmSubscriptionRepository(abc.ABC):
 153     def __init__(self):
 154         self.seen = set()  # type: Set[obj.AlarmSubscription]
 155
 156     def add(self, subscription: obj.AlarmSubscription):
 157         self._add(subscription)
 158         self.seen.add(subscription)
 159
 160     def get(self, subscription_id) -> obj.AlarmSubscription:
 161         subscription = self._get(subscription_id)
 162         if subscription:
 163             self.seen.add(subscription)
 164         return subscription
 165
 166     def list(self, **kwargs) -> List[obj.AlarmSubscription]:
 167         return self._list(*[], **kwargs)[1]
 168
 169     def list_with_count(self, *args, **kwargs) -> \
 170             Tuple[int, List[obj.AlarmSubscription]]:
 171         return self._list(*args, **kwargs)
 172
 173     def update(self, subscription: obj.AlarmSubscription):
 174         self._update(subscription)
 175
 176     def delete(self, subscription_id):
 177         self._delete(subscription_id)
 178
 179     @abc.abstractmethod
 180     def _add(self, subscription: obj.AlarmSubscription):
 181         raise NotImplementedError
 182
 183     @abc.abstractmethod
 184     def _get(self, subscription_id) -> obj.AlarmSubscription:
 185         raise NotImplementedError
 186
 187     @abc.abstractmethod
 188     def _list(self, **kwargs) -> Tuple[int, List[obj.AlarmSubscription]]:
 189         raise NotImplementedError
 190
 191     @abc.abstractmethod
 192     def _update(self, subscription: obj.AlarmSubscription):
 193         raise NotImplementedError
 194
 195     @abc.abstractmethod
 196     def _delete(self, subscription_id):
 197         raise NotImplementedError
 198
 199
 200 class AlarmProbableCauseRepository(abc.ABC):
 201     def __init__(self):
 202         self.seen = set()  # type: Set[obj.ProbableCause]
 203
 204     def add(self, probable_cause: obj.ProbableCause):
 205         self._add(probable_cause)
 206         self.seen.add(probable_cause)
 207
 208     def get(self, probable_cause_id) -> obj.ProbableCause:
 209         probable_cause = self._get(probable_cause_id)
 210         if probable_cause:
 211             self.seen.add(probable_cause)
 212         return probable_cause
 213
 214     def list(self) -> List[obj.ProbableCause]:
 215         return self._list()
 216
 217     def update(self, probable_cause: obj.ProbableCause):
 218         self._update(probable_cause)
 219
 220     def delete(self, probable_cause_id):
 221         self._delete(probable_cause_id)
 222
 223     @abc.abstractmethod
 224     def _add(self, probable_cause: obj.ProbableCause):
 225         raise NotImplementedError
 226
 227     @abc.abstractmethod
 228     def _get(self, probable_cause_id) -> obj.ProbableCause:
 229         raise NotImplementedError
 230
 231     @abc.abstractmethod
 232     def _list(self, **kwargs) -> List[obj.ProbableCause]:
 233         raise NotImplementedError
 234
 235     @abc.abstractmethod
 236     def _update(self, probable_cause: obj.ProbableCause):
 237         raise NotImplementedError
 238
 239     @abc.abstractmethod
 240     def _delete(self, probable_cause_id):
 241         raise NotImplementedError