1 # Copyright 2021 Canonical Ltd.
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
7 # http://www.apache.org/licenses/LICENSE-2.0
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.
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
24 def __init__(self
, store
: Store
, vca_id
: str = None):
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.
32 self
.default
= vca_id
is None
37 """Load VCA connection data"""
38 await self
._load
_vca
_connection
_data
()
42 return self
._vca
_id
is None
45 def data(self
) -> ConnectionData
:
48 async def _load_vca_connection_data(self
) -> typing
.NoReturn
:
50 Load VCA connection data
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).
57 self
._data
= await self
._store
.get_vca_connection_data(self
._vca
_id
)
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()
63 envs
["endpoints"].split(",")
64 if "endpoints" in envs
65 else ["{}:{}".format(envs
["host"], envs
.get("port", 17070))]
68 db_update_needed
= not all(e
in db_endpoints
for e
in env_endpoints
)
70 endpoints
= env_endpoints
if db_update_needed
else db_endpoints
72 "endpoints": endpoints
,
74 "secret": envs
["secret"],
75 "cacert": envs
["cacert"],
76 "pubkey": envs
.get("pubkey"),
77 "lxd-cloud": envs
.get("cloud"),
78 "lxd-credentials": envs
.get("credentials", envs
.get("cloud")),
79 "k8s-cloud": envs
.get("k8s_cloud"),
80 "k8s-credentials": envs
.get("k8s_credentials", envs
.get("k8s_cloud")),
81 "model-config": ModelConfig(envs
),
82 "api-proxy": envs
.get("api_proxy", None),
84 self
._data
= ConnectionData(**config
)
86 await self
.update_endpoints(endpoints
)
90 return self
._data
.endpoints
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
97 def lxd_cloud(self
) -> Cloud
:
98 return Cloud(self
.data
.lxd_cloud
, self
.data
.lxd_credentials
)
101 def k8s_cloud(self
) -> Cloud
:
102 return Cloud(self
.data
.k8s_cloud
, self
.data
.k8s_credentials
)
105 async def get_connection(store
: Store
, vca_id
: str = None) -> Connection
:
109 Method to get a Connection object with the VCA information loaded
111 connection
= Connection(store
, vca_id
=vca_id
)
112 await connection
.load()