src/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 Set
  17 from o2ims.adapter import orm
  18 from o2ims.domain import ocloud
  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, ocloudid) -> ocloud.Ocloud:
  29         ocloud = self._get(ocloudid)
  30         if ocloud:
  31             self.seen.add(ocloud)
  32         return ocloud
  33
  34     def update(self, ocloud: ocloud.Ocloud):
  35         self._update(ocloud)
  36
  37     # def update_fields(self, ocloudid: str, updatefields: dict):
  38     #     self._update(ocloudid, updatefields)
  39
  40     @abc.abstractmethod
  41     def _add(self, ocloud: ocloud.Ocloud):
  42         raise NotImplementedError
  43
  44     @abc.abstractmethod
  45     def _get(self, ocloudid) -> ocloud.Ocloud:
  46         raise NotImplementedError
  47
  48     @abc.abstractmethod
  49     def _update(self, ocloud: ocloud.Ocloud):
  50         raise NotImplementedError
  51
  52
  53 class OcloudSqlAlchemyRepository(OcloudRepository):
  54     def __init__(self, session):
  55         super().__init__()
  56         self.session = session
  57
  58     def _add(self, ocloud: ocloud.Ocloud):
  59         self.session.add(ocloud)
  60         # self.session.add_all(ocloud.deploymentManagers)
  61
  62     def _get(self, ocloudid) -> ocloud.Ocloud:
  63         return self.session.query(ocloud.Ocloud).filter_by(oCloudId=ocloudid).first()
  64
  65     def _update(self, ocloud: ocloud.Ocloud):
  66         self.session.add(ocloud)
  67
  68     # def _update_fields(self, ocloudid: str, updatefields: dict):
  69     #     dmslist = updatefields.pop("deploymentManagers", None)
  70     #     if dmslist:
  71     #         self._update_dms_list(dmslist)
  72     #     if updatefields:
  73     #         self.session.query(ocloud.Ocloud).filter_by(oCloudId=ocloudid).update(updatefields)
  74
  75     # def _update_dms_list(self, dms_list: list):
  76     #     for dms in dms_list or []:
  77     #         self.session.query(ocloud.DeploymentManager).filter_by(deploymentManagerId=dms.deploymentManagerId).update(dms)