o2ims/views/alarm_view.py

   1 # Copyright (C) 2021 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 uuid as uuid
  16
  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, \
  21     NotFoundException
  22
  23 from o2ims.views.alarm_dto import SubscriptionDTO
  24 from o2ims.domain.alarm_obj import AlarmSubscription, AlarmEventRecord
  25
  26 from o2common.helper import o2logging
  27 # from o2common.config import config
  28 logger = o2logging.get_logger(__name__)
  29
  30
  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 []
  36     with uow:
  37         li = uow.alarm_event_records.list_with_count(*args, **query_kwargs)
  38     return pagination.get_result(li)
  39
  40
  41 def alarm_event_record_one(alarmEventRecordId: str,
  42                            uow: unit_of_work.AbstractUnitOfWork):
  43     with uow:
  44         first = uow.alarm_event_records.get(alarmEventRecordId)
  45         return first.serialize() if first is not None else None
  46
  47
  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 []
  53     with uow:
  54         li = uow.alarm_subscriptions.list_with_count(*args, **query_kwargs)
  55     return pagination.get_result(li)
  56
  57
  58 def subscription_one(subscriptionId: str,
  59                      uow: unit_of_work.AbstractUnitOfWork):
  60     with uow:
  61         first = uow.alarm_subscriptions.get(subscriptionId)
  62         return first.serialize() if first is not None else None
  63
  64
  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', '')
  69
  70     check_filter(AlarmEventRecord, filter)
  71
  72     sub_uuid = str(uuid.uuid4())
  73     subscription = AlarmSubscription(
  74         sub_uuid, subscriptionDto['callback'],
  75         consumer_subs_id, filter)
  76     with uow:
  77         args = list()
  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)
  84         if count > 0:
  85             raise BadRequestException("The value of parameters is duplicated")
  86         uow.alarm_subscriptions.add(subscription)
  87         uow.commit()
  88         first = uow.alarm_subscriptions.get(sub_uuid)
  89         return first.serialize()
  90
  91
  92 def subscription_delete(subscriptionId: str,
  93                         uow: unit_of_work.AbstractUnitOfWork):
  94     with uow:
  95         first = uow.alarm_subscriptions.get(subscriptionId)
  96         if not first:
  97             raise NotFoundException(
  98                 "Alarm Subscription {} not found.".format(subscriptionId))
  99         uow.alarm_subscriptions.delete(subscriptionId)
 100         uow.commit()
 101     return True