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.
15 # pylint: disable=unused-argument
16 from __future__ import annotations
17 from typing import Callable
19 # from dataclasses import asdict
20 # from typing import List, Dict, Callable, Type
23 from o2common.config import config
24 # from o2common.service.messagebus import MessageBus
25 from o2common.service.unit_of_work import AbstractUnitOfWork
26 from o2ims.domain import events, commands
27 from o2ims.domain.ocloud import Ocloud
28 from o2ims.domain.stx_object import StxGenericModel
29 from o2ims.domain.resource_type import InvalidOcloudState, MismatchedModel
30 from o2ims.domain.subscription_obj import NotificationEventEnum
32 # from . import unit_of_work
34 from o2common.helper import o2logging
35 logger = o2logging.get_logger(__name__)
38 class InvalidResourceType(Exception):
43 cmd: commands.UpdateOCloud,
44 uow: AbstractUnitOfWork,
49 oclouds = uow.oclouds.list()
50 if oclouds and oclouds.count() > 1:
51 raise InvalidOcloudState("More than 1 ocloud is found")
52 elif not oclouds or oclouds.count() == 0:
53 logger.info("add ocloud:" + stxobj.name
54 + " update_at: " + str(stxobj.updatetime)
55 + " id: " + str(stxobj.id)
56 + " hash: " + str(stxobj.hash))
57 entry = create_by(stxobj)
58 uow.oclouds.add(entry)
60 logger.info("Add the ocloud: " + stxobj.id
61 + ", name: " + stxobj.name)
63 localmodel = oclouds.first()
64 if is_outdated(localmodel, stxobj):
65 logger.info("update ocloud:" + stxobj.name
66 + " update_at: " + str(stxobj.updatetime)
67 + " id: " + str(stxobj.id)
68 + " hash: " + str(stxobj.hash))
69 update_by(localmodel, stxobj)
70 uow.oclouds.update(localmodel)
72 logger.info("Update the ocloud: " + stxobj.id
73 + ", name: " + stxobj.name)
77 def is_outdated(ocloud: Ocloud, stxobj: StxGenericModel):
78 # if stxobj.updatetime:
79 # return True if Ocloud.updatetime < stxobj.updatetime else False
81 return True if ocloud.hash != stxobj.hash else False
84 def create_by(stxobj: StxGenericModel) -> Ocloud:
85 imsendpoint = config.get_api_url() + config.get_o2ims_api_base()
86 globalcloudId = stxobj.id # to be updated
87 description = "An ocloud"
88 ocloud = Ocloud(stxobj.id, stxobj.name, imsendpoint,
89 globalcloudId, description, 1)
90 ocloud.createtime = stxobj.createtime
91 ocloud.updatetime = stxobj.updatetime
92 ocloud.hash = stxobj.hash
97 def update_by(ocloud: Ocloud, stxobj: StxGenericModel) -> None:
98 if ocloud.oCloudId != stxobj.id:
99 raise MismatchedModel("More than 1 ocloud found")
100 ocloud.name = stxobj.name
101 ocloud.createtime = stxobj.createtime
102 ocloud.updatetime = stxobj.updatetime
103 # ocloud.content = stxobj.content
104 ocloud.hash = stxobj.hash
105 ocloud.version_number = ocloud.version_number + 1
106 ocloud.events.append(events.OcloudChanged(
108 notificationEventType=NotificationEventEnum.MODIFY,
109 updatetime=stxobj.updatetime