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
  22 from o2ims.views.alarm_dto import SubscriptionDTO
  23 from o2ims.domain.alarm_obj import AlarmSubscription, AlarmEventRecord
  24
  25 from o2common.helper import o2logging
  26 # from o2common.config import config
  27 logger = o2logging.get_logger(__name__)
  28
  29
  30 def alarm_event_records(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
  31     pagination = Pagination(**kwargs)
  32     query_kwargs = pagination.get_pagination()
  33     args = gen_filter(AlarmEventRecord,
  34                       kwargs['filter']) if 'filter' in kwargs else []
  35     with uow:
  36         li = uow.alarm_event_records.list_with_count(*args, **query_kwargs)
  37     return pagination.get_result(li)
  38
  39
  40 def alarm_event_record_one(alarmEventRecordId: str,
  41                            uow: unit_of_work.AbstractUnitOfWork):
  42     with uow:
  43         first = uow.alarm_event_records.get(alarmEventRecordId)
  44         return first.serialize() if first is not None else None
  45
  46
  47 def subscriptions(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
  48     pagination = Pagination(**kwargs)
  49     query_kwargs = pagination.get_pagination()
  50     args = gen_filter(AlarmSubscription,
  51                       kwargs['filter']) if 'filter' in kwargs else []
  52     with uow:
  53         li = uow.alarm_subscriptions.list_with_count(*args, **query_kwargs)
  54     return pagination.get_result(li)
  55
  56
  57 def subscription_one(subscriptionId: str,
  58                      uow: unit_of_work.AbstractUnitOfWork):
  59     with uow:
  60         first = uow.alarm_subscriptions.get(subscriptionId)
  61         return first.serialize() if first is not None else None
  62
  63
  64 def subscription_create(subscriptionDto: SubscriptionDTO.subscription_create,
  65                         uow: unit_of_work.AbstractUnitOfWork):
  66     filter = subscriptionDto.get('filter', '')
  67     consumer_subs_id = subscriptionDto.get('consumerSubscriptionId', '')
  68
  69     check_filter(AlarmEventRecord, filter)
  70
  71     sub_uuid = str(uuid.uuid4())
  72     subscription = AlarmSubscription(
  73         sub_uuid, subscriptionDto['callback'],
  74         consumer_subs_id, filter)
  75     with uow:
  76         args = list()
  77         args.append(getattr(AlarmSubscription, 'callback')
  78                     == subscriptionDto['callback'])
  79         args.append(getattr(AlarmSubscription, 'filter') == filter)
  80         args.append(getattr(AlarmSubscription,
  81                     'consumerSubscriptionId') == consumer_subs_id)
  82         count, _ = uow.alarm_subscriptions.list_with_count(*args)
  83         if count > 0:
  84             raise BadRequestException("The value of parameters is duplicated")
  85         uow.alarm_subscriptions.add(subscription)
  86         uow.commit()
  87         first = uow.alarm_subscriptions.get(sub_uuid)
  88         return first.serialize()
  89
  90
  91 def subscription_delete(subscriptionId: str,
  92                         uow: unit_of_work.AbstractUnitOfWork):
  93     with uow:
  94         uow.alarm_subscriptions.delete(subscriptionId)
  95         uow.commit()
  96     return True