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, **kwargs) -> \
  38             Tuple[int, List[obj.AlarmEventRecord]]:
  39         return self._list(**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 _update(self, definition: obj.AlarmDefinition):
 101         raise NotImplementedError
 102
 103     @abc.abstractmethod
 104     def _delete(self, definition_id):
 105         raise NotImplementedError
 106
 107
 108 class AlarmDictionaryRepository(abc.ABC):
 109     def __init__(self):
 110         self.seen = set()  # type: Set[obj.AlarmDictionary]
 111
 112     def add(self, dictionary: obj.AlarmDictionary):
 113         self._add(dictionary)
 114         self.seen.add(dictionary)
 115
 116     def get(self, dictionary_id) -> obj.AlarmDictionary:
 117         dictionary = self._get(dictionary_id)
 118         if dictionary:
 119             self.seen.add(dictionary)
 120         return dictionary
 121
 122     def list(self) -> List[obj.AlarmDictionary]:
 123         return self._list()
 124
 125     def update(self, dictionary: obj.AlarmDictionary):
 126         self._update(dictionary)
 127
 128     def delete(self, dictionary_id):
 129         self._delete(dictionary_id)
 130
 131     @abc.abstractmethod
 132     def _add(self, dictionary: obj.AlarmDictionary):
 133         raise NotImplementedError
 134
 135     @abc.abstractmethod
 136     def _get(self, dictionary_id) -> obj.AlarmDictionary:
 137         raise NotImplementedError
 138
 139     @abc.abstractmethod
 140     def _update(self, dictionary: obj.AlarmDictionary):
 141         raise NotImplementedError
 142
 143     @abc.abstractmethod
 144     def _delete(self, dictionary_id):
 145         raise NotImplementedError
 146
 147
 148 class AlarmSubscriptionRepository(abc.ABC):
 149     def __init__(self):
 150         self.seen = set()  # type: Set[obj.AlarmSubscription]
 151
 152     def add(self, subscription: obj.AlarmSubscription):
 153         self._add(subscription)
 154         self.seen.add(subscription)
 155
 156     def get(self, subscription_id) -> obj.AlarmSubscription:
 157         subscription = self._get(subscription_id)
 158         if subscription:
 159             self.seen.add(subscription)
 160         return subscription
 161
 162     def list(self, **kwargs) -> List[obj.AlarmSubscription]:
 163         return self._list(**kwargs)[1]
 164
 165     def list_with_count(self, **kwargs) -> \
 166             Tuple[int, List[obj.AlarmSubscription]]:
 167         return self._list(**kwargs)
 168
 169     def update(self, subscription: obj.AlarmSubscription):
 170         self._update(subscription)
 171
 172     def delete(self, subscription_id):
 173         self._delete(subscription_id)
 174
 175     @abc.abstractmethod
 176     def _add(self, subscription: obj.AlarmSubscription):
 177         raise NotImplementedError
 178
 179     @abc.abstractmethod
 180     def _get(self, subscription_id) -> obj.AlarmSubscription:
 181         raise NotImplementedError
 182
 183     @abc.abstractmethod
 184     def _list(self, **kwargs) -> Tuple[int, List[obj.AlarmSubscription]]:
 185         raise NotImplementedError
 186
 187     @abc.abstractmethod
 188     def _update(self, subscription: obj.AlarmSubscription):
 189         raise NotImplementedError
 190
 191     @abc.abstractmethod
 192     def _delete(self, subscription_id):
 193         raise NotImplementedError
 194
 195
 196 class AlarmProbableCauseRepository(abc.ABC):
 197     def __init__(self):
 198         self.seen = set()  # type: Set[obj.ProbableCause]
 199
 200     def add(self, probable_cause: obj.ProbableCause):
 201         self._add(probable_cause)
 202         self.seen.add(probable_cause)
 203
 204     def get(self, probable_cause_id) -> obj.ProbableCause:
 205         probable_cause = self._get(probable_cause_id)
 206         if probable_cause:
 207             self.seen.add(probable_cause)
 208         return probable_cause
 209
 210     def list(self) -> List[obj.ProbableCause]:
 211         return self._list()
 212
 213     def update(self, probable_cause: obj.ProbableCause):
 214         self._update(probable_cause)
 215
 216     def delete(self, probable_cause_id):
 217         self._delete(probable_cause_id)
 218
 219     @abc.abstractmethod
 220     def _add(self, probable_cause: obj.ProbableCause):
 221         raise NotImplementedError
 222
 223     @abc.abstractmethod
 224     def _get(self, probable_cause_id) -> obj.ProbableCause:
 225         raise NotImplementedError
 226
 227     @abc.abstractmethod
 228     def _update(self, probable_cause: obj.ProbableCause):
 229         raise NotImplementedError
 230
 231     @abc.abstractmethod
 232     def _delete(self, probable_cause_id):
 233         raise NotImplementedError