+++ /dev/null
-#!/usr/bin/env python3
-# Copyright 2022 Canonical Ltd.
-#
-# Licensed under the Apache License, Version 2.0 (the "License"); you may
-# not use this file except in compliance with the License. You may obtain
-# a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-# License for the specific language governing permissions and limitations
-# under the License.
-#
-# For those usages not covered by the Apache License, Version 2.0 please
-# contact: legal@canonical.com
-#
-# To get in touch with the maintainers, please contact:
-# osm-charmers@lists.launchpad.net
-#
-# flake8: noqa
-
-import ops
-
-
-class BaseRelationClient(ops.framework.Object):
- """Requires side of a Kafka Endpoint"""
-
- def __init__(
- self, charm: ops.charm.CharmBase, relation_name: str, mandatory_fields: list = []
- ):
- super().__init__(charm, relation_name)
- self.relation_name = relation_name
- self.mandatory_fields = mandatory_fields
- self._update_relation()
-
- def get_data_from_unit(self, key: str):
- if not self.relation:
- # This update relation doesn't seem to be needed, but I added it because apparently
- # the data is empty in the unit tests.
- # In reality, the constructor is called in every hook.
- # In the unit tests when doing an update_relation_data, apparently it is not called.
- self._update_relation()
- if self.relation:
- for unit in self.relation.units:
- data = self.relation.data[unit].get(key)
- if data:
- return data
-
- def get_data_from_app(self, key: str):
- if not self.relation or self.relation.app not in self.relation.data:
- # This update relation doesn't seem to be needed, but I added it because apparently
- # the data is empty in the unit tests.
- # In reality, the constructor is called in every hook.
- # In the unit tests when doing an update_relation_data, apparently it is not called.
- self._update_relation()
- if self.relation and self.relation.app in self.relation.data:
- data = self.relation.data[self.relation.app].get(key)
- if data:
- return data
-
- def is_missing_data_in_unit(self):
- return not all([self.get_data_from_unit(field) for field in self.mandatory_fields])
-
- def is_missing_data_in_app(self):
- return not all([self.get_data_from_app(field) for field in self.mandatory_fields])
-
- def _update_relation(self):
- self.relation = self.framework.model.get_relation(self.relation_name)
-
-
-class MongoClient(BaseRelationClient):
- """Requires side of a Mongo Endpoint"""
-
- mandatory_fields_mapping = {
- "reactive": ["connection_string"],
- "ops": ["replica_set_uri", "replica_set_name"],
- }
-
- def __init__(self, charm: ops.charm.CharmBase, relation_name: str):
- super().__init__(charm, relation_name, mandatory_fields=[])
-
- @property
- def connection_string(self):
- if self.is_opts():
- replica_set_uri = self.get_data_from_unit("replica_set_uri")
- replica_set_name = self.get_data_from_unit("replica_set_name")
- return f"{replica_set_uri}?replicaSet={replica_set_name}"
- else:
- return self.get_data_from_unit("connection_string")
-
- def is_opts(self):
- return not self.is_missing_data_in_unit_ops()
-
- def is_missing_data_in_unit(self):
- return self.is_missing_data_in_unit_ops() and self.is_missing_data_in_unit_reactive()
-
- def is_missing_data_in_unit_ops(self):
- return not all(
- [self.get_data_from_unit(field) for field in self.mandatory_fields_mapping["ops"]]
- )
-
- def is_missing_data_in_unit_reactive(self):
- return not all(
- [self.get_data_from_unit(field) for field in self.mandatory_fields_mapping["reactive"]]
- )