installers/charm/osm-lcm/src/legacy_interfaces.py

   1 #!/usr/bin/env python3
   2 # Copyright 2022 Canonical Ltd.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 # not use this file except in compliance with the License. You may obtain
   6 # a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 # License for the specific language governing permissions and limitations
  14 # under the License.
  15 #
  16 # For those usages not covered by the Apache License, Version 2.0 please
  17 # contact: legal@canonical.com
  18 #
  19 # To get in touch with the maintainers, please contact:
  20 # osm-charmers@lists.launchpad.net
  21 #
  22 # flake8: noqa
  23
  24 import ops
  25
  26
  27 class BaseRelationClient(ops.framework.Object):
  28     """Requires side of a Kafka Endpoint"""
  29
  30     def __init__(
  31         self, charm: ops.charm.CharmBase, relation_name: str, mandatory_fields: list = []
  32     ):
  33         super().__init__(charm, relation_name)
  34         self.relation_name = relation_name
  35         self.mandatory_fields = mandatory_fields
  36         self._update_relation()
  37
  38     def get_data_from_unit(self, key: str):
  39         if not self.relation:
  40             # This update relation doesn't seem to be needed, but I added it because apparently
  41             # the data is empty in the unit tests.
  42             # In reality, the constructor is called in every hook.
  43             # In the unit tests when doing an update_relation_data, apparently it is not called.
  44             self._update_relation()
  45         if self.relation:
  46             for unit in self.relation.units:
  47                 data = self.relation.data[unit].get(key)
  48                 if data:
  49                     return data
  50
  51     def get_data_from_app(self, key: str):
  52         if not self.relation or self.relation.app not in self.relation.data:
  53             # This update relation doesn't seem to be needed, but I added it because apparently
  54             # the data is empty in the unit tests.
  55             # In reality, the constructor is called in every hook.
  56             # In the unit tests when doing an update_relation_data, apparently it is not called.
  57             self._update_relation()
  58         if self.relation and self.relation.app in self.relation.data:
  59             data = self.relation.data[self.relation.app].get(key)
  60             if data:
  61                 return data
  62
  63     def is_missing_data_in_unit(self):
  64         return not all([self.get_data_from_unit(field) for field in self.mandatory_fields])
  65
  66     def is_missing_data_in_app(self):
  67         return not all([self.get_data_from_app(field) for field in self.mandatory_fields])
  68
  69     def _update_relation(self):
  70         self.relation = self.framework.model.get_relation(self.relation_name)
  71
  72
  73 class MongoClient(BaseRelationClient):
  74     """Requires side of a Mongo Endpoint"""
  75
  76     mandatory_fields_mapping = {
  77         "reactive": ["connection_string"],
  78         "ops": ["replica_set_uri", "replica_set_name"],
  79     }
  80
  81     def __init__(self, charm: ops.charm.CharmBase, relation_name: str):
  82         super().__init__(charm, relation_name, mandatory_fields=[])
  83
  84     @property
  85     def connection_string(self):
  86         if self.is_opts():
  87             replica_set_uri = self.get_data_from_unit("replica_set_uri")
  88             replica_set_name = self.get_data_from_unit("replica_set_name")
  89             return f"{replica_set_uri}?replicaSet={replica_set_name}"
  90         else:
  91             return self.get_data_from_unit("connection_string")
  92
  93     def is_opts(self):
  94         return not self.is_missing_data_in_unit_ops()
  95
  96     def is_missing_data_in_unit(self):
  97         return self.is_missing_data_in_unit_ops() and self.is_missing_data_in_unit_reactive()
  98
  99     def is_missing_data_in_unit_ops(self):
 100         return not all(
 101             [self.get_data_from_unit(field) for field in self.mandatory_fields_mapping["ops"]]
 102         )
 103
 104     def is_missing_data_in_unit_reactive(self):
 105         return not all(
 106             [self.get_data_from_unit(field) for field in self.mandatory_fields_mapping["reactive"]]
 107         )