# limitations under the License.
import abc
-from typing import List, Set
+from typing import List, Set, Tuple
from o2ims.domain import alarm_obj as obj
self.seen.add(alarm_event_record)
return alarm_event_record
- def list(self) -> List[obj.AlarmEventRecord]:
- return self._list()
+ def list(self, **kwargs) -> List[obj.AlarmEventRecord]:
+ return self._list(*[], **kwargs)[1]
+
+ def list_with_count(self, *args, **kwargs) -> \
+ Tuple[int, List[obj.AlarmEventRecord]]:
+ return self._list(*args, **kwargs)
def update(self, alarm_event_record: obj.AlarmEventRecord):
self._update(alarm_event_record)
raise NotImplementedError
@abc.abstractmethod
- def _list(self) -> List[obj.AlarmEventRecord]:
+ def _list(self, **kwargs) -> Tuple[int, List[obj.AlarmEventRecord]]:
raise NotImplementedError
@abc.abstractmethod
self.seen.add(subscription)
return subscription
- def list(self) -> List[obj.AlarmSubscription]:
- return self._list()
+ def list(self, **kwargs) -> List[obj.AlarmSubscription]:
+ return self._list(*[], **kwargs)[1]
+
+ def list_with_count(self, *args, **kwargs) -> \
+ Tuple[int, List[obj.AlarmSubscription]]:
+ return self._list(*args, **kwargs)
def update(self, subscription: obj.AlarmSubscription):
self._update(subscription)
def _get(self, subscription_id) -> obj.AlarmSubscription:
raise NotImplementedError
+ @abc.abstractmethod
+ def _list(self, **kwargs) -> Tuple[int, List[obj.AlarmSubscription]]:
+ raise NotImplementedError
+
@abc.abstractmethod
def _update(self, subscription: obj.AlarmSubscription):
raise NotImplementedError