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, Tuple
  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, *args) -> List[ocloud.Ocloud]:
  35         return self._list(*args)
  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 _list(self, *args) -> List[ocloud.Ocloud]:
  54         raise NotImplementedError
  55
  56     @abc.abstractmethod
  57     def _update(self, ocloud: ocloud.Ocloud):
  58         raise NotImplementedError
  59
  60
  61 class ResourceTypeRepository(abc.ABC):
  62     def __init__(self):
  63         self.seen = set()  # type: Set[ocloud.ResourceType]
  64
  65     def add(self, resource_type: ocloud.ResourceType):
  66         self._add(resource_type)
  67         self.seen.add(resource_type)
  68
  69     def get(self, resource_type_id) -> ocloud.ResourceType:
  70         resource_type = self._get(resource_type_id)
  71         if resource_type:
  72             self.seen.add(resource_type)
  73         return resource_type
  74
  75     def get_by_name(self, resource_type_name) -> ocloud.ResourceType:
  76         resource_type = self._get_by_name(resource_type_name)
  77         if resource_type:
  78             self.seen.add(resource_type)
  79         return resource_type
  80
  81     def list(self, *args) -> List[ocloud.ResourceType]:
  82         return self._list(*args)[1]
  83
  84     def list_with_count(self, *args, **kwargs) -> \
  85             Tuple[int, List[ocloud.ResourceType]]:
  86         return self._list(*args, **kwargs)
  87
  88     def update(self, resource_type: ocloud.ResourceType):
  89         self._update(resource_type)
  90         self.seen.add(resource_type)
  91
  92     @abc.abstractmethod
  93     def _add(self, resource_type: ocloud.ResourceType):
  94         raise NotImplementedError
  95
  96     @abc.abstractmethod
  97     def _get(self, resource_type_id) -> ocloud.ResourceType:
  98         raise NotImplementedError
  99
 100     @abc.abstractmethod
 101     def _get_by_name(self, resource_type_name) -> ocloud.ResourceType:
 102         raise NotImplementedError
 103
 104     @abc.abstractmethod
 105     def _list(self, **kwargs) -> Tuple[int, List[ocloud.ResourceType]]:
 106         raise NotImplementedError
 107
 108     @abc.abstractmethod
 109     def _update(self, resource_type: ocloud.ResourceType):
 110         raise NotImplementedError
 111
 112
 113 class ResourcePoolRepository(abc.ABC):
 114     def __init__(self):
 115         self.seen = set()  # type: Set[ocloud.ResourcePool]
 116
 117     def add(self, resource_pool: ocloud.ResourcePool):
 118         self._add(resource_pool)
 119         self.seen.add(resource_pool)
 120
 121     def get(self, resource_pool_id) -> ocloud.ResourcePool:
 122         resource_pool = self._get(resource_pool_id)
 123         if resource_pool:
 124             self.seen.add(resource_pool)
 125         return resource_pool
 126
 127     def list(self, *args) -> List[ocloud.ResourcePool]:
 128         return self._list(*args)[1]
 129
 130     def list_with_count(self, *args, **kwargs) -> \
 131             Tuple[int, List[ocloud.ResourcePool]]:
 132         return self._list(*args, **kwargs)
 133
 134     def update(self, resource_pool: ocloud.ResourcePool):
 135         self._update(resource_pool)
 136         self.seen.add(resource_pool)
 137
 138     @abc.abstractmethod
 139     def _add(self, resource_pool: ocloud.ResourcePool):
 140         raise NotImplementedError
 141
 142     @abc.abstractmethod
 143     def _get(self, resource_pool_id) -> ocloud.ResourcePool:
 144         raise NotImplementedError
 145
 146     @abc.abstractmethod
 147     def _list(self, **kwargs) -> Tuple[int, List[ocloud.ResourcePool]]:
 148         raise NotImplementedError
 149
 150     @abc.abstractmethod
 151     def _update(self, resource_pool: ocloud.ResourcePool):
 152         raise NotImplementedError
 153
 154
 155 class ResourceRepository(abc.ABC):
 156     def __init__(self):
 157         self.seen = set()  # type: Set[ocloud.Resource]
 158
 159     def add(self, resource: ocloud.Resource):
 160         self._add(resource)
 161         self.seen.add(resource)
 162
 163     def get(self, resource_id) -> ocloud.Resource:
 164         resource = self._get(resource_id)
 165         if resource:
 166             self.seen.add(resource)
 167         return resource
 168
 169     def list(self, resourcepool_id, *args) -> List[ocloud.Resource]:
 170         return self._list(resourcepool_id, *args)[1]
 171
 172     def list_with_count(self, resourcepool_id, *args, **kwargs) -> \
 173             Tuple[int, List[ocloud.Resource]]:
 174         return self._list(resourcepool_id, *args, **kwargs)
 175
 176     def update(self, resource: ocloud.Resource):
 177         self._update(resource)
 178         self.seen.add(resource)
 179
 180     @abc.abstractmethod
 181     def _add(self, resource: ocloud.Resource):
 182         raise NotImplementedError
 183
 184     @abc.abstractmethod
 185     def _get(self, resource_id) -> ocloud.Resource:
 186         raise NotImplementedError
 187
 188     @abc.abstractmethod
 189     def _list(self, resourcepool_id, **kwargs) -> \
 190             Tuple[int, List[ocloud.Resource]]:
 191         raise NotImplementedError
 192
 193     @abc.abstractmethod
 194     def _update(self, resource: ocloud.Resource):
 195         raise NotImplementedError
 196
 197
 198 class DeploymentManagerRepository(abc.ABC):
 199     def __init__(self):
 200         self.seen = set()  # type: Set[ocloud.DeploymentManager]
 201
 202     def add(self, deployment_manager: ocloud.DeploymentManager):
 203         self._add(deployment_manager)
 204         self.seen.add(deployment_manager)
 205
 206     def get(self, deployment_manager_id) -> ocloud.DeploymentManager:
 207         deployment_manager = self._get(deployment_manager_id)
 208         if deployment_manager:
 209             self.seen.add(deployment_manager)
 210         return deployment_manager
 211
 212     def list(self, *args) -> List[ocloud.DeploymentManager]:
 213         return self._list(*args)[1]
 214
 215     def list_with_count(self, *args, **kwargs) -> \
 216             Tuple[int, List[ocloud.DeploymentManager]]:
 217         return self._list(*args, **kwargs)
 218
 219     def update(self, deployment_manager: ocloud.DeploymentManager):
 220         self._update(deployment_manager)
 221
 222     @abc.abstractmethod
 223     def _add(self, deployment_manager: ocloud.DeploymentManager):
 224         raise NotImplementedError
 225
 226     @abc.abstractmethod
 227     def _get(self, deployment_manager_id) -> ocloud.DeploymentManager:
 228         raise NotImplementedError
 229
 230     @abc.abstractmethod
 231     def _list(self, **kwargs) -> Tuple[int, List[ocloud.DeploymentManager]]:
 232         raise NotImplementedError
 233
 234     @abc.abstractmethod
 235     def _update(self, deployment_manager: ocloud.DeploymentManager):
 236         raise NotImplementedError