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
  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) -> List[obj.AlarmEventRecord]:
  35         return self._list()
  36
  37     def update(self, alarm_event_record: obj.AlarmEventRecord):
  38         self._update(alarm_event_record)
  39
  40     def delete(self, alarm_event_record_id):
  41         self._delete(alarm_event_record_id)
  42
  43     @abc.abstractmethod
  44     def _add(self, alarm_event_record: obj.AlarmEventRecord):
  45         raise NotImplementedError
  46
  47     @abc.abstractmethod
  48     def _get(self, alarm_event_record_id) -> obj.AlarmEventRecord:
  49         raise NotImplementedError
  50
  51     @abc.abstractmethod
  52     def _list(self) -> List[obj.AlarmEventRecord]:
  53         raise NotImplementedError
  54
  55     @abc.abstractmethod
  56     def _update(self, alarm_event_record: obj.AlarmEventRecord):
  57         raise NotImplementedError
  58
  59     @abc.abstractmethod
  60     def _delete(self, alarm_event_record_id):
  61         raise NotImplementedError
  62
  63
  64 class AlarmDefinitionRepository(abc.ABC):
  65     def __init__(self):
  66         self.seen = set()  # type: Set[obj.AlarmDefinition]
  67
  68     def add(self, definition: obj.AlarmDefinition):
  69         self._add(definition)
  70         self.seen.add(definition)
  71
  72     def get(self, definition_id) -> obj.AlarmDefinition:
  73         definition = self._get(definition_id)
  74         if definition:
  75             self.seen.add(definition)
  76         return definition
  77
  78     def list(self) -> List[obj.AlarmDefinition]:
  79         return self._list()
  80
  81     def update(self, definition: obj.AlarmDefinition):
  82         self._update(definition)
  83
  84     def delete(self, definition_id):
  85         self._delete(definition_id)
  86
  87     @abc.abstractmethod
  88     def _add(self, definition: obj.AlarmDefinition):
  89         raise NotImplementedError
  90
  91     @abc.abstractmethod
  92     def _get(self, definition_id) -> obj.AlarmDefinition:
  93         raise NotImplementedError
  94
  95     @abc.abstractmethod
  96     def _update(self, definition: obj.AlarmDefinition):
  97         raise NotImplementedError
  98
  99     @abc.abstractmethod
 100     def _delete(self, definition_id):
 101         raise NotImplementedError
 102
 103
 104 class AlarmDictionaryRepository(abc.ABC):
 105     def __init__(self):
 106         self.seen = set()  # type: Set[obj.AlarmDictionary]
 107
 108     def add(self, dictionary: obj.AlarmDictionary):
 109         self._add(dictionary)
 110         self.seen.add(dictionary)
 111
 112     def get(self, dictionary_id) -> obj.AlarmDictionary:
 113         dictionary = self._get(dictionary_id)
 114         if dictionary:
 115             self.seen.add(dictionary)
 116         return dictionary
 117
 118     def list(self) -> List[obj.AlarmDictionary]:
 119         return self._list()
 120
 121     def update(self, dictionary: obj.AlarmDictionary):
 122         self._update(dictionary)
 123
 124     def delete(self, dictionary_id):
 125         self._delete(dictionary_id)
 126
 127     @abc.abstractmethod
 128     def _add(self, dictionary: obj.AlarmDictionary):
 129         raise NotImplementedError
 130
 131     @abc.abstractmethod
 132     def _get(self, dictionary_id) -> obj.AlarmDictionary:
 133         raise NotImplementedError
 134
 135     @abc.abstractmethod
 136     def _update(self, dictionary: obj.AlarmDictionary):
 137         raise NotImplementedError
 138
 139     @abc.abstractmethod
 140     def _delete(self, dictionary_id):
 141         raise NotImplementedError
 142
 143
 144 class AlarmSubscriptionRepository(abc.ABC):
 145     def __init__(self):
 146         self.seen = set()  # type: Set[obj.AlarmSubscription]
 147
 148     def add(self, subscription: obj.AlarmSubscription):
 149         self._add(subscription)
 150         self.seen.add(subscription)
 151
 152     def get(self, subscription_id) -> obj.AlarmSubscription:
 153         subscription = self._get(subscription_id)
 154         if subscription:
 155             self.seen.add(subscription)
 156         return subscription
 157
 158     def list(self) -> List[obj.AlarmSubscription]:
 159         return self._list()
 160
 161     def update(self, subscription: obj.AlarmSubscription):
 162         self._update(subscription)
 163
 164     def delete(self, subscription_id):
 165         self._delete(subscription_id)
 166
 167     @abc.abstractmethod
 168     def _add(self, subscription: obj.AlarmSubscription):
 169         raise NotImplementedError
 170
 171     @abc.abstractmethod
 172     def _get(self, subscription_id) -> obj.AlarmSubscription:
 173         raise NotImplementedError
 174
 175     @abc.abstractmethod
 176     def _update(self, subscription: obj.AlarmSubscription):
 177         raise NotImplementedError
 178
 179     @abc.abstractmethod
 180     def _delete(self, subscription_id):
 181         raise NotImplementedError
 182
 183
 184 class AlarmProbableCauseRepository(abc.ABC):
 185     def __init__(self):
 186         self.seen = set()  # type: Set[obj.ProbableCause]
 187
 188     def add(self, probable_cause: obj.ProbableCause):
 189         self._add(probable_cause)
 190         self.seen.add(probable_cause)
 191
 192     def get(self, probable_cause_id) -> obj.ProbableCause:
 193         probable_cause = self._get(probable_cause_id)
 194         if probable_cause:
 195             self.seen.add(probable_cause)
 196         return probable_cause
 197
 198     def list(self) -> List[obj.ProbableCause]:
 199         return self._list()
 200
 201     def update(self, probable_cause: obj.ProbableCause):
 202         self._update(probable_cause)
 203
 204     def delete(self, probable_cause_id):
 205         self._delete(probable_cause_id)
 206
 207     @abc.abstractmethod
 208     def _add(self, probable_cause: obj.ProbableCause):
 209         raise NotImplementedError
 210
 211     @abc.abstractmethod
 212     def _get(self, probable_cause_id) -> obj.ProbableCause:
 213         raise NotImplementedError
 214
 215     @abc.abstractmethod
 216     def _update(self, probable_cause: obj.ProbableCause):
 217         raise NotImplementedError
 218
 219     @abc.abstractmethod
 220     def _delete(self, probable_cause_id):
 221         raise NotImplementedError