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