n2vc/vca/connection.py

   1 # Copyright 2021 Canonical Ltd.
   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 typing
  16
  17 from n2vc.config import EnvironConfig, ModelConfig
  18 from n2vc.store import Store
  19 from n2vc.vca.cloud import Cloud
  20 from n2vc.vca.connection_data import ConnectionData
  21
  22
  23 class Connection:
  24     def __init__(self, store: Store, vca_id: str = None):
  25         """
  26         Contructor
  27
  28         :param: store: Store object. Used to communicate wuth the DB
  29         :param: vca_id: Id of the VCA. If none specified, the default VCA will be used.
  30         """
  31         self._data = None
  32         self.default = vca_id is None
  33         self._vca_id = vca_id
  34         self._store = store
  35
  36     async def load(self):
  37         """Load VCA connection data"""
  38         await self._load_vca_connection_data()
  39
  40     @property
  41     def is_default(self):
  42         return self._vca_id is None
  43
  44     @property
  45     def data(self) -> ConnectionData:
  46         return self._data
  47
  48     async def _load_vca_connection_data(self) -> typing.NoReturn:
  49         """
  50         Load VCA connection data
  51
  52         If self._vca_id is None, it will get the VCA data from the Environment variables,
  53         and the default VCA will be used. If it is not None, then it means that it will
  54         load the credentials from the database (A non-default VCA will be used).
  55         """
  56         if self._vca_id:
  57             self._data = await self._store.get_vca_connection_data(self._vca_id)
  58         else:
  59             envs = EnvironConfig()
  60             # Get endpoints from the DB and ENV. Check if update in the database is needed or not.
  61             db_endpoints = await self._store.get_vca_endpoints()
  62             env_endpoints = (
  63                 envs["endpoints"].split(",")
  64                 if "endpoints" in envs
  65                 else ["{}:{}".format(envs["host"], envs.get("port", 17070))]
  66             )
  67
  68             db_update_needed = not all(e in db_endpoints for e in env_endpoints)
  69
  70             endpoints = env_endpoints if db_update_needed else db_endpoints
  71             config = {
  72                 "endpoints": endpoints,
  73                 "user": envs["user"],
  74                 "secret": envs["secret"],
  75                 "cacert": envs["cacert"],
  76                 "pubkey": envs["pubkey"],
  77                 "lxd-cloud": envs["cloud"],
  78                 "lxd-credentials": envs.get("credentials", envs["cloud"]),
  79                 "k8s-cloud": envs["k8s_cloud"],
  80                 "k8s-credentials": envs.get("k8s_credentials", envs["k8s_cloud"]),
  81                 "model-config": ModelConfig(envs),
  82                 "api-proxy": envs.get("api_proxy", None),
  83             }
  84             self._data = ConnectionData(**config)
  85             if db_update_needed:
  86                 await self.update_endpoints(endpoints)
  87
  88     @property
  89     def endpoints(self):
  90         return self._data.endpoints
  91
  92     async def update_endpoints(self, endpoints: typing.List[str]):
  93         await self._store.update_vca_endpoints(endpoints, self._vca_id)
  94         self._data.endpoints = endpoints
  95
  96     @property
  97     def lxd_cloud(self) -> Cloud:
  98         return Cloud(self.data.lxd_cloud, self.data.lxd_credentials)
  99
 100     @property
 101     def k8s_cloud(self) -> Cloud:
 102         return Cloud(self.data.k8s_cloud, self.data.k8s_credentials)
 103
 104
 105 async def get_connection(store: Store, vca_id: str = None) -> Connection:
 106     """
 107     Get Connection
 108
 109     Method to get a Connection object with the VCA information loaded
 110     """
 111     connection = Connection(store, vca_id=vca_id)
 112     await connection.load()
 113     return connection