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, conf
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
48 oclouds = uow.oclouds.list()
49 if oclouds and oclouds.count() > 1:
50 raise InvalidOcloudState("More than 1 ocloud is found")
51 elif not oclouds or oclouds.count() == 0:
52 logger.info("add ocloud:" + stxobj.name
53 + " update_at: " + str(stxobj.updatetime)
54 + " id: " + str(stxobj.id)
55 + " hash: " + str(stxobj.hash))
56 entry = create_by(stxobj)
57 uow.oclouds.add(entry)
59 logger.info("Add the ocloud: " + stxobj.id
60 + ", name: " + stxobj.name)
62 localmodel = oclouds.first()
63 if is_outdated(localmodel, stxobj):
64 logger.info("update ocloud:" + stxobj.name
65 + " update_at: " + str(stxobj.updatetime)
66 + " id: " + str(stxobj.id)
67 + " hash: " + str(stxobj.hash))
68 update_by(localmodel, stxobj)
69 uow.oclouds.update(localmodel)
71 logger.info("Update the ocloud: " + stxobj.id
72 + ", name: " + stxobj.name)
76 def is_outdated(ocloud: Ocloud, stxobj: StxGenericModel):
77 # if stxobj.updatetime:
78 # return True if Ocloud.updatetime < stxobj.updatetime else False
80 return True if ocloud.hash != stxobj.hash else False
83 def create_by(stxobj: StxGenericModel) -> Ocloud:
84 imsendpoint = config.get_api_url() + config.get_o2ims_api_base() + '/'
85 globalcloudId = conf.DEFAULT.ocloud_global_id
86 description = "An ocloud"
87 ocloud = Ocloud(stxobj.id, stxobj.name, imsendpoint,
88 globalcloudId, description, 1)
89 ocloud.createtime = stxobj.createtime
90 ocloud.updatetime = stxobj.updatetime
91 ocloud.hash = stxobj.hash
92 ocloud.events.append(events.OcloudChanged(
94 notificationEventType=NotificationEventEnum.CREATE,
95 updatetime=stxobj.updatetime
101 def update_by(ocloud: Ocloud, stxobj: StxGenericModel) -> None:
102 if ocloud.oCloudId != stxobj.id:
103 raise MismatchedModel("More than 1 ocloud found")
104 ocloud.name = stxobj.name
105 ocloud.createtime = stxobj.createtime
106 ocloud.updatetime = stxobj.updatetime
107 # ocloud.content = stxobj.content
108 ocloud.hash = stxobj.hash
109 ocloud.version_number = ocloud.version_number + 1
110 ocloud.events.append(events.OcloudChanged(
112 notificationEventType=NotificationEventEnum.MODIFY,
113 updatetime=stxobj.updatetime