o2ims/adapter/ocloud_repository.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.adapter import orm
  18 from o2ims.domain import ocloud
  19
  20
  21 class OcloudRepository(abc.ABC):
  22     def __init__(self):
  23         self.seen = set()  # type: Set[ocloud.Ocloud]
  24
  25     def add(self, ocloud: ocloud.Ocloud):
  26         self._add(ocloud)
  27         self.seen.add(ocloud)
  28
  29     def get(self, ocloudid) -> ocloud.Ocloud:
  30         ocloud = self._get(ocloudid)
  31         if ocloud:
  32             self.seen.add(ocloud)
  33         return ocloud
  34
  35     def list(self) -> List[ocloud.Ocloud]:
  36         return self._list()
  37
  38     def update(self, ocloud: ocloud.Ocloud):
  39         self._update(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, ocloudid) -> ocloud.Ocloud:
  50         raise NotImplementedError
  51
  52     @abc.abstractmethod
  53     def _update(self, ocloud: ocloud.Ocloud):
  54         raise NotImplementedError
  55
  56
  57 class OcloudSqlAlchemyRepository(OcloudRepository):
  58     def __init__(self, session):
  59         super().__init__()
  60         self.session = session
  61
  62     def _add(self, ocloud: ocloud.Ocloud):
  63         self.session.add(ocloud)
  64         # self.session.add_all(ocloud.deploymentManagers)
  65
  66     def _get(self, ocloudid) -> ocloud.Ocloud:
  67         return self.session.query(ocloud.Ocloud).filter_by(
  68             oCloudId=ocloudid).first()
  69
  70     def _list(self) -> List[ocloud.Ocloud]:
  71         return self.session.query()
  72
  73     def _update(self, ocloud: ocloud.Ocloud):
  74         self.session.add(ocloud)