1 # Copyright (C) 2022 Wind River Systems, Inc.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
15 from typing import List
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__)
25 class AlarmEventRecordSqlAlchemyRepository(AlarmEventRecordRepository):
26 def __init__(self, session):
28 self.session = session
30 def _add(self, alarm_event_record: alarm_obj.AlarmEventRecord):
31 self.session.add(alarm_event_record)
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()
37 def _list(self) -> List[alarm_obj.AlarmEventRecord]:
38 return self.session.query(alarm_obj.AlarmEventRecord)
40 def _update(self, alarm_event_record: alarm_obj.AlarmEventRecord):
41 self.session.add(alarm_event_record)
43 def _delete(self, alarm_event_record_id):
44 self.session.query(alarm_obj.AlarmEventRecord).filter_by(
45 alarmEventRecordId=alarm_event_record_id).delete()
48 class AlarmDefinitionSqlAlchemyRepository(AlarmDefinitionRepository):
49 def __init__(self, session):
51 self.session = session
53 def _add(self, definition: alarm_obj.AlarmDefinition):
54 self.session.add(definition)
56 def _get(self, definition_id) -> alarm_obj.AlarmDefinition:
57 return self.session.query(alarm_obj.AlarmDefinition).filter_by(
58 alarmDefinitionId=definition_id).first()
60 def _list(self) -> List[alarm_obj.AlarmDefinition]:
61 return self.session.query(alarm_obj.AlarmDefinition)
63 def _update(self, definition: alarm_obj.AlarmDefinition):
64 self.session.add(definition)
66 def _delete(self, alarm_definition_id):
67 self.session.query(alarm_obj.AlarmDefinition).filter_by(
68 alarmDefinitionId=alarm_definition_id).delete()
71 class AlarmSubscriptionSqlAlchemyRepository(AlarmSubscriptionRepository):
72 def __init__(self, session):
74 self.session = session
76 def _add(self, subscription: alarm_obj.AlarmSubscription):
77 self.session.add(subscription)
79 def _get(self, subscription_id) -> alarm_obj.AlarmSubscription:
80 return self.session.query(alarm_obj.AlarmSubscription).filter_by(
81 alarmSubscriptionId=subscription_id).first()
83 def _list(self) -> List[alarm_obj.AlarmSubscription]:
84 return self.session.query(alarm_obj.AlarmSubscription)
86 def _update(self, subscription: alarm_obj.AlarmSubscription):
87 self.session.add(subscription)
89 def _delete(self, alarm_subscription_id):
90 self.session.query(alarm_obj.AlarmSubscription).filter_by(
91 alarmSubscriptionId=alarm_subscription_id).delete()
94 class AlarmProbableCauseSqlAlchemyRepository(AlarmProbableCauseRepository):
95 def __init__(self, session):
97 self.session = session
99 def _add(self, probable_cause: alarm_obj.ProbableCause):
100 self.session.add(probable_cause)
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()
106 def _list(self) -> List[alarm_obj.ProbableCause]:
107 return self.session.query(alarm_obj.ProbableCause)
109 def _update(self, probable_cause: alarm_obj.ProbableCause):
110 self.session.add(probable_cause)
112 def _delete(self, probable_cause_id):
113 self.session.query(alarm_obj.ProbableCause).filter_by(
114 probableCauseId=probable_cause_id).delete()