Docs: Enable O2 DMS by exposing k8s API endpoint
[pti/o2.git] / o2dms / domain / dms.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 from __future__ import annotations
16 # from os import stat
17 import json
18 from o2dms.domain import events
19 from o2dms.domain.states import NfDeploymentState
20
21 from o2common.domain.base import AgRoot, Serializer
22
23
24 class NfDeploymentDesc(AgRoot, Serializer):
25     def __init__(self, id: str, name: str, dmsId: str, description: str = '',
26                  inputParams: str = '', outputParams: str = '',
27                  artifactRepoUrl: str = '', artifactName: str = '') -> None:
28         super().__init__()
29         self.id = id
30         self.version_number = 0
31         self.deploymentManagerId = dmsId
32         self.name = name
33         self.description = description
34         self.artifactRepoUrl = artifactRepoUrl
35         self.artifactName = artifactName
36         self.status = 0
37
38         if type(inputParams) is str:
39             inputParams = json.loads(inputParams)
40         self.inputParams = json.dumps(inputParams)
41
42         if type(outputParams) is str:
43             outputParams = json.loads(outputParams)
44         self.outputParams = json.dumps(outputParams)
45
46         # self.extensions = []
47
48
49 class NfDeployment(AgRoot, Serializer):
50     def __init__(self, id: str, name: str, dmsId: str, description: str = '',
51                  descriptorId: str = '', parentId: str = '',) -> None:
52         super().__init__()
53         self.id = id
54         self.version_number = 0
55         self.deploymentManagerId = dmsId
56         self.name = name
57         self.description = description
58         self.descriptorId = descriptorId
59         self.parentDeploymentId = parentId
60         self.status = NfDeploymentState.Initial
61
62     def transit_state(self, state: NfDeploymentState):
63         if (self.status != state):
64             self._append_event(self.status, state)
65             # self.status = state
66
67     def set_state(self, state: NfDeploymentState):
68         if (self.status != state):
69             self.status = state
70
71     def _append_event(self, fromState, toState):
72         if not hasattr(self, "events"):
73             self.events = []
74         self.events.append(
75             events.NfDeploymentStateChanged(
76                 NfDeploymentId=self.id, FromState=fromState, ToState=toState))
77
78
79 class NfOCloudVResource(AgRoot, Serializer):
80     def __init__(self, id: str, name: str, dmsId: str, description: str = '',
81                  descriptorId: str = '', nfDeploymentId: str = '',
82                  vresourceType: int = 0,) -> None:
83         super().__init__()
84         self.id = id
85         self.version_number = 0
86         self.deploymentManagerId = dmsId
87         self.name = name
88         self.description = description
89         self.descriptorId = descriptorId
90         self.nfDeploymentId = nfDeploymentId
91         self.vresourceType = vresourceType
92         self.status = 0
93         self.metadata = []