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