1 # Copyright (C) 2021 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.
17 from o2common.service import unit_of_work
18 from o2common.views.view import gen_filter, check_filter
19 from o2common.views.pagination_view import Pagination
20 from o2common.views.route_exception import BadRequestException, \
23 from o2ims.views.alarm_dto import SubscriptionDTO
24 from o2ims.domain.alarm_obj import AlarmSubscription, AlarmEventRecord
26 from o2common.helper import o2logging
27 # from o2common.config import config
28 logger = o2logging.get_logger(__name__)
31 def alarm_event_records(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
32 pagination = Pagination(**kwargs)
33 query_kwargs = pagination.get_pagination()
34 args = gen_filter(AlarmEventRecord,
35 kwargs['filter']) if 'filter' in kwargs else []
37 li = uow.alarm_event_records.list_with_count(*args, **query_kwargs)
38 return pagination.get_result(li)
41 def alarm_event_record_one(alarmEventRecordId: str,
42 uow: unit_of_work.AbstractUnitOfWork):
44 first = uow.alarm_event_records.get(alarmEventRecordId)
45 return first.serialize() if first is not None else None
48 def subscriptions(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
49 pagination = Pagination(**kwargs)
50 query_kwargs = pagination.get_pagination()
51 args = gen_filter(AlarmSubscription,
52 kwargs['filter']) if 'filter' in kwargs else []
54 li = uow.alarm_subscriptions.list_with_count(*args, **query_kwargs)
55 return pagination.get_result(li)
58 def subscription_one(subscriptionId: str,
59 uow: unit_of_work.AbstractUnitOfWork):
61 first = uow.alarm_subscriptions.get(subscriptionId)
62 return first.serialize() if first is not None else None
65 def subscription_create(subscriptionDto: SubscriptionDTO.subscription_create,
66 uow: unit_of_work.AbstractUnitOfWork):
67 filter = subscriptionDto.get('filter', '')
68 consumer_subs_id = subscriptionDto.get('consumerSubscriptionId', '')
70 check_filter(AlarmEventRecord, filter)
72 sub_uuid = str(uuid.uuid4())
73 subscription = AlarmSubscription(
74 sub_uuid, subscriptionDto['callback'],
75 consumer_subs_id, filter)
78 args.append(getattr(AlarmSubscription, 'callback')
79 == subscriptionDto['callback'])
80 args.append(getattr(AlarmSubscription, 'filter') == filter)
81 args.append(getattr(AlarmSubscription,
82 'consumerSubscriptionId') == consumer_subs_id)
83 count, _ = uow.alarm_subscriptions.list_with_count(*args)
85 raise BadRequestException("The value of parameters is duplicated")
86 uow.alarm_subscriptions.add(subscription)
88 first = uow.alarm_subscriptions.get(sub_uuid)
89 return first.serialize()
92 def subscription_delete(subscriptionId: str,
93 uow: unit_of_work.AbstractUnitOfWork):
95 first = uow.alarm_subscriptions.get(subscriptionId)
97 raise NotFoundException(
98 "Alarm Subscription {} not found.".format(subscriptionId))
99 uow.alarm_subscriptions.delete(subscriptionId)