o2ims/domain/ocloud_repo.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 abc
  16 from typing import List, Set
  17 from o2ims.domain import ocloud
  18
  19
  20 class OcloudRepository(abc.ABC):
  21     def __init__(self):
  22         self.seen = set()  # type: Set[ocloud.Ocloud]
  23
  24     def add(self, ocloud: ocloud.Ocloud):
  25         self._add(ocloud)
  26         self.seen.add(ocloud)
  27
  28     def get(self, ocloud_id) -> ocloud.Ocloud:
  29         ocloud = self._get(ocloud_id)
  30         if ocloud:
  31             self.seen.add(ocloud)
  32         return ocloud
  33
  34     def list(self) -> List[ocloud.Ocloud]:
  35         return self._list()
  36
  37     def update(self, ocloud: ocloud.Ocloud):
  38         self._update(ocloud)
  39         self.seen.add(ocloud)
  40
  41     # def update_fields(self, ocloudid: str, updatefields: dict):
  42     #     self._update(ocloudid, updatefields)
  43
  44     @abc.abstractmethod
  45     def _add(self, ocloud: ocloud.Ocloud):
  46         raise NotImplementedError
  47
  48     @abc.abstractmethod
  49     def _get(self, ocloud_id) -> ocloud.Ocloud:
  50         raise NotImplementedError
  51
  52     @abc.abstractmethod
  53     def _update(self, ocloud: ocloud.Ocloud):
  54         raise NotImplementedError
  55
  56
  57 class ResourceTypeRepository(abc.ABC):
  58     def __init__(self):
  59         self.seen = set()  # type: Set[ocloud.ResourceType]
  60
  61     def add(self, resource_type: ocloud.ResourceType):
  62         self._add(resource_type)
  63         self.seen.add(resource_type)
  64
  65     def get(self, resource_type_id) -> ocloud.ResourceType:
  66         resource_type = self._get(resource_type_id)
  67         if resource_type:
  68             self.seen.add(resource_type)
  69         return resource_type
  70
  71     def get_by_name(self, resource_type_name) -> ocloud.ResourceType:
  72         resource_type = self._get_by_name(resource_type_name)
  73         if resource_type:
  74             self.seen.add(resource_type)
  75         return resource_type
  76
  77     def list(self) -> List[ocloud.ResourceType]:
  78         return self._list()
  79
  80     def update(self, resource_type: ocloud.ResourceType):
  81         self._update(resource_type)
  82         self.seen.add(resource_type)
  83
  84     @abc.abstractmethod
  85     def _add(self, resource_type: ocloud.ResourceType):
  86         raise NotImplementedError
  87
  88     @abc.abstractmethod
  89     def _get(self, resource_type_id) -> ocloud.ResourceType:
  90         raise NotImplementedError
  91
  92     @abc.abstractmethod
  93     def _get_by_name(self, resource_type_name) -> ocloud.ResourceType:
  94         raise NotImplementedError
  95
  96     @abc.abstractmethod
  97     def _update(self, resource_type: ocloud.ResourceType):
  98         raise NotImplementedError
  99
 100
 101 class ResourcePoolRepository(abc.ABC):
 102     def __init__(self):
 103         self.seen = set()  # type: Set[ocloud.ResourcePool]
 104
 105     def add(self, resource_pool: ocloud.ResourcePool):
 106         self._add(resource_pool)
 107         self.seen.add(resource_pool)
 108
 109     def get(self, resource_pool_id) -> ocloud.ResourcePool:
 110         resource_pool = self._get(resource_pool_id)
 111         if resource_pool:
 112             self.seen.add(resource_pool)
 113         return resource_pool
 114
 115     def list(self) -> List[ocloud.ResourcePool]:
 116         return self._list()
 117
 118     def update(self, resource_pool: ocloud.ResourcePool):
 119         self._update(resource_pool)
 120         self.seen.add(resource_pool)
 121
 122     @abc.abstractmethod
 123     def _add(self, resource_pool: ocloud.ResourcePool):
 124         raise NotImplementedError
 125
 126     @abc.abstractmethod
 127     def _get(self, resource_pool_id) -> ocloud.ResourcePool:
 128         raise NotImplementedError
 129
 130     @abc.abstractmethod
 131     def _update(self, resource_pool: ocloud.ResourcePool):
 132         raise NotImplementedError
 133
 134
 135 class ResourceRepository(abc.ABC):
 136     def __init__(self):
 137         self.seen = set()  # type: Set[ocloud.Resource]
 138
 139     def add(self, resource: ocloud.Resource):
 140         self._add(resource)
 141         self.seen.add(resource)
 142
 143     def get(self, resource_id) -> ocloud.Resource:
 144         resource = self._get(resource_id)
 145         if resource:
 146             self.seen.add(resource)
 147         return resource
 148
 149     def list(self, resourcepool_id, **kwargs) -> List[ocloud.Resource]:
 150         return self._list(resourcepool_id, **kwargs)
 151
 152     def update(self, resource: ocloud.Resource):
 153         self._update(resource)
 154         self.seen.add(resource)
 155
 156     @abc.abstractmethod
 157     def _add(self, resource: ocloud.Resource):
 158         raise NotImplementedError
 159
 160     @abc.abstractmethod
 161     def _get(self, resource_id) -> ocloud.Resource:
 162         raise NotImplementedError
 163
 164     @abc.abstractmethod
 165     def _list(self, resourcepool_id, **kwargs) -> ocloud.Resource:
 166         raise NotImplementedError
 167
 168     @abc.abstractmethod
 169     def _update(self, resource: ocloud.Resource):
 170         raise NotImplementedError
 171
 172
 173 class DeploymentManagerRepository(abc.ABC):
 174     def __init__(self):
 175         self.seen = set()  # type: Set[ocloud.DeploymentManager]
 176
 177     def add(self, deployment_manager: ocloud.DeploymentManager):
 178         self._add(deployment_manager)
 179         self.seen.add(deployment_manager)
 180
 181     def get(self, deployment_manager_id) -> ocloud.DeploymentManager:
 182         deployment_manager = self._get(deployment_manager_id)
 183         if deployment_manager:
 184             self.seen.add(deployment_manager)
 185         return deployment_manager
 186
 187     def list(self) -> List[ocloud.DeploymentManager]:
 188         return self._list()
 189
 190     def update(self, deployment_manager: ocloud.DeploymentManager):
 191         self._update(deployment_manager)
 192
 193     @abc.abstractmethod
 194     def _add(self, deployment_manager: ocloud.DeploymentManager):
 195         raise NotImplementedError
 196
 197     @abc.abstractmethod
 198     def _get(self, deployment_manager_id) -> ocloud.DeploymentManager:
 199         raise NotImplementedError
 200
 201     @abc.abstractmethod
 202     def _update(self, deployment_manager: ocloud.DeploymentManager):
 203         raise NotImplementedError