Fix INF-364 alarmEventRecord does not comply with spec
[pti/o2.git] / 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