o2ims/adapter/alarm_repository.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 from typing import List
  16
  17 from o2ims.domain import alarm_obj
  18 from o2ims.domain.alarm_repo import AlarmDefinitionRepository, \
  19     AlarmEventRecordRepository, AlarmSubscriptionRepository, \
  20     AlarmProbableCauseRepository
  21 from o2common.helper import o2logging
  22 logger = o2logging.get_logger(__name__)
  23
  24
  25 class AlarmEventRecordSqlAlchemyRepository(AlarmEventRecordRepository):
  26     def __init__(self, session):
  27         super().__init__()
  28         self.session = session
  29
  30     def _add(self, alarm_event_record: alarm_obj.AlarmEventRecord):
  31         self.session.add(alarm_event_record)
  32
  33     def _get(self, alarm_event_record_id) -> alarm_obj.AlarmEventRecord:
  34         return self.session.query(alarm_obj.AlarmEventRecord).filter_by(
  35             alarmEventRecordId=alarm_event_record_id).first()
  36
  37     def _list(self) -> List[alarm_obj.AlarmEventRecord]:
  38         return self.session.query(alarm_obj.AlarmEventRecord)
  39
  40     def _update(self, alarm_event_record: alarm_obj.AlarmEventRecord):
  41         self.session.add(alarm_event_record)
  42
  43     def _delete(self, alarm_event_record_id):
  44         self.session.query(alarm_obj.AlarmEventRecord).filter_by(
  45             alarmEventRecordId=alarm_event_record_id).delete()
  46
  47
  48 class AlarmDefinitionSqlAlchemyRepository(AlarmDefinitionRepository):
  49     def __init__(self, session):
  50         super().__init__()
  51         self.session = session
  52
  53     def _add(self, definition: alarm_obj.AlarmDefinition):
  54         self.session.add(definition)
  55
  56     def _get(self, definition_id) -> alarm_obj.AlarmDefinition:
  57         return self.session.query(alarm_obj.AlarmDefinition).filter_by(
  58             alarmDefinitionId=definition_id).first()
  59
  60     def _list(self) -> List[alarm_obj.AlarmDefinition]:
  61         return self.session.query(alarm_obj.AlarmDefinition)
  62
  63     def _update(self, definition: alarm_obj.AlarmDefinition):
  64         self.session.add(definition)
  65
  66     def _delete(self, alarm_definition_id):
  67         self.session.query(alarm_obj.AlarmDefinition).filter_by(
  68             alarmDefinitionId=alarm_definition_id).delete()
  69
  70
  71 class AlarmSubscriptionSqlAlchemyRepository(AlarmSubscriptionRepository):
  72     def __init__(self, session):
  73         super().__init__()
  74         self.session = session
  75
  76     def _add(self, subscription: alarm_obj.AlarmSubscription):
  77         self.session.add(subscription)
  78
  79     def _get(self, subscription_id) -> alarm_obj.AlarmSubscription:
  80         return self.session.query(alarm_obj.AlarmSubscription).filter_by(
  81             alarmSubscriptionId=subscription_id).first()
  82
  83     def _list(self) -> List[alarm_obj.AlarmSubscription]:
  84         return self.session.query(alarm_obj.AlarmSubscription)
  85
  86     def _update(self, subscription: alarm_obj.AlarmSubscription):
  87         self.session.add(subscription)
  88
  89     def _delete(self, alarm_subscription_id):
  90         self.session.query(alarm_obj.AlarmSubscription).filter_by(
  91             alarmSubscriptionId=alarm_subscription_id).delete()
  92
  93
  94 class AlarmProbableCauseSqlAlchemyRepository(AlarmProbableCauseRepository):
  95     def __init__(self, session):
  96         super().__init__()
  97         self.session = session
  98
  99     def _add(self, probable_cause: alarm_obj.ProbableCause):
 100         self.session.add(probable_cause)
 101
 102     def _get(self, probable_cause_id) -> alarm_obj.ProbableCause:
 103         return self.session.query(alarm_obj.ProbableCause).filter_by(
 104             probableCauseId=probable_cause_id).first()
 105
 106     def _list(self) -> List[alarm_obj.ProbableCause]:
 107         return self.session.query(alarm_obj.ProbableCause)
 108
 109     def _update(self, probable_cause: alarm_obj.ProbableCause):
 110         self.session.add(probable_cause)
 111
 112     def _delete(self, probable_cause_id):
 113         self.session.query(alarm_obj.ProbableCause).filter_by(
 114             probableCauseId=probable_cause_id).delete()