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.pagination_view import Pagination
  19 from o2common.views.view import gen_filter
  20 from o2ims.views.alarm_dto import SubscriptionDTO
  21 from o2ims.domain.alarm_obj import AlarmSubscription, AlarmEventRecord
  22
  23 from o2common.helper import o2logging
  24 # from o2common.config import config
  25 logger = o2logging.get_logger(__name__)
  26
  27
  28 def alarm_event_records(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
  29     pagination = Pagination(**kwargs)
  30     query_kwargs = pagination.get_pagination()
  31     args = gen_filter(AlarmEventRecord,
  32                       kwargs['filter']) if 'filter' in kwargs else []
  33     with uow:
  34         li = uow.alarm_event_records.list_with_count(*args, **query_kwargs)
  35     return pagination.get_result(li)
  36
  37
  38 def alarm_event_record_one(alarmEventRecordId: str,
  39                            uow: unit_of_work.AbstractUnitOfWork):
  40     with uow:
  41         first = uow.alarm_event_records.get(alarmEventRecordId)
  42         return first.serialize() if first is not None else None
  43
  44
  45 def subscriptions(uow: unit_of_work.AbstractUnitOfWork, **kwargs):
  46     pagination = Pagination(**kwargs)
  47     query_kwargs = pagination.get_pagination()
  48     args = gen_filter(AlarmSubscription,
  49                       kwargs['filter']) if 'filter' in kwargs else []
  50     with uow:
  51         li = uow.alarm_subscriptions.list_with_count(*args, **query_kwargs)
  52     return pagination.get_result(li)
  53
  54
  55 def subscription_one(subscriptionId: str,
  56                      uow: unit_of_work.AbstractUnitOfWork):
  57     with uow:
  58         first = uow.alarm_subscriptions.get(subscriptionId)
  59         return first.serialize() if first is not None else None
  60
  61
  62 def subscription_create(subscriptionDto: SubscriptionDTO.subscription,
  63                         uow: unit_of_work.AbstractUnitOfWork):
  64
  65     sub_uuid = str(uuid.uuid4())
  66     subscription = AlarmSubscription(
  67         sub_uuid, subscriptionDto['callback'],
  68         subscriptionDto['consumerSubscriptionId'],
  69         subscriptionDto['filter'])
  70     with uow:
  71         uow.alarm_subscriptions.add(subscription)
  72         uow.commit()
  73     return {"alarmSubscriptionId": sub_uuid}
  74
  75
  76 def subscription_delete(subscriptionId: str,
  77                         uow: unit_of_work.AbstractUnitOfWork):
  78     with uow:
  79         uow.alarm_subscriptions.delete(subscriptionId)
  80         uow.commit()
  81     return True