diff --git a/installers/charm/osm-lcm/lib/charms/data_platform_libs/v0/data_interfaces.py b/installers/charm/osm-lcm/lib/charms/data_platform_libs/v0/data_interfaces.py
new file mode 100644
index 0000000..b3da5aa
--- /dev/null
+++ b/installers/charm/osm-lcm/lib/charms/data_platform_libs/v0/data_interfaces.py
@@ -0,0 +1,1130 @@
+# Copyright 2023 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.
+
+"""Library to manage the relation for the data-platform products.
+
+This library contains the Requires and Provides classes for handling the relation
+between an application and multiple managed application supported by the data-team:
+MySQL, Postgresql, MongoDB, Redis,  and Kakfa.
+
+### Database (MySQL, Postgresql, MongoDB, and Redis)
+
+#### Requires Charm
+This library is a uniform interface to a selection of common database
+metadata, with added custom events that add convenience to database management,
+and methods to consume the application related data.
+
+
+Following an example of using the DatabaseCreatedEvent, in the context of the
+application charm code:
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    DatabaseCreatedEvent,
+    DatabaseRequires,
+)
+
+class ApplicationCharm(CharmBase):
+    # Application charm that connects to database charms.
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Charm events defined in the database requires charm library.
+        self.database = DatabaseRequires(self, relation_name="database", database_name="database")
+        self.framework.observe(self.database.on.database_created, self._on_database_created)
+
+    def _on_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+
+        # Start application with rendered configuration
+        self._start_application(config_file)
+
+        # Set active status
+        self.unit.status = ActiveStatus("received database credentials")
+```
+
+As shown above, the library provides some custom events to handle specific situations,
+which are listed below:
+
+-  database_created: event emitted when the requested database is created.
+-  endpoints_changed: event emitted when the read/write endpoints of the database have changed.
+-  read_only_endpoints_changed: event emitted when the read-only endpoints of the database
+  have changed. Event is not triggered if read/write endpoints changed too.
+
+If it is needed to connect multiple database clusters to the same relation endpoint
+the application charm can implement the same code as if it would connect to only
+one database cluster (like the above code example).
+
+To differentiate multiple clusters connected to the same relation endpoint
+the application charm can use the name of the remote application:
+
+```python
+
+def _on_database_created(self, event: DatabaseCreatedEvent) -> None:
+    # Get the remote app name of the cluster that triggered this event
+    cluster = event.relation.app.name
+```
+
+It is also possible to provide an alias for each different database cluster/relation.
+
+So, it is possible to differentiate the clusters in two ways.
+The first is to use the remote application name, i.e., `event.relation.app.name`, as above.
+
+The second way is to use different event handlers to handle each cluster events.
+The implementation would be something like the following code:
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    DatabaseCreatedEvent,
+    DatabaseRequires,
+)
+
+class ApplicationCharm(CharmBase):
+    # Application charm that connects to database charms.
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Define the cluster aliases and one handler for each cluster database created event.
+        self.database = DatabaseRequires(
+            self,
+            relation_name="database",
+            database_name="database",
+            relations_aliases = ["cluster1", "cluster2"],
+        )
+        self.framework.observe(
+            self.database.on.cluster1_database_created, self._on_cluster1_database_created
+        )
+        self.framework.observe(
+            self.database.on.cluster2_database_created, self._on_cluster2_database_created
+        )
+
+    def _on_cluster1_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database on the cluster named cluster1
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+        ...
+
+    def _on_cluster2_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database on the cluster named cluster2
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+        ...
+
+```
+
+### Provider Charm
+
+Following an example of using the DatabaseRequestedEvent, in the context of the
+database charm code:
+
+```python
+from charms.data_platform_libs.v0.data_interfaces import DatabaseProvides
+
+class SampleCharm(CharmBase):
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        # Charm events defined in the database provides charm library.
+        self.provided_database = DatabaseProvides(self, relation_name="database")
+        self.framework.observe(self.provided_database.on.database_requested,
+            self._on_database_requested)
+        # Database generic helper
+        self.database = DatabaseHelper()
+
+    def _on_database_requested(self, event: DatabaseRequestedEvent) -> None:
+        # Handle the event triggered by a new database requested in the relation
+        # Retrieve the database name using the charm library.
+        db_name = event.database
+        # generate a new user credential
+        username = self.database.generate_user()
+        password = self.database.generate_password()
+        # set the credentials for the relation
+        self.provided_database.set_credentials(event.relation.id, username, password)
+        # set other variables for the relation event.set_tls("False")
+```
+As shown above, the library provides a custom event (database_requested) to handle
+the situation when an application charm requests a new database to be created.
+It's preferred to subscribe to this event instead of relation changed event to avoid
+creating a new database when other information other than a database name is
+exchanged in the relation databag.
+
+### Kafka
+
+This library is the interface to use and interact with the Kafka charm. This library contains
+custom events that add convenience to manage Kafka, and provides methods to consume the
+application related data.
+
+#### Requirer Charm
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    BootstrapServerChangedEvent,
+    KafkaRequires,
+    TopicCreatedEvent,
+)
+
+class ApplicationCharm(CharmBase):
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.kafka = KafkaRequires(self, "kafka_client", "test-topic")
+        self.framework.observe(
+            self.kafka.on.bootstrap_server_changed, self._on_kafka_bootstrap_server_changed
+        )
+        self.framework.observe(
+            self.kafka.on.topic_created, self._on_kafka_topic_created
+        )
+
+    def _on_kafka_bootstrap_server_changed(self, event: BootstrapServerChangedEvent):
+        # Event triggered when a bootstrap server was changed for this application
+
+        new_bootstrap_server = event.bootstrap_server
+        ...
+
+    def _on_kafka_topic_created(self, event: TopicCreatedEvent):
+        # Event triggered when a topic was created for this application
+        username = event.username
+        password = event.password
+        tls = event.tls
+        tls_ca= event.tls_ca
+        bootstrap_server event.bootstrap_server
+        consumer_group_prefic = event.consumer_group_prefix
+        zookeeper_uris = event.zookeeper_uris
+        ...
+
+```
+
+As shown above, the library provides some custom events to handle specific situations,
+which are listed below:
+
+- topic_created: event emitted when the requested topic is created.
+- bootstrap_server_changed: event emitted when the bootstrap server have changed.
+- credential_changed: event emitted when the credentials of Kafka changed.
+
+### Provider Charm
+
+Following the previous example, this is an example of the provider charm.
+
+```python
+class SampleCharm(CharmBase):
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    KafkaProvides,
+    TopicRequestedEvent,
+)
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Default charm events.
+        self.framework.observe(self.on.start, self._on_start)
+
+        # Charm events defined in the Kafka Provides charm library.
+        self.kafka_provider = KafkaProvides(self, relation_name="kafka_client")
+        self.framework.observe(self.kafka_provider.on.topic_requested, self._on_topic_requested)
+        # Kafka generic helper
+        self.kafka = KafkaHelper()
+
+    def _on_topic_requested(self, event: TopicRequestedEvent):
+        # Handle the on_topic_requested event.
+
+        topic = event.topic
+        relation_id = event.relation.id
+        # set connection info in the databag relation
+        self.kafka_provider.set_bootstrap_server(relation_id, self.kafka.get_bootstrap_server())
+        self.kafka_provider.set_credentials(relation_id, username=username, password=password)
+        self.kafka_provider.set_consumer_group_prefix(relation_id, ...)
+        self.kafka_provider.set_tls(relation_id, "False")
+        self.kafka_provider.set_zookeeper_uris(relation_id, ...)
+
+```
+As shown above, the library provides a custom event (topic_requested) to handle
+the situation when an application charm requests a new topic to be created.
+It is preferred to subscribe to this event instead of relation changed event to avoid
+creating a new topic when other information other than a topic name is
+exchanged in the relation databag.
+"""
+
+import json
+import logging
+from abc import ABC, abstractmethod
+from collections import namedtuple
+from datetime import datetime
+from typing import List, Optional
+
+from ops.charm import (
+    CharmBase,
+    CharmEvents,
+    RelationChangedEvent,
+    RelationEvent,
+    RelationJoinedEvent,
+)
+from ops.framework import EventSource, Object
+from ops.model import Relation
+
+# The unique Charmhub library identifier, never change it
+LIBID = "6c3e6b6680d64e9c89e611d1a15f65be"
+
+# Increment this major API version when introducing breaking changes
+LIBAPI = 0
+
+# Increment this PATCH version before using `charmcraft publish-lib` or reset
+# to 0 if you are raising the major API version
+LIBPATCH = 7
+
+PYDEPS = ["ops>=2.0.0"]
+
+logger = logging.getLogger(__name__)
+
+Diff = namedtuple("Diff", "added changed deleted")
+Diff.__doc__ = """
+A tuple for storing the diff between two data mappings.
+
+added - keys that were added
+changed - keys that still exist but have new values
+deleted - key that were deleted"""
+
+
+def diff(event: RelationChangedEvent, bucket: str) -> Diff:
+    """Retrieves the diff of the data in the relation changed databag.
+
+    Args:
+        event: relation changed event.
+        bucket: bucket of the databag (app or unit)
+
+    Returns:
+        a Diff instance containing the added, deleted and changed
+            keys from the event relation databag.
+    """
+    # Retrieve the old data from the data key in the application relation databag.
+    old_data = json.loads(event.relation.data[bucket].get("data", "{}"))
+    # Retrieve the new data from the event relation databag.
+    new_data = {
+        key: value for key, value in event.relation.data[event.app].items() if key != "data"
+    }
+
+    # These are the keys that were added to the databag and triggered this event.
+    added = new_data.keys() - old_data.keys()
+    # These are the keys that were removed from the databag and triggered this event.
+    deleted = old_data.keys() - new_data.keys()
+    # These are the keys that already existed in the databag,
+    # but had their values changed.
+    changed = {key for key in old_data.keys() & new_data.keys() if old_data[key] != new_data[key]}
+    # Convert the new_data to a serializable format and save it for a next diff check.
+    event.relation.data[bucket].update({"data": json.dumps(new_data)})
+
+    # Return the diff with all possible changes.
+    return Diff(added, changed, deleted)
+
+
+# Base DataProvides and DataRequires
+
+
+class DataProvides(Object, ABC):
+    """Base provides-side of the data products relation."""
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+        self.charm = charm
+        self.local_app = self.charm.model.app
+        self.local_unit = self.charm.unit
+        self.relation_name = relation_name
+        self.framework.observe(
+            charm.on[relation_name].relation_changed,
+            self._on_relation_changed,
+        )
+
+    def _diff(self, event: RelationChangedEvent) -> Diff:
+        """Retrieves the diff of the data in the relation changed databag.
+
+        Args:
+            event: relation changed event.
+
+        Returns:
+            a Diff instance containing the added, deleted and changed
+                keys from the event relation databag.
+        """
+        return diff(event, self.local_app)
+
+    @abstractmethod
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation data has changed."""
+        raise NotImplementedError
+
+    def fetch_relation_data(self) -> dict:
+        """Retrieves data from relation.
+
+        This function can be used to retrieve data from a relation
+        in the charm code when outside an event callback.
+
+        Returns:
+            a dict of the values stored in the relation data bag
+                for all relation instances (indexed by the relation id).
+        """
+        data = {}
+        for relation in self.relations:
+            data[relation.id] = {
+                key: value for key, value in relation.data[relation.app].items() if key != "data"
+            }
+        return data
+
+    def _update_relation_data(self, relation_id: int, data: dict) -> None:
+        """Updates a set of key-value pairs in the relation.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            data: dict containing the key-value pairs
+                that should be updated in the relation.
+        """
+        if self.local_unit.is_leader():
+            relation = self.charm.model.get_relation(self.relation_name, relation_id)
+            relation.data[self.local_app].update(data)
+
+    @property
+    def relations(self) -> List[Relation]:
+        """The list of Relation instances associated with this relation_name."""
+        return list(self.charm.model.relations[self.relation_name])
+
+    def set_credentials(self, relation_id: int, username: str, password: str) -> None:
+        """Set credentials.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            username: user that was created.
+            password: password of the created user.
+        """
+        self._update_relation_data(
+            relation_id,
+            {
+                "username": username,
+                "password": password,
+            },
+        )
+
+    def set_tls(self, relation_id: int, tls: str) -> None:
+        """Set whether TLS is enabled.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            tls: whether tls is enabled (True or False).
+        """
+        self._update_relation_data(relation_id, {"tls": tls})
+
+    def set_tls_ca(self, relation_id: int, tls_ca: str) -> None:
+        """Set the TLS CA in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            tls_ca: TLS certification authority.
+        """
+        self._update_relation_data(relation_id, {"tls_ca": tls_ca})
+
+
+class DataRequires(Object, ABC):
+    """Requires-side of the relation."""
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        extra_user_roles: str = None,
+    ):
+        """Manager of base client relations."""
+        super().__init__(charm, relation_name)
+        self.charm = charm
+        self.extra_user_roles = extra_user_roles
+        self.local_app = self.charm.model.app
+        self.local_unit = self.charm.unit
+        self.relation_name = relation_name
+        self.framework.observe(
+            self.charm.on[relation_name].relation_joined, self._on_relation_joined_event
+        )
+        self.framework.observe(
+            self.charm.on[relation_name].relation_changed, self._on_relation_changed_event
+        )
+
+    @abstractmethod
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the relation."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        raise NotImplementedError
+
+    def fetch_relation_data(self) -> dict:
+        """Retrieves data from relation.
+
+        This function can be used to retrieve data from a relation
+        in the charm code when outside an event callback.
+        Function cannot be used in `*-relation-broken` events and will raise an exception.
+
+        Returns:
+            a dict of the values stored in the relation data bag
+                for all relation instances (indexed by the relation ID).
+        """
+        data = {}
+        for relation in self.relations:
+            data[relation.id] = {
+                key: value for key, value in relation.data[relation.app].items() if key != "data"
+            }
+        return data
+
+    def _update_relation_data(self, relation_id: int, data: dict) -> None:
+        """Updates a set of key-value pairs in the relation.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            data: dict containing the key-value pairs
+                that should be updated in the relation.
+        """
+        if self.local_unit.is_leader():
+            relation = self.charm.model.get_relation(self.relation_name, relation_id)
+            relation.data[self.local_app].update(data)
+
+    def _diff(self, event: RelationChangedEvent) -> Diff:
+        """Retrieves the diff of the data in the relation changed databag.
+
+        Args:
+            event: relation changed event.
+
+        Returns:
+            a Diff instance containing the added, deleted and changed
+                keys from the event relation databag.
+        """
+        return diff(event, self.local_unit)
+
+    @property
+    def relations(self) -> List[Relation]:
+        """The list of Relation instances associated with this relation_name."""
+        return [
+            relation
+            for relation in self.charm.model.relations[self.relation_name]
+            if self._is_relation_active(relation)
+        ]
+
+    @staticmethod
+    def _is_relation_active(relation: Relation):
+        try:
+            _ = repr(relation.data)
+            return True
+        except RuntimeError:
+            return False
+
+    @staticmethod
+    def _is_resource_created_for_relation(relation: Relation):
+        return (
+            "username" in relation.data[relation.app] and "password" in relation.data[relation.app]
+        )
+
+    def is_resource_created(self, relation_id: Optional[int] = None) -> bool:
+        """Check if the resource has been created.
+
+        This function can be used to check if the Provider answered with data in the charm code
+        when outside an event callback.
+
+        Args:
+            relation_id (int, optional): When provided the check is done only for the relation id
+                provided, otherwise the check is done for all relations
+
+        Returns:
+            True or False
+
+        Raises:
+            IndexError: If relation_id is provided but that relation does not exist
+        """
+        if relation_id is not None:
+            try:
+                relation = [relation for relation in self.relations if relation.id == relation_id][
+                    0
+                ]
+                return self._is_resource_created_for_relation(relation)
+            except IndexError:
+                raise IndexError(f"relation id {relation_id} cannot be accessed")
+        else:
+            return (
+                all(
+                    [
+                        self._is_resource_created_for_relation(relation)
+                        for relation in self.relations
+                    ]
+                )
+                if self.relations
+                else False
+            )
+
+
+# General events
+
+
+class ExtraRoleEvent(RelationEvent):
+    """Base class for data events."""
+
+    @property
+    def extra_user_roles(self) -> Optional[str]:
+        """Returns the extra user roles that were requested."""
+        return self.relation.data[self.relation.app].get("extra-user-roles")
+
+
+class AuthenticationEvent(RelationEvent):
+    """Base class for authentication fields for events."""
+
+    @property
+    def username(self) -> Optional[str]:
+        """Returns the created username."""
+        return self.relation.data[self.relation.app].get("username")
+
+    @property
+    def password(self) -> Optional[str]:
+        """Returns the password for the created user."""
+        return self.relation.data[self.relation.app].get("password")
+
+    @property
+    def tls(self) -> Optional[str]:
+        """Returns whether TLS is configured."""
+        return self.relation.data[self.relation.app].get("tls")
+
+    @property
+    def tls_ca(self) -> Optional[str]:
+        """Returns TLS CA."""
+        return self.relation.data[self.relation.app].get("tls-ca")
+
+
+# Database related events and fields
+
+
+class DatabaseProvidesEvent(RelationEvent):
+    """Base class for database events."""
+
+    @property
+    def database(self) -> Optional[str]:
+        """Returns the database that was requested."""
+        return self.relation.data[self.relation.app].get("database")
+
+
+class DatabaseRequestedEvent(DatabaseProvidesEvent, ExtraRoleEvent):
+    """Event emitted when a new database is requested for use on this relation."""
+
+
+class DatabaseProvidesEvents(CharmEvents):
+    """Database events.
+
+    This class defines the events that the database can emit.
+    """
+
+    database_requested = EventSource(DatabaseRequestedEvent)
+
+
+class DatabaseRequiresEvent(RelationEvent):
+    """Base class for database events."""
+
+    @property
+    def endpoints(self) -> Optional[str]:
+        """Returns a comma separated list of read/write endpoints."""
+        return self.relation.data[self.relation.app].get("endpoints")
+
+    @property
+    def read_only_endpoints(self) -> Optional[str]:
+        """Returns a comma separated list of read only endpoints."""
+        return self.relation.data[self.relation.app].get("read-only-endpoints")
+
+    @property
+    def replset(self) -> Optional[str]:
+        """Returns the replicaset name.
+
+        MongoDB only.
+        """
+        return self.relation.data[self.relation.app].get("replset")
+
+    @property
+    def uris(self) -> Optional[str]:
+        """Returns the connection URIs.
+
+        MongoDB, Redis, OpenSearch.
+        """
+        return self.relation.data[self.relation.app].get("uris")
+
+    @property
+    def version(self) -> Optional[str]:
+        """Returns the version of the database.
+
+        Version as informed by the database daemon.
+        """
+        return self.relation.data[self.relation.app].get("version")
+
+
+class DatabaseCreatedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when a new database is created for use on this relation."""
+
+
+class DatabaseEndpointsChangedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when the read/write endpoints are changed."""
+
+
+class DatabaseReadOnlyEndpointsChangedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when the read only endpoints are changed."""
+
+
+class DatabaseRequiresEvents(CharmEvents):
+    """Database events.
+
+    This class defines the events that the database can emit.
+    """
+
+    database_created = EventSource(DatabaseCreatedEvent)
+    endpoints_changed = EventSource(DatabaseEndpointsChangedEvent)
+    read_only_endpoints_changed = EventSource(DatabaseReadOnlyEndpointsChangedEvent)
+
+
+# Database Provider and Requires
+
+
+class DatabaseProvides(DataProvides):
+    """Provider-side of the database relations."""
+
+    on = DatabaseProvidesEvents()
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Only the leader should handle this event.
+        if not self.local_unit.is_leader():
+            return
+
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Emit a database requested event if the setup key (database name and optional
+        # extra user roles) was added to the relation databag by the application.
+        if "database" in diff.added:
+            self.on.database_requested.emit(event.relation, app=event.app, unit=event.unit)
+
+    def set_endpoints(self, relation_id: int, connection_strings: str) -> None:
+        """Set database primary connections.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            connection_strings: database hosts and ports comma separated list.
+        """
+        self._update_relation_data(relation_id, {"endpoints": connection_strings})
+
+    def set_read_only_endpoints(self, relation_id: int, connection_strings: str) -> None:
+        """Set database replicas connection strings.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            connection_strings: database hosts and ports comma separated list.
+        """
+        self._update_relation_data(relation_id, {"read-only-endpoints": connection_strings})
+
+    def set_replset(self, relation_id: int, replset: str) -> None:
+        """Set replica set name in the application relation databag.
+
+        MongoDB only.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            replset: replica set name.
+        """
+        self._update_relation_data(relation_id, {"replset": replset})
+
+    def set_uris(self, relation_id: int, uris: str) -> None:
+        """Set the database connection URIs in the application relation databag.
+
+        MongoDB, Redis, and OpenSearch only.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            uris: connection URIs.
+        """
+        self._update_relation_data(relation_id, {"uris": uris})
+
+    def set_version(self, relation_id: int, version: str) -> None:
+        """Set the database version in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            version: database version.
+        """
+        self._update_relation_data(relation_id, {"version": version})
+
+
+class DatabaseRequires(DataRequires):
+    """Requires-side of the database relation."""
+
+    on = DatabaseRequiresEvents()
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        database_name: str,
+        extra_user_roles: str = None,
+        relations_aliases: List[str] = None,
+    ):
+        """Manager of database client relations."""
+        super().__init__(charm, relation_name, extra_user_roles)
+        self.database = database_name
+        self.relations_aliases = relations_aliases
+
+        # Define custom event names for each alias.
+        if relations_aliases:
+            # Ensure the number of aliases does not exceed the maximum
+            # of connections allowed in the specific relation.
+            relation_connection_limit = self.charm.meta.requires[relation_name].limit
+            if len(relations_aliases) != relation_connection_limit:
+                raise ValueError(
+                    f"The number of aliases must match the maximum number of connections allowed in the relation. "
+                    f"Expected {relation_connection_limit}, got {len(relations_aliases)}"
+                )
+
+            for relation_alias in relations_aliases:
+                self.on.define_event(f"{relation_alias}_database_created", DatabaseCreatedEvent)
+                self.on.define_event(
+                    f"{relation_alias}_endpoints_changed", DatabaseEndpointsChangedEvent
+                )
+                self.on.define_event(
+                    f"{relation_alias}_read_only_endpoints_changed",
+                    DatabaseReadOnlyEndpointsChangedEvent,
+                )
+
+    def _assign_relation_alias(self, relation_id: int) -> None:
+        """Assigns an alias to a relation.
+
+        This function writes in the unit data bag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+        """
+        # If no aliases were provided, return immediately.
+        if not self.relations_aliases:
+            return
+
+        # Return if an alias was already assigned to this relation
+        # (like when there are more than one unit joining the relation).
+        if (
+            self.charm.model.get_relation(self.relation_name, relation_id)
+            .data[self.local_unit]
+            .get("alias")
+        ):
+            return
+
+        # Retrieve the available aliases (the ones that weren't assigned to any relation).
+        available_aliases = self.relations_aliases[:]
+        for relation in self.charm.model.relations[self.relation_name]:
+            alias = relation.data[self.local_unit].get("alias")
+            if alias:
+                logger.debug("Alias %s was already assigned to relation %d", alias, relation.id)
+                available_aliases.remove(alias)
+
+        # Set the alias in the unit relation databag of the specific relation.
+        relation = self.charm.model.get_relation(self.relation_name, relation_id)
+        relation.data[self.local_unit].update({"alias": available_aliases[0]})
+
+    def _emit_aliased_event(self, event: RelationChangedEvent, event_name: str) -> None:
+        """Emit an aliased event to a particular relation if it has an alias.
+
+        Args:
+            event: the relation changed event that was received.
+            event_name: the name of the event to emit.
+        """
+        alias = self._get_relation_alias(event.relation.id)
+        if alias:
+            getattr(self.on, f"{alias}_{event_name}").emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+    def _get_relation_alias(self, relation_id: int) -> Optional[str]:
+        """Returns the relation alias.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+
+        Returns:
+            the relation alias or None if the relation was not found.
+        """
+        for relation in self.charm.model.relations[self.relation_name]:
+            if relation.id == relation_id:
+                return relation.data[self.local_unit].get("alias")
+        return None
+
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the database relation."""
+        # If relations aliases were provided, assign one to the relation.
+        self._assign_relation_alias(event.relation.id)
+
+        # Sets both database and extra user roles in the relation
+        # if the roles are provided. Otherwise, sets only the database.
+        if self.extra_user_roles:
+            self._update_relation_data(
+                event.relation.id,
+                {
+                    "database": self.database,
+                    "extra-user-roles": self.extra_user_roles,
+                },
+            )
+        else:
+            self._update_relation_data(event.relation.id, {"database": self.database})
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the database relation has changed."""
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Check if the database is created
+        # (the database charm shared the credentials).
+        if "username" in diff.added and "password" in diff.added:
+            # Emit the default event (the one without an alias).
+            logger.info("database created at %s", datetime.now())
+            self.on.database_created.emit(event.relation, app=event.app, unit=event.unit)
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "database_created")
+
+            # To avoid unnecessary application restarts do not trigger
+            # “endpoints_changed“ event if “database_created“ is triggered.
+            return
+
+        # Emit an endpoints changed event if the database
+        # added or changed this info in the relation databag.
+        if "endpoints" in diff.added or "endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("endpoints changed on %s", datetime.now())
+            self.on.endpoints_changed.emit(event.relation, app=event.app, unit=event.unit)
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "endpoints_changed")
+
+            # To avoid unnecessary application restarts do not trigger
+            # “read_only_endpoints_changed“ event if “endpoints_changed“ is triggered.
+            return
+
+        # Emit a read only endpoints changed event if the database
+        # added or changed this info in the relation databag.
+        if "read-only-endpoints" in diff.added or "read-only-endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("read-only-endpoints changed on %s", datetime.now())
+            self.on.read_only_endpoints_changed.emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "read_only_endpoints_changed")
+
+
+# Kafka related events
+
+
+class KafkaProvidesEvent(RelationEvent):
+    """Base class for Kafka events."""
+
+    @property
+    def topic(self) -> Optional[str]:
+        """Returns the topic that was requested."""
+        return self.relation.data[self.relation.app].get("topic")
+
+
+class TopicRequestedEvent(KafkaProvidesEvent, ExtraRoleEvent):
+    """Event emitted when a new topic is requested for use on this relation."""
+
+
+class KafkaProvidesEvents(CharmEvents):
+    """Kafka events.
+
+    This class defines the events that the Kafka can emit.
+    """
+
+    topic_requested = EventSource(TopicRequestedEvent)
+
+
+class KafkaRequiresEvent(RelationEvent):
+    """Base class for Kafka events."""
+
+    @property
+    def bootstrap_server(self) -> Optional[str]:
+        """Returns a a comma-seperated list of broker uris."""
+        return self.relation.data[self.relation.app].get("endpoints")
+
+    @property
+    def consumer_group_prefix(self) -> Optional[str]:
+        """Returns the consumer-group-prefix."""
+        return self.relation.data[self.relation.app].get("consumer-group-prefix")
+
+    @property
+    def zookeeper_uris(self) -> Optional[str]:
+        """Returns a comma separated list of Zookeeper uris."""
+        return self.relation.data[self.relation.app].get("zookeeper-uris")
+
+
+class TopicCreatedEvent(AuthenticationEvent, KafkaRequiresEvent):
+    """Event emitted when a new topic is created for use on this relation."""
+
+
+class BootstrapServerChangedEvent(AuthenticationEvent, KafkaRequiresEvent):
+    """Event emitted when the bootstrap server is changed."""
+
+
+class KafkaRequiresEvents(CharmEvents):
+    """Kafka events.
+
+    This class defines the events that the Kafka can emit.
+    """
+
+    topic_created = EventSource(TopicCreatedEvent)
+    bootstrap_server_changed = EventSource(BootstrapServerChangedEvent)
+
+
+# Kafka Provides and Requires
+
+
+class KafkaProvides(DataProvides):
+    """Provider-side of the Kafka relation."""
+
+    on = KafkaProvidesEvents()
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Only the leader should handle this event.
+        if not self.local_unit.is_leader():
+            return
+
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Emit a topic requested event if the setup key (topic name and optional
+        # extra user roles) was added to the relation databag by the application.
+        if "topic" in diff.added:
+            self.on.topic_requested.emit(event.relation, app=event.app, unit=event.unit)
+
+    def set_bootstrap_server(self, relation_id: int, bootstrap_server: str) -> None:
+        """Set the bootstrap server in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            bootstrap_server: the bootstrap server address.
+        """
+        self._update_relation_data(relation_id, {"endpoints": bootstrap_server})
+
+    def set_consumer_group_prefix(self, relation_id: int, consumer_group_prefix: str) -> None:
+        """Set the consumer group prefix in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            consumer_group_prefix: the consumer group prefix string.
+        """
+        self._update_relation_data(relation_id, {"consumer-group-prefix": consumer_group_prefix})
+
+    def set_zookeeper_uris(self, relation_id: int, zookeeper_uris: str) -> None:
+        """Set the zookeeper uris in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            zookeeper_uris: comma-seperated list of ZooKeeper server uris.
+        """
+        self._update_relation_data(relation_id, {"zookeeper-uris": zookeeper_uris})
+
+
+class KafkaRequires(DataRequires):
+    """Requires-side of the Kafka relation."""
+
+    on = KafkaRequiresEvents()
+
+    def __init__(self, charm, relation_name: str, topic: str, extra_user_roles: str = None):
+        """Manager of Kafka client relations."""
+        # super().__init__(charm, relation_name)
+        super().__init__(charm, relation_name, extra_user_roles)
+        self.charm = charm
+        self.topic = topic
+
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the Kafka relation."""
+        # Sets both topic and extra user roles in the relation
+        # if the roles are provided. Otherwise, sets only the topic.
+        self._update_relation_data(
+            event.relation.id,
+            {
+                "topic": self.topic,
+                "extra-user-roles": self.extra_user_roles,
+            }
+            if self.extra_user_roles is not None
+            else {"topic": self.topic},
+        )
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the Kafka relation has changed."""
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Check if the topic is created
+        # (the Kafka charm shared the credentials).
+        if "username" in diff.added and "password" in diff.added:
+            # Emit the default event (the one without an alias).
+            logger.info("topic created at %s", datetime.now())
+            self.on.topic_created.emit(event.relation, app=event.app, unit=event.unit)
+
+            # To avoid unnecessary application restarts do not trigger
+            # “endpoints_changed“ event if “topic_created“ is triggered.
+            return
+
+        # Emit an endpoints (bootstap-server) changed event if the Kakfa endpoints
+        # added or changed this info in the relation databag.
+        if "endpoints" in diff.added or "endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("endpoints changed on %s", datetime.now())
+            self.on.bootstrap_server_changed.emit(
+                event.relation, app=event.app, unit=event.unit
+            )  # here check if this is the right design
+            return
diff --git a/installers/charm/osm-lcm/metadata.yaml b/installers/charm/osm-lcm/metadata.yaml
index bd54541..e38e2b5 100644
--- a/installers/charm/osm-lcm/metadata.yaml
+++ b/installers/charm/osm-lcm/metadata.yaml
@@ -57,7 +57,7 @@
     interface: kafka
     limit: 1
   mongodb:
-    interface: mongodb
+    interface: mongodb_client
     limit: 1
   ro:
     interface: ro
diff --git a/installers/charm/osm-lcm/src/charm.py b/installers/charm/osm-lcm/src/charm.py
index 4a362a6..2ea9086 100755
--- a/installers/charm/osm-lcm/src/charm.py
+++ b/installers/charm/osm-lcm/src/charm.py
@@ -30,6 +30,7 @@
 import logging
 from typing import Any, Dict
 
+from charms.data_platform_libs.v0.data_interfaces import DatabaseRequires
 from charms.kafka_k8s.v0.kafka import KafkaRequires, _KafkaAvailableEvent
 from charms.osm_libs.v0.utils import (
     CharmError,
@@ -45,8 +46,6 @@
 from ops.main import main
 from ops.model import ActiveStatus, Container
 
-from legacy_interfaces import MongoClient
-
 HOSTPATHS = [
     HostPath(
         config="lcm-hostpath",
@@ -84,7 +83,9 @@
         super().__init__(*args)
         self.vca = VcaRequires(self)
         self.kafka = KafkaRequires(self)
-        self.mongodb_client = MongoClient(self, "mongodb")
+        self.mongodb_client = DatabaseRequires(
+            self, "mongodb", database_name="osm", extra_user_roles="admin"
+        )
         self._observe_charm_events()
         self.ro = RoRequires(self)
         self.container: Container = self.unit.get_container(self.container_name)
@@ -176,7 +177,7 @@
             # Relation events
             self.on.kafka_available: self._on_config_changed,
             self.on["kafka"].relation_broken: self._on_required_relation_broken,
-            self.on["mongodb"].relation_changed: self._on_config_changed,
+            self.mongodb_client.on.database_created: self._on_config_changed,
             self.on["mongodb"].relation_broken: self._on_required_relation_broken,
             self.on["ro"].relation_changed: self._on_config_changed,
             self.on["ro"].relation_broken: self._on_required_relation_broken,
@@ -199,7 +200,7 @@
 
         if not self.kafka.host or not self.kafka.port:
             missing_relations.append("kafka")
-        if self.mongodb_client.is_missing_data_in_unit():
+        if not self._is_database_available():
             missing_relations.append("mongodb")
         if not self.ro.host or not self.ro.port:
             missing_relations.append("ro")
@@ -211,6 +212,12 @@
             logger.warning(error_msg)
             raise CharmError(error_msg)
 
+    def _is_database_available(self) -> bool:
+        try:
+            return self.mongodb_client.is_resource_created()
+        except KeyError:
+            return False
+
     def _configure_service(self, container: Container) -> None:
         """Add Pebble layer with the lcm service."""
         logger.debug(f"configuring {self.app.name} service")
@@ -232,13 +239,13 @@
             "OSMLCM_RO_TENANT": "osm",
             # Database configuration
             "OSMLCM_DATABASE_DRIVER": "mongo",
-            "OSMLCM_DATABASE_URI": self.mongodb_client.connection_string,
+            "OSMLCM_DATABASE_URI": self._get_mongodb_uri(),
             "OSMLCM_DATABASE_COMMONKEY": self.config["database-commonkey"],
             # Storage configuration
             "OSMLCM_STORAGE_DRIVER": "mongo",
             "OSMLCM_STORAGE_PATH": "/app/storage",
             "OSMLCM_STORAGE_COLLECTION": "files",
-            "OSMLCM_STORAGE_URI": self.mongodb_client.connection_string,
+            "OSMLCM_STORAGE_URI": self._get_mongodb_uri(),
             "OSMLCM_VCA_HELM_CA_CERTS": self.config["helm-ca-certs"],
             "OSMLCM_VCA_STABLEREPOURL": self.config["helm-stable-repo-url"],
         }
@@ -275,6 +282,9 @@
         }
         return layer_config
 
+    def _get_mongodb_uri(self):
+        return list(self.mongodb_client.fetch_relation_data().values())[0]["uris"]
+
 
 if __name__ == "__main__":  # pragma: no cover
     main(OsmLcmCharm)
diff --git a/installers/charm/osm-lcm/tests/integration/test_charm.py b/installers/charm/osm-lcm/tests/integration/test_charm.py
index 889e287..bb0f34e 100644
--- a/installers/charm/osm-lcm/tests/integration/test_charm.py
+++ b/installers/charm/osm-lcm/tests/integration/test_charm.py
@@ -37,7 +37,6 @@
 KAFKA_APP = "kafka"
 MONGO_DB_CHARM = "mongodb-k8s"
 MONGO_DB_APP = "mongodb"
-RO_CHARM = "osm-ro"
 RO_APP = "ro"
 ZOOKEEPER_CHARM = "zookeeper-k8s"
 ZOOKEEPER_APP = "zookeeper"
@@ -50,14 +49,18 @@
 async def test_lcm_is_deployed(ops_test: OpsTest):
     charm = await ops_test.build_charm(".")
     resources = {"lcm-image": METADATA["resources"]["lcm-image"]["upstream-source"]}
+    ro_charm = await ops_test.build_charm("../osm-ro/")
+    ro_resources = {"ro-image": "opensourcemano/ro"}
 
     await asyncio.gather(
         ops_test.model.deploy(
             charm, resources=resources, application_name=LCM_APP, series="focal"
         ),
-        ops_test.model.deploy(RO_CHARM, application_name=RO_APP, channel="beta"),
+        ops_test.model.deploy(
+            ro_charm, resources=ro_resources, application_name=RO_APP, series="focal"
+        ),
         ops_test.model.deploy(KAFKA_CHARM, application_name=KAFKA_APP, channel="stable"),
-        ops_test.model.deploy(MONGO_DB_CHARM, application_name=MONGO_DB_APP, channel="stable"),
+        ops_test.model.deploy(MONGO_DB_CHARM, application_name=MONGO_DB_APP, channel="edge"),
         ops_test.model.deploy(ZOOKEEPER_CHARM, application_name=ZOOKEEPER_APP, channel="stable"),
     )
 
diff --git a/installers/charm/osm-lcm/tests/unit/test_charm.py b/installers/charm/osm-lcm/tests/unit/test_charm.py
index 8233d32..41cfb00 100644
--- a/installers/charm/osm-lcm/tests/unit/test_charm.py
+++ b/installers/charm/osm-lcm/tests/unit/test_charm.py
@@ -36,6 +36,7 @@
 def harness(mocker: MockerFixture):
     harness = Harness(OsmLcmCharm)
     harness.begin()
+    harness.container_pebble_ready(container_name)
     yield harness
     harness.cleanup()
 
@@ -69,7 +70,9 @@
     relation_id = harness.add_relation("mongodb", "mongodb")
     harness.add_relation_unit(relation_id, "mongodb/0")
     harness.update_relation_data(
-        relation_id, "mongodb/0", {"connection_string": "mongodb://:1234"}
+        relation_id,
+        "mongodb",
+        {"uris": "mongodb://:1234", "username": "user", "password": "password"},
     )
     relation_ids.append(relation_id)
     # Add kafka relation
diff --git a/installers/charm/osm-lcm/tox.ini b/installers/charm/osm-lcm/tox.ini
index 71cf2a6..6d39e4b 100644
--- a/installers/charm/osm-lcm/tox.ini
+++ b/installers/charm/osm-lcm/tox.ini
@@ -85,7 +85,7 @@
 description = Run integration tests
 deps =
     pytest
-    juju
+    juju<3
     pytest-operator
     -r{toxinidir}/requirements.txt
 commands =
diff --git a/installers/charm/osm-mon/lib/charms/data_platform_libs/v0/data_interfaces.py b/installers/charm/osm-mon/lib/charms/data_platform_libs/v0/data_interfaces.py
new file mode 100644
index 0000000..b3da5aa
--- /dev/null
+++ b/installers/charm/osm-mon/lib/charms/data_platform_libs/v0/data_interfaces.py
@@ -0,0 +1,1130 @@
+# Copyright 2023 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.
+
+"""Library to manage the relation for the data-platform products.
+
+This library contains the Requires and Provides classes for handling the relation
+between an application and multiple managed application supported by the data-team:
+MySQL, Postgresql, MongoDB, Redis,  and Kakfa.
+
+### Database (MySQL, Postgresql, MongoDB, and Redis)
+
+#### Requires Charm
+This library is a uniform interface to a selection of common database
+metadata, with added custom events that add convenience to database management,
+and methods to consume the application related data.
+
+
+Following an example of using the DatabaseCreatedEvent, in the context of the
+application charm code:
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    DatabaseCreatedEvent,
+    DatabaseRequires,
+)
+
+class ApplicationCharm(CharmBase):
+    # Application charm that connects to database charms.
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Charm events defined in the database requires charm library.
+        self.database = DatabaseRequires(self, relation_name="database", database_name="database")
+        self.framework.observe(self.database.on.database_created, self._on_database_created)
+
+    def _on_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+
+        # Start application with rendered configuration
+        self._start_application(config_file)
+
+        # Set active status
+        self.unit.status = ActiveStatus("received database credentials")
+```
+
+As shown above, the library provides some custom events to handle specific situations,
+which are listed below:
+
+-  database_created: event emitted when the requested database is created.
+-  endpoints_changed: event emitted when the read/write endpoints of the database have changed.
+-  read_only_endpoints_changed: event emitted when the read-only endpoints of the database
+  have changed. Event is not triggered if read/write endpoints changed too.
+
+If it is needed to connect multiple database clusters to the same relation endpoint
+the application charm can implement the same code as if it would connect to only
+one database cluster (like the above code example).
+
+To differentiate multiple clusters connected to the same relation endpoint
+the application charm can use the name of the remote application:
+
+```python
+
+def _on_database_created(self, event: DatabaseCreatedEvent) -> None:
+    # Get the remote app name of the cluster that triggered this event
+    cluster = event.relation.app.name
+```
+
+It is also possible to provide an alias for each different database cluster/relation.
+
+So, it is possible to differentiate the clusters in two ways.
+The first is to use the remote application name, i.e., `event.relation.app.name`, as above.
+
+The second way is to use different event handlers to handle each cluster events.
+The implementation would be something like the following code:
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    DatabaseCreatedEvent,
+    DatabaseRequires,
+)
+
+class ApplicationCharm(CharmBase):
+    # Application charm that connects to database charms.
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Define the cluster aliases and one handler for each cluster database created event.
+        self.database = DatabaseRequires(
+            self,
+            relation_name="database",
+            database_name="database",
+            relations_aliases = ["cluster1", "cluster2"],
+        )
+        self.framework.observe(
+            self.database.on.cluster1_database_created, self._on_cluster1_database_created
+        )
+        self.framework.observe(
+            self.database.on.cluster2_database_created, self._on_cluster2_database_created
+        )
+
+    def _on_cluster1_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database on the cluster named cluster1
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+        ...
+
+    def _on_cluster2_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database on the cluster named cluster2
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+        ...
+
+```
+
+### Provider Charm
+
+Following an example of using the DatabaseRequestedEvent, in the context of the
+database charm code:
+
+```python
+from charms.data_platform_libs.v0.data_interfaces import DatabaseProvides
+
+class SampleCharm(CharmBase):
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        # Charm events defined in the database provides charm library.
+        self.provided_database = DatabaseProvides(self, relation_name="database")
+        self.framework.observe(self.provided_database.on.database_requested,
+            self._on_database_requested)
+        # Database generic helper
+        self.database = DatabaseHelper()
+
+    def _on_database_requested(self, event: DatabaseRequestedEvent) -> None:
+        # Handle the event triggered by a new database requested in the relation
+        # Retrieve the database name using the charm library.
+        db_name = event.database
+        # generate a new user credential
+        username = self.database.generate_user()
+        password = self.database.generate_password()
+        # set the credentials for the relation
+        self.provided_database.set_credentials(event.relation.id, username, password)
+        # set other variables for the relation event.set_tls("False")
+```
+As shown above, the library provides a custom event (database_requested) to handle
+the situation when an application charm requests a new database to be created.
+It's preferred to subscribe to this event instead of relation changed event to avoid
+creating a new database when other information other than a database name is
+exchanged in the relation databag.
+
+### Kafka
+
+This library is the interface to use and interact with the Kafka charm. This library contains
+custom events that add convenience to manage Kafka, and provides methods to consume the
+application related data.
+
+#### Requirer Charm
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    BootstrapServerChangedEvent,
+    KafkaRequires,
+    TopicCreatedEvent,
+)
+
+class ApplicationCharm(CharmBase):
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.kafka = KafkaRequires(self, "kafka_client", "test-topic")
+        self.framework.observe(
+            self.kafka.on.bootstrap_server_changed, self._on_kafka_bootstrap_server_changed
+        )
+        self.framework.observe(
+            self.kafka.on.topic_created, self._on_kafka_topic_created
+        )
+
+    def _on_kafka_bootstrap_server_changed(self, event: BootstrapServerChangedEvent):
+        # Event triggered when a bootstrap server was changed for this application
+
+        new_bootstrap_server = event.bootstrap_server
+        ...
+
+    def _on_kafka_topic_created(self, event: TopicCreatedEvent):
+        # Event triggered when a topic was created for this application
+        username = event.username
+        password = event.password
+        tls = event.tls
+        tls_ca= event.tls_ca
+        bootstrap_server event.bootstrap_server
+        consumer_group_prefic = event.consumer_group_prefix
+        zookeeper_uris = event.zookeeper_uris
+        ...
+
+```
+
+As shown above, the library provides some custom events to handle specific situations,
+which are listed below:
+
+- topic_created: event emitted when the requested topic is created.
+- bootstrap_server_changed: event emitted when the bootstrap server have changed.
+- credential_changed: event emitted when the credentials of Kafka changed.
+
+### Provider Charm
+
+Following the previous example, this is an example of the provider charm.
+
+```python
+class SampleCharm(CharmBase):
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    KafkaProvides,
+    TopicRequestedEvent,
+)
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Default charm events.
+        self.framework.observe(self.on.start, self._on_start)
+
+        # Charm events defined in the Kafka Provides charm library.
+        self.kafka_provider = KafkaProvides(self, relation_name="kafka_client")
+        self.framework.observe(self.kafka_provider.on.topic_requested, self._on_topic_requested)
+        # Kafka generic helper
+        self.kafka = KafkaHelper()
+
+    def _on_topic_requested(self, event: TopicRequestedEvent):
+        # Handle the on_topic_requested event.
+
+        topic = event.topic
+        relation_id = event.relation.id
+        # set connection info in the databag relation
+        self.kafka_provider.set_bootstrap_server(relation_id, self.kafka.get_bootstrap_server())
+        self.kafka_provider.set_credentials(relation_id, username=username, password=password)
+        self.kafka_provider.set_consumer_group_prefix(relation_id, ...)
+        self.kafka_provider.set_tls(relation_id, "False")
+        self.kafka_provider.set_zookeeper_uris(relation_id, ...)
+
+```
+As shown above, the library provides a custom event (topic_requested) to handle
+the situation when an application charm requests a new topic to be created.
+It is preferred to subscribe to this event instead of relation changed event to avoid
+creating a new topic when other information other than a topic name is
+exchanged in the relation databag.
+"""
+
+import json
+import logging
+from abc import ABC, abstractmethod
+from collections import namedtuple
+from datetime import datetime
+from typing import List, Optional
+
+from ops.charm import (
+    CharmBase,
+    CharmEvents,
+    RelationChangedEvent,
+    RelationEvent,
+    RelationJoinedEvent,
+)
+from ops.framework import EventSource, Object
+from ops.model import Relation
+
+# The unique Charmhub library identifier, never change it
+LIBID = "6c3e6b6680d64e9c89e611d1a15f65be"
+
+# Increment this major API version when introducing breaking changes
+LIBAPI = 0
+
+# Increment this PATCH version before using `charmcraft publish-lib` or reset
+# to 0 if you are raising the major API version
+LIBPATCH = 7
+
+PYDEPS = ["ops>=2.0.0"]
+
+logger = logging.getLogger(__name__)
+
+Diff = namedtuple("Diff", "added changed deleted")
+Diff.__doc__ = """
+A tuple for storing the diff between two data mappings.
+
+added - keys that were added
+changed - keys that still exist but have new values
+deleted - key that were deleted"""
+
+
+def diff(event: RelationChangedEvent, bucket: str) -> Diff:
+    """Retrieves the diff of the data in the relation changed databag.
+
+    Args:
+        event: relation changed event.
+        bucket: bucket of the databag (app or unit)
+
+    Returns:
+        a Diff instance containing the added, deleted and changed
+            keys from the event relation databag.
+    """
+    # Retrieve the old data from the data key in the application relation databag.
+    old_data = json.loads(event.relation.data[bucket].get("data", "{}"))
+    # Retrieve the new data from the event relation databag.
+    new_data = {
+        key: value for key, value in event.relation.data[event.app].items() if key != "data"
+    }
+
+    # These are the keys that were added to the databag and triggered this event.
+    added = new_data.keys() - old_data.keys()
+    # These are the keys that were removed from the databag and triggered this event.
+    deleted = old_data.keys() - new_data.keys()
+    # These are the keys that already existed in the databag,
+    # but had their values changed.
+    changed = {key for key in old_data.keys() & new_data.keys() if old_data[key] != new_data[key]}
+    # Convert the new_data to a serializable format and save it for a next diff check.
+    event.relation.data[bucket].update({"data": json.dumps(new_data)})
+
+    # Return the diff with all possible changes.
+    return Diff(added, changed, deleted)
+
+
+# Base DataProvides and DataRequires
+
+
+class DataProvides(Object, ABC):
+    """Base provides-side of the data products relation."""
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+        self.charm = charm
+        self.local_app = self.charm.model.app
+        self.local_unit = self.charm.unit
+        self.relation_name = relation_name
+        self.framework.observe(
+            charm.on[relation_name].relation_changed,
+            self._on_relation_changed,
+        )
+
+    def _diff(self, event: RelationChangedEvent) -> Diff:
+        """Retrieves the diff of the data in the relation changed databag.
+
+        Args:
+            event: relation changed event.
+
+        Returns:
+            a Diff instance containing the added, deleted and changed
+                keys from the event relation databag.
+        """
+        return diff(event, self.local_app)
+
+    @abstractmethod
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation data has changed."""
+        raise NotImplementedError
+
+    def fetch_relation_data(self) -> dict:
+        """Retrieves data from relation.
+
+        This function can be used to retrieve data from a relation
+        in the charm code when outside an event callback.
+
+        Returns:
+            a dict of the values stored in the relation data bag
+                for all relation instances (indexed by the relation id).
+        """
+        data = {}
+        for relation in self.relations:
+            data[relation.id] = {
+                key: value for key, value in relation.data[relation.app].items() if key != "data"
+            }
+        return data
+
+    def _update_relation_data(self, relation_id: int, data: dict) -> None:
+        """Updates a set of key-value pairs in the relation.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            data: dict containing the key-value pairs
+                that should be updated in the relation.
+        """
+        if self.local_unit.is_leader():
+            relation = self.charm.model.get_relation(self.relation_name, relation_id)
+            relation.data[self.local_app].update(data)
+
+    @property
+    def relations(self) -> List[Relation]:
+        """The list of Relation instances associated with this relation_name."""
+        return list(self.charm.model.relations[self.relation_name])
+
+    def set_credentials(self, relation_id: int, username: str, password: str) -> None:
+        """Set credentials.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            username: user that was created.
+            password: password of the created user.
+        """
+        self._update_relation_data(
+            relation_id,
+            {
+                "username": username,
+                "password": password,
+            },
+        )
+
+    def set_tls(self, relation_id: int, tls: str) -> None:
+        """Set whether TLS is enabled.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            tls: whether tls is enabled (True or False).
+        """
+        self._update_relation_data(relation_id, {"tls": tls})
+
+    def set_tls_ca(self, relation_id: int, tls_ca: str) -> None:
+        """Set the TLS CA in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            tls_ca: TLS certification authority.
+        """
+        self._update_relation_data(relation_id, {"tls_ca": tls_ca})
+
+
+class DataRequires(Object, ABC):
+    """Requires-side of the relation."""
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        extra_user_roles: str = None,
+    ):
+        """Manager of base client relations."""
+        super().__init__(charm, relation_name)
+        self.charm = charm
+        self.extra_user_roles = extra_user_roles
+        self.local_app = self.charm.model.app
+        self.local_unit = self.charm.unit
+        self.relation_name = relation_name
+        self.framework.observe(
+            self.charm.on[relation_name].relation_joined, self._on_relation_joined_event
+        )
+        self.framework.observe(
+            self.charm.on[relation_name].relation_changed, self._on_relation_changed_event
+        )
+
+    @abstractmethod
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the relation."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        raise NotImplementedError
+
+    def fetch_relation_data(self) -> dict:
+        """Retrieves data from relation.
+
+        This function can be used to retrieve data from a relation
+        in the charm code when outside an event callback.
+        Function cannot be used in `*-relation-broken` events and will raise an exception.
+
+        Returns:
+            a dict of the values stored in the relation data bag
+                for all relation instances (indexed by the relation ID).
+        """
+        data = {}
+        for relation in self.relations:
+            data[relation.id] = {
+                key: value for key, value in relation.data[relation.app].items() if key != "data"
+            }
+        return data
+
+    def _update_relation_data(self, relation_id: int, data: dict) -> None:
+        """Updates a set of key-value pairs in the relation.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            data: dict containing the key-value pairs
+                that should be updated in the relation.
+        """
+        if self.local_unit.is_leader():
+            relation = self.charm.model.get_relation(self.relation_name, relation_id)
+            relation.data[self.local_app].update(data)
+
+    def _diff(self, event: RelationChangedEvent) -> Diff:
+        """Retrieves the diff of the data in the relation changed databag.
+
+        Args:
+            event: relation changed event.
+
+        Returns:
+            a Diff instance containing the added, deleted and changed
+                keys from the event relation databag.
+        """
+        return diff(event, self.local_unit)
+
+    @property
+    def relations(self) -> List[Relation]:
+        """The list of Relation instances associated with this relation_name."""
+        return [
+            relation
+            for relation in self.charm.model.relations[self.relation_name]
+            if self._is_relation_active(relation)
+        ]
+
+    @staticmethod
+    def _is_relation_active(relation: Relation):
+        try:
+            _ = repr(relation.data)
+            return True
+        except RuntimeError:
+            return False
+
+    @staticmethod
+    def _is_resource_created_for_relation(relation: Relation):
+        return (
+            "username" in relation.data[relation.app] and "password" in relation.data[relation.app]
+        )
+
+    def is_resource_created(self, relation_id: Optional[int] = None) -> bool:
+        """Check if the resource has been created.
+
+        This function can be used to check if the Provider answered with data in the charm code
+        when outside an event callback.
+
+        Args:
+            relation_id (int, optional): When provided the check is done only for the relation id
+                provided, otherwise the check is done for all relations
+
+        Returns:
+            True or False
+
+        Raises:
+            IndexError: If relation_id is provided but that relation does not exist
+        """
+        if relation_id is not None:
+            try:
+                relation = [relation for relation in self.relations if relation.id == relation_id][
+                    0
+                ]
+                return self._is_resource_created_for_relation(relation)
+            except IndexError:
+                raise IndexError(f"relation id {relation_id} cannot be accessed")
+        else:
+            return (
+                all(
+                    [
+                        self._is_resource_created_for_relation(relation)
+                        for relation in self.relations
+                    ]
+                )
+                if self.relations
+                else False
+            )
+
+
+# General events
+
+
+class ExtraRoleEvent(RelationEvent):
+    """Base class for data events."""
+
+    @property
+    def extra_user_roles(self) -> Optional[str]:
+        """Returns the extra user roles that were requested."""
+        return self.relation.data[self.relation.app].get("extra-user-roles")
+
+
+class AuthenticationEvent(RelationEvent):
+    """Base class for authentication fields for events."""
+
+    @property
+    def username(self) -> Optional[str]:
+        """Returns the created username."""
+        return self.relation.data[self.relation.app].get("username")
+
+    @property
+    def password(self) -> Optional[str]:
+        """Returns the password for the created user."""
+        return self.relation.data[self.relation.app].get("password")
+
+    @property
+    def tls(self) -> Optional[str]:
+        """Returns whether TLS is configured."""
+        return self.relation.data[self.relation.app].get("tls")
+
+    @property
+    def tls_ca(self) -> Optional[str]:
+        """Returns TLS CA."""
+        return self.relation.data[self.relation.app].get("tls-ca")
+
+
+# Database related events and fields
+
+
+class DatabaseProvidesEvent(RelationEvent):
+    """Base class for database events."""
+
+    @property
+    def database(self) -> Optional[str]:
+        """Returns the database that was requested."""
+        return self.relation.data[self.relation.app].get("database")
+
+
+class DatabaseRequestedEvent(DatabaseProvidesEvent, ExtraRoleEvent):
+    """Event emitted when a new database is requested for use on this relation."""
+
+
+class DatabaseProvidesEvents(CharmEvents):
+    """Database events.
+
+    This class defines the events that the database can emit.
+    """
+
+    database_requested = EventSource(DatabaseRequestedEvent)
+
+
+class DatabaseRequiresEvent(RelationEvent):
+    """Base class for database events."""
+
+    @property
+    def endpoints(self) -> Optional[str]:
+        """Returns a comma separated list of read/write endpoints."""
+        return self.relation.data[self.relation.app].get("endpoints")
+
+    @property
+    def read_only_endpoints(self) -> Optional[str]:
+        """Returns a comma separated list of read only endpoints."""
+        return self.relation.data[self.relation.app].get("read-only-endpoints")
+
+    @property
+    def replset(self) -> Optional[str]:
+        """Returns the replicaset name.
+
+        MongoDB only.
+        """
+        return self.relation.data[self.relation.app].get("replset")
+
+    @property
+    def uris(self) -> Optional[str]:
+        """Returns the connection URIs.
+
+        MongoDB, Redis, OpenSearch.
+        """
+        return self.relation.data[self.relation.app].get("uris")
+
+    @property
+    def version(self) -> Optional[str]:
+        """Returns the version of the database.
+
+        Version as informed by the database daemon.
+        """
+        return self.relation.data[self.relation.app].get("version")
+
+
+class DatabaseCreatedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when a new database is created for use on this relation."""
+
+
+class DatabaseEndpointsChangedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when the read/write endpoints are changed."""
+
+
+class DatabaseReadOnlyEndpointsChangedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when the read only endpoints are changed."""
+
+
+class DatabaseRequiresEvents(CharmEvents):
+    """Database events.
+
+    This class defines the events that the database can emit.
+    """
+
+    database_created = EventSource(DatabaseCreatedEvent)
+    endpoints_changed = EventSource(DatabaseEndpointsChangedEvent)
+    read_only_endpoints_changed = EventSource(DatabaseReadOnlyEndpointsChangedEvent)
+
+
+# Database Provider and Requires
+
+
+class DatabaseProvides(DataProvides):
+    """Provider-side of the database relations."""
+
+    on = DatabaseProvidesEvents()
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Only the leader should handle this event.
+        if not self.local_unit.is_leader():
+            return
+
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Emit a database requested event if the setup key (database name and optional
+        # extra user roles) was added to the relation databag by the application.
+        if "database" in diff.added:
+            self.on.database_requested.emit(event.relation, app=event.app, unit=event.unit)
+
+    def set_endpoints(self, relation_id: int, connection_strings: str) -> None:
+        """Set database primary connections.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            connection_strings: database hosts and ports comma separated list.
+        """
+        self._update_relation_data(relation_id, {"endpoints": connection_strings})
+
+    def set_read_only_endpoints(self, relation_id: int, connection_strings: str) -> None:
+        """Set database replicas connection strings.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            connection_strings: database hosts and ports comma separated list.
+        """
+        self._update_relation_data(relation_id, {"read-only-endpoints": connection_strings})
+
+    def set_replset(self, relation_id: int, replset: str) -> None:
+        """Set replica set name in the application relation databag.
+
+        MongoDB only.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            replset: replica set name.
+        """
+        self._update_relation_data(relation_id, {"replset": replset})
+
+    def set_uris(self, relation_id: int, uris: str) -> None:
+        """Set the database connection URIs in the application relation databag.
+
+        MongoDB, Redis, and OpenSearch only.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            uris: connection URIs.
+        """
+        self._update_relation_data(relation_id, {"uris": uris})
+
+    def set_version(self, relation_id: int, version: str) -> None:
+        """Set the database version in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            version: database version.
+        """
+        self._update_relation_data(relation_id, {"version": version})
+
+
+class DatabaseRequires(DataRequires):
+    """Requires-side of the database relation."""
+
+    on = DatabaseRequiresEvents()
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        database_name: str,
+        extra_user_roles: str = None,
+        relations_aliases: List[str] = None,
+    ):
+        """Manager of database client relations."""
+        super().__init__(charm, relation_name, extra_user_roles)
+        self.database = database_name
+        self.relations_aliases = relations_aliases
+
+        # Define custom event names for each alias.
+        if relations_aliases:
+            # Ensure the number of aliases does not exceed the maximum
+            # of connections allowed in the specific relation.
+            relation_connection_limit = self.charm.meta.requires[relation_name].limit
+            if len(relations_aliases) != relation_connection_limit:
+                raise ValueError(
+                    f"The number of aliases must match the maximum number of connections allowed in the relation. "
+                    f"Expected {relation_connection_limit}, got {len(relations_aliases)}"
+                )
+
+            for relation_alias in relations_aliases:
+                self.on.define_event(f"{relation_alias}_database_created", DatabaseCreatedEvent)
+                self.on.define_event(
+                    f"{relation_alias}_endpoints_changed", DatabaseEndpointsChangedEvent
+                )
+                self.on.define_event(
+                    f"{relation_alias}_read_only_endpoints_changed",
+                    DatabaseReadOnlyEndpointsChangedEvent,
+                )
+
+    def _assign_relation_alias(self, relation_id: int) -> None:
+        """Assigns an alias to a relation.
+
+        This function writes in the unit data bag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+        """
+        # If no aliases were provided, return immediately.
+        if not self.relations_aliases:
+            return
+
+        # Return if an alias was already assigned to this relation
+        # (like when there are more than one unit joining the relation).
+        if (
+            self.charm.model.get_relation(self.relation_name, relation_id)
+            .data[self.local_unit]
+            .get("alias")
+        ):
+            return
+
+        # Retrieve the available aliases (the ones that weren't assigned to any relation).
+        available_aliases = self.relations_aliases[:]
+        for relation in self.charm.model.relations[self.relation_name]:
+            alias = relation.data[self.local_unit].get("alias")
+            if alias:
+                logger.debug("Alias %s was already assigned to relation %d", alias, relation.id)
+                available_aliases.remove(alias)
+
+        # Set the alias in the unit relation databag of the specific relation.
+        relation = self.charm.model.get_relation(self.relation_name, relation_id)
+        relation.data[self.local_unit].update({"alias": available_aliases[0]})
+
+    def _emit_aliased_event(self, event: RelationChangedEvent, event_name: str) -> None:
+        """Emit an aliased event to a particular relation if it has an alias.
+
+        Args:
+            event: the relation changed event that was received.
+            event_name: the name of the event to emit.
+        """
+        alias = self._get_relation_alias(event.relation.id)
+        if alias:
+            getattr(self.on, f"{alias}_{event_name}").emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+    def _get_relation_alias(self, relation_id: int) -> Optional[str]:
+        """Returns the relation alias.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+
+        Returns:
+            the relation alias or None if the relation was not found.
+        """
+        for relation in self.charm.model.relations[self.relation_name]:
+            if relation.id == relation_id:
+                return relation.data[self.local_unit].get("alias")
+        return None
+
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the database relation."""
+        # If relations aliases were provided, assign one to the relation.
+        self._assign_relation_alias(event.relation.id)
+
+        # Sets both database and extra user roles in the relation
+        # if the roles are provided. Otherwise, sets only the database.
+        if self.extra_user_roles:
+            self._update_relation_data(
+                event.relation.id,
+                {
+                    "database": self.database,
+                    "extra-user-roles": self.extra_user_roles,
+                },
+            )
+        else:
+            self._update_relation_data(event.relation.id, {"database": self.database})
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the database relation has changed."""
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Check if the database is created
+        # (the database charm shared the credentials).
+        if "username" in diff.added and "password" in diff.added:
+            # Emit the default event (the one without an alias).
+            logger.info("database created at %s", datetime.now())
+            self.on.database_created.emit(event.relation, app=event.app, unit=event.unit)
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "database_created")
+
+            # To avoid unnecessary application restarts do not trigger
+            # “endpoints_changed“ event if “database_created“ is triggered.
+            return
+
+        # Emit an endpoints changed event if the database
+        # added or changed this info in the relation databag.
+        if "endpoints" in diff.added or "endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("endpoints changed on %s", datetime.now())
+            self.on.endpoints_changed.emit(event.relation, app=event.app, unit=event.unit)
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "endpoints_changed")
+
+            # To avoid unnecessary application restarts do not trigger
+            # “read_only_endpoints_changed“ event if “endpoints_changed“ is triggered.
+            return
+
+        # Emit a read only endpoints changed event if the database
+        # added or changed this info in the relation databag.
+        if "read-only-endpoints" in diff.added or "read-only-endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("read-only-endpoints changed on %s", datetime.now())
+            self.on.read_only_endpoints_changed.emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "read_only_endpoints_changed")
+
+
+# Kafka related events
+
+
+class KafkaProvidesEvent(RelationEvent):
+    """Base class for Kafka events."""
+
+    @property
+    def topic(self) -> Optional[str]:
+        """Returns the topic that was requested."""
+        return self.relation.data[self.relation.app].get("topic")
+
+
+class TopicRequestedEvent(KafkaProvidesEvent, ExtraRoleEvent):
+    """Event emitted when a new topic is requested for use on this relation."""
+
+
+class KafkaProvidesEvents(CharmEvents):
+    """Kafka events.
+
+    This class defines the events that the Kafka can emit.
+    """
+
+    topic_requested = EventSource(TopicRequestedEvent)
+
+
+class KafkaRequiresEvent(RelationEvent):
+    """Base class for Kafka events."""
+
+    @property
+    def bootstrap_server(self) -> Optional[str]:
+        """Returns a a comma-seperated list of broker uris."""
+        return self.relation.data[self.relation.app].get("endpoints")
+
+    @property
+    def consumer_group_prefix(self) -> Optional[str]:
+        """Returns the consumer-group-prefix."""
+        return self.relation.data[self.relation.app].get("consumer-group-prefix")
+
+    @property
+    def zookeeper_uris(self) -> Optional[str]:
+        """Returns a comma separated list of Zookeeper uris."""
+        return self.relation.data[self.relation.app].get("zookeeper-uris")
+
+
+class TopicCreatedEvent(AuthenticationEvent, KafkaRequiresEvent):
+    """Event emitted when a new topic is created for use on this relation."""
+
+
+class BootstrapServerChangedEvent(AuthenticationEvent, KafkaRequiresEvent):
+    """Event emitted when the bootstrap server is changed."""
+
+
+class KafkaRequiresEvents(CharmEvents):
+    """Kafka events.
+
+    This class defines the events that the Kafka can emit.
+    """
+
+    topic_created = EventSource(TopicCreatedEvent)
+    bootstrap_server_changed = EventSource(BootstrapServerChangedEvent)
+
+
+# Kafka Provides and Requires
+
+
+class KafkaProvides(DataProvides):
+    """Provider-side of the Kafka relation."""
+
+    on = KafkaProvidesEvents()
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Only the leader should handle this event.
+        if not self.local_unit.is_leader():
+            return
+
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Emit a topic requested event if the setup key (topic name and optional
+        # extra user roles) was added to the relation databag by the application.
+        if "topic" in diff.added:
+            self.on.topic_requested.emit(event.relation, app=event.app, unit=event.unit)
+
+    def set_bootstrap_server(self, relation_id: int, bootstrap_server: str) -> None:
+        """Set the bootstrap server in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            bootstrap_server: the bootstrap server address.
+        """
+        self._update_relation_data(relation_id, {"endpoints": bootstrap_server})
+
+    def set_consumer_group_prefix(self, relation_id: int, consumer_group_prefix: str) -> None:
+        """Set the consumer group prefix in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            consumer_group_prefix: the consumer group prefix string.
+        """
+        self._update_relation_data(relation_id, {"consumer-group-prefix": consumer_group_prefix})
+
+    def set_zookeeper_uris(self, relation_id: int, zookeeper_uris: str) -> None:
+        """Set the zookeeper uris in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            zookeeper_uris: comma-seperated list of ZooKeeper server uris.
+        """
+        self._update_relation_data(relation_id, {"zookeeper-uris": zookeeper_uris})
+
+
+class KafkaRequires(DataRequires):
+    """Requires-side of the Kafka relation."""
+
+    on = KafkaRequiresEvents()
+
+    def __init__(self, charm, relation_name: str, topic: str, extra_user_roles: str = None):
+        """Manager of Kafka client relations."""
+        # super().__init__(charm, relation_name)
+        super().__init__(charm, relation_name, extra_user_roles)
+        self.charm = charm
+        self.topic = topic
+
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the Kafka relation."""
+        # Sets both topic and extra user roles in the relation
+        # if the roles are provided. Otherwise, sets only the topic.
+        self._update_relation_data(
+            event.relation.id,
+            {
+                "topic": self.topic,
+                "extra-user-roles": self.extra_user_roles,
+            }
+            if self.extra_user_roles is not None
+            else {"topic": self.topic},
+        )
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the Kafka relation has changed."""
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Check if the topic is created
+        # (the Kafka charm shared the credentials).
+        if "username" in diff.added and "password" in diff.added:
+            # Emit the default event (the one without an alias).
+            logger.info("topic created at %s", datetime.now())
+            self.on.topic_created.emit(event.relation, app=event.app, unit=event.unit)
+
+            # To avoid unnecessary application restarts do not trigger
+            # “endpoints_changed“ event if “topic_created“ is triggered.
+            return
+
+        # Emit an endpoints (bootstap-server) changed event if the Kakfa endpoints
+        # added or changed this info in the relation databag.
+        if "endpoints" in diff.added or "endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("endpoints changed on %s", datetime.now())
+            self.on.bootstrap_server_changed.emit(
+                event.relation, app=event.app, unit=event.unit
+            )  # here check if this is the right design
+            return
diff --git a/installers/charm/osm-mon/metadata.yaml b/installers/charm/osm-mon/metadata.yaml
index ee2f2f9..5bd1236 100644
--- a/installers/charm/osm-mon/metadata.yaml
+++ b/installers/charm/osm-mon/metadata.yaml
@@ -58,7 +58,7 @@
     interface: kafka
     limit: 1
   mongodb:
-    interface: mongodb
+    interface: mongodb_client
     limit: 1
   keystone:
     interface: keystone
diff --git a/installers/charm/osm-mon/src/charm.py b/installers/charm/osm-mon/src/charm.py
index e07d607..db72dfe 100755
--- a/installers/charm/osm-mon/src/charm.py
+++ b/installers/charm/osm-mon/src/charm.py
@@ -22,7 +22,7 @@
 #
 # Learn more at: https://juju.is/docs/sdk
 
-"""OSM NBI charm.
+"""OSM MON charm.
 
 See more: https://charmhub.io/osm
 """
@@ -30,6 +30,7 @@
 import logging
 from typing import Any, Dict
 
+from charms.data_platform_libs.v0.data_interfaces import DatabaseRequires
 from charms.kafka_k8s.v0.kafka import KafkaRequires, _KafkaAvailableEvent
 from charms.observability_libs.v1.kubernetes_service_patch import KubernetesServicePatch
 from charms.osm_libs.v0.utils import (
@@ -46,7 +47,7 @@
 from ops.main import main
 from ops.model import ActiveStatus, Container
 
-from legacy_interfaces import KeystoneClient, MongoClient, PrometheusClient
+from legacy_interfaces import KeystoneClient, PrometheusClient
 
 HOSTPATHS = [
     HostPath(
@@ -85,7 +86,7 @@
     def __init__(self, *args):
         super().__init__(*args)
         self.kafka = KafkaRequires(self)
-        self.mongodb_client = MongoClient(self, "mongodb")
+        self.mongodb_client = DatabaseRequires(self, "mongodb", database_name="osm")
         self.prometheus_client = PrometheusClient(self, "prometheus")
         self.keystone_client = KeystoneClient(self, "keystone")
         self.vca = VcaRequires(self)
@@ -174,16 +175,24 @@
             self.on.vca_data_changed: self._on_config_changed,
             self.on.kafka_available: self._on_config_changed,
             self.on["kafka"].relation_broken: self._on_required_relation_broken,
+            self.mongodb_client.on.database_created: self._on_config_changed,
+            self.on["mongodb"].relation_broken: self._on_required_relation_broken,
             # Action events
             self.on.get_debug_mode_information_action: self._on_get_debug_mode_information_action,
         }
-        for relation in [self.on[rel_name] for rel_name in ["mongodb", "prometheus", "keystone"]]:
+        for relation in [self.on[rel_name] for rel_name in ["prometheus", "keystone"]]:
             event_handler_mapping[relation.relation_changed] = self._on_config_changed
             event_handler_mapping[relation.relation_broken] = self._on_required_relation_broken
 
         for event, handler in event_handler_mapping.items():
             self.framework.observe(event, handler)
 
+    def _is_database_available(self) -> bool:
+        try:
+            return self.mongodb_client.is_resource_created()
+        except KeyError:
+            return False
+
     def _validate_config(self) -> None:
         """Validate charm configuration.
 
@@ -203,7 +212,7 @@
 
         if not self.kafka.host or not self.kafka.port:
             missing_relations.append("kafka")
-        if self.mongodb_client.is_missing_data_in_unit():
+        if not self._is_database_available():
             missing_relations.append("mongodb")
         if self.prometheus_client.is_missing_data_in_app():
             missing_relations.append("prometheus")
@@ -239,7 +248,7 @@
             "OSMMON_MESSAGE_PORT": self.kafka.port,
             # Database configuration
             "OSMMON_DATABASE_DRIVER": "mongo",
-            "OSMMON_DATABASE_URI": self.mongodb_client.connection_string,
+            "OSMMON_DATABASE_URI": self._get_mongodb_uri(),
             "OSMMON_DATABASE_COMMONKEY": self.config["database-commonkey"],
             # Prometheus/grafana configuration
             "OSMMON_PROMETHEUS_URL": f"http://{self.prometheus_client.hostname}:{self.prometheus_client.port}",
@@ -278,6 +287,9 @@
             },
         }
 
+    def _get_mongodb_uri(self):
+        return list(self.mongodb_client.fetch_relation_data().values())[0]["uris"]
+
     def _patch_k8s_service(self) -> None:
         port = ServicePort(SERVICE_PORT, name=f"{self.app.name}")
         self.service_patcher = KubernetesServicePatch(self, [port])
diff --git a/installers/charm/osm-mon/tests/integration/test_charm.py b/installers/charm/osm-mon/tests/integration/test_charm.py
index 12c261b..c5807e9 100644
--- a/installers/charm/osm-mon/tests/integration/test_charm.py
+++ b/installers/charm/osm-mon/tests/integration/test_charm.py
@@ -61,7 +61,7 @@
             charm, resources=resources, application_name=MON_APP, series="focal"
         ),
         ops_test.model.deploy(KAFKA_CHARM, application_name=KAFKA_APP, channel="stable"),
-        ops_test.model.deploy(MONGO_DB_CHARM, application_name=MONGO_DB_APP, channel="stable"),
+        ops_test.model.deploy(MONGO_DB_CHARM, application_name=MONGO_DB_APP, channel="edge"),
         ops_test.model.deploy(MARIADB_CHARM, application_name=MARIADB_APP, channel="stable"),
         ops_test.model.deploy(PROMETHEUS_CHARM, application_name=PROMETHEUS_APP, channel="stable"),
         ops_test.model.deploy(ZOOKEEPER_CHARM, application_name=ZOOKEEPER_APP, channel="stable"),
diff --git a/installers/charm/osm-mon/tests/unit/test_charm.py b/installers/charm/osm-mon/tests/unit/test_charm.py
index 5c9fd99..33598fe 100644
--- a/installers/charm/osm-mon/tests/unit/test_charm.py
+++ b/installers/charm/osm-mon/tests/unit/test_charm.py
@@ -72,7 +72,9 @@
     relation_id = harness.add_relation("mongodb", "mongodb")
     harness.add_relation_unit(relation_id, "mongodb/0")
     harness.update_relation_data(
-        relation_id, "mongodb/0", {"connection_string": "mongodb://:1234"}
+        relation_id,
+        "mongodb",
+        {"uris": "mongodb://:1234", "username": "user", "password": "password"},
     )
     relation_ids.append(relation_id)
     # Add kafka relation
diff --git a/installers/charm/osm-nbi/lib/charms/data_platform_libs/v0/data_interfaces.py b/installers/charm/osm-nbi/lib/charms/data_platform_libs/v0/data_interfaces.py
new file mode 100644
index 0000000..b3da5aa
--- /dev/null
+++ b/installers/charm/osm-nbi/lib/charms/data_platform_libs/v0/data_interfaces.py
@@ -0,0 +1,1130 @@
+# Copyright 2023 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.
+
+"""Library to manage the relation for the data-platform products.
+
+This library contains the Requires and Provides classes for handling the relation
+between an application and multiple managed application supported by the data-team:
+MySQL, Postgresql, MongoDB, Redis,  and Kakfa.
+
+### Database (MySQL, Postgresql, MongoDB, and Redis)
+
+#### Requires Charm
+This library is a uniform interface to a selection of common database
+metadata, with added custom events that add convenience to database management,
+and methods to consume the application related data.
+
+
+Following an example of using the DatabaseCreatedEvent, in the context of the
+application charm code:
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    DatabaseCreatedEvent,
+    DatabaseRequires,
+)
+
+class ApplicationCharm(CharmBase):
+    # Application charm that connects to database charms.
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Charm events defined in the database requires charm library.
+        self.database = DatabaseRequires(self, relation_name="database", database_name="database")
+        self.framework.observe(self.database.on.database_created, self._on_database_created)
+
+    def _on_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+
+        # Start application with rendered configuration
+        self._start_application(config_file)
+
+        # Set active status
+        self.unit.status = ActiveStatus("received database credentials")
+```
+
+As shown above, the library provides some custom events to handle specific situations,
+which are listed below:
+
+-  database_created: event emitted when the requested database is created.
+-  endpoints_changed: event emitted when the read/write endpoints of the database have changed.
+-  read_only_endpoints_changed: event emitted when the read-only endpoints of the database
+  have changed. Event is not triggered if read/write endpoints changed too.
+
+If it is needed to connect multiple database clusters to the same relation endpoint
+the application charm can implement the same code as if it would connect to only
+one database cluster (like the above code example).
+
+To differentiate multiple clusters connected to the same relation endpoint
+the application charm can use the name of the remote application:
+
+```python
+
+def _on_database_created(self, event: DatabaseCreatedEvent) -> None:
+    # Get the remote app name of the cluster that triggered this event
+    cluster = event.relation.app.name
+```
+
+It is also possible to provide an alias for each different database cluster/relation.
+
+So, it is possible to differentiate the clusters in two ways.
+The first is to use the remote application name, i.e., `event.relation.app.name`, as above.
+
+The second way is to use different event handlers to handle each cluster events.
+The implementation would be something like the following code:
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    DatabaseCreatedEvent,
+    DatabaseRequires,
+)
+
+class ApplicationCharm(CharmBase):
+    # Application charm that connects to database charms.
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Define the cluster aliases and one handler for each cluster database created event.
+        self.database = DatabaseRequires(
+            self,
+            relation_name="database",
+            database_name="database",
+            relations_aliases = ["cluster1", "cluster2"],
+        )
+        self.framework.observe(
+            self.database.on.cluster1_database_created, self._on_cluster1_database_created
+        )
+        self.framework.observe(
+            self.database.on.cluster2_database_created, self._on_cluster2_database_created
+        )
+
+    def _on_cluster1_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database on the cluster named cluster1
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+        ...
+
+    def _on_cluster2_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database on the cluster named cluster2
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+        ...
+
+```
+
+### Provider Charm
+
+Following an example of using the DatabaseRequestedEvent, in the context of the
+database charm code:
+
+```python
+from charms.data_platform_libs.v0.data_interfaces import DatabaseProvides
+
+class SampleCharm(CharmBase):
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        # Charm events defined in the database provides charm library.
+        self.provided_database = DatabaseProvides(self, relation_name="database")
+        self.framework.observe(self.provided_database.on.database_requested,
+            self._on_database_requested)
+        # Database generic helper
+        self.database = DatabaseHelper()
+
+    def _on_database_requested(self, event: DatabaseRequestedEvent) -> None:
+        # Handle the event triggered by a new database requested in the relation
+        # Retrieve the database name using the charm library.
+        db_name = event.database
+        # generate a new user credential
+        username = self.database.generate_user()
+        password = self.database.generate_password()
+        # set the credentials for the relation
+        self.provided_database.set_credentials(event.relation.id, username, password)
+        # set other variables for the relation event.set_tls("False")
+```
+As shown above, the library provides a custom event (database_requested) to handle
+the situation when an application charm requests a new database to be created.
+It's preferred to subscribe to this event instead of relation changed event to avoid
+creating a new database when other information other than a database name is
+exchanged in the relation databag.
+
+### Kafka
+
+This library is the interface to use and interact with the Kafka charm. This library contains
+custom events that add convenience to manage Kafka, and provides methods to consume the
+application related data.
+
+#### Requirer Charm
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    BootstrapServerChangedEvent,
+    KafkaRequires,
+    TopicCreatedEvent,
+)
+
+class ApplicationCharm(CharmBase):
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.kafka = KafkaRequires(self, "kafka_client", "test-topic")
+        self.framework.observe(
+            self.kafka.on.bootstrap_server_changed, self._on_kafka_bootstrap_server_changed
+        )
+        self.framework.observe(
+            self.kafka.on.topic_created, self._on_kafka_topic_created
+        )
+
+    def _on_kafka_bootstrap_server_changed(self, event: BootstrapServerChangedEvent):
+        # Event triggered when a bootstrap server was changed for this application
+
+        new_bootstrap_server = event.bootstrap_server
+        ...
+
+    def _on_kafka_topic_created(self, event: TopicCreatedEvent):
+        # Event triggered when a topic was created for this application
+        username = event.username
+        password = event.password
+        tls = event.tls
+        tls_ca= event.tls_ca
+        bootstrap_server event.bootstrap_server
+        consumer_group_prefic = event.consumer_group_prefix
+        zookeeper_uris = event.zookeeper_uris
+        ...
+
+```
+
+As shown above, the library provides some custom events to handle specific situations,
+which are listed below:
+
+- topic_created: event emitted when the requested topic is created.
+- bootstrap_server_changed: event emitted when the bootstrap server have changed.
+- credential_changed: event emitted when the credentials of Kafka changed.
+
+### Provider Charm
+
+Following the previous example, this is an example of the provider charm.
+
+```python
+class SampleCharm(CharmBase):
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    KafkaProvides,
+    TopicRequestedEvent,
+)
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Default charm events.
+        self.framework.observe(self.on.start, self._on_start)
+
+        # Charm events defined in the Kafka Provides charm library.
+        self.kafka_provider = KafkaProvides(self, relation_name="kafka_client")
+        self.framework.observe(self.kafka_provider.on.topic_requested, self._on_topic_requested)
+        # Kafka generic helper
+        self.kafka = KafkaHelper()
+
+    def _on_topic_requested(self, event: TopicRequestedEvent):
+        # Handle the on_topic_requested event.
+
+        topic = event.topic
+        relation_id = event.relation.id
+        # set connection info in the databag relation
+        self.kafka_provider.set_bootstrap_server(relation_id, self.kafka.get_bootstrap_server())
+        self.kafka_provider.set_credentials(relation_id, username=username, password=password)
+        self.kafka_provider.set_consumer_group_prefix(relation_id, ...)
+        self.kafka_provider.set_tls(relation_id, "False")
+        self.kafka_provider.set_zookeeper_uris(relation_id, ...)
+
+```
+As shown above, the library provides a custom event (topic_requested) to handle
+the situation when an application charm requests a new topic to be created.
+It is preferred to subscribe to this event instead of relation changed event to avoid
+creating a new topic when other information other than a topic name is
+exchanged in the relation databag.
+"""
+
+import json
+import logging
+from abc import ABC, abstractmethod
+from collections import namedtuple
+from datetime import datetime
+from typing import List, Optional
+
+from ops.charm import (
+    CharmBase,
+    CharmEvents,
+    RelationChangedEvent,
+    RelationEvent,
+    RelationJoinedEvent,
+)
+from ops.framework import EventSource, Object
+from ops.model import Relation
+
+# The unique Charmhub library identifier, never change it
+LIBID = "6c3e6b6680d64e9c89e611d1a15f65be"
+
+# Increment this major API version when introducing breaking changes
+LIBAPI = 0
+
+# Increment this PATCH version before using `charmcraft publish-lib` or reset
+# to 0 if you are raising the major API version
+LIBPATCH = 7
+
+PYDEPS = ["ops>=2.0.0"]
+
+logger = logging.getLogger(__name__)
+
+Diff = namedtuple("Diff", "added changed deleted")
+Diff.__doc__ = """
+A tuple for storing the diff between two data mappings.
+
+added - keys that were added
+changed - keys that still exist but have new values
+deleted - key that were deleted"""
+
+
+def diff(event: RelationChangedEvent, bucket: str) -> Diff:
+    """Retrieves the diff of the data in the relation changed databag.
+
+    Args:
+        event: relation changed event.
+        bucket: bucket of the databag (app or unit)
+
+    Returns:
+        a Diff instance containing the added, deleted and changed
+            keys from the event relation databag.
+    """
+    # Retrieve the old data from the data key in the application relation databag.
+    old_data = json.loads(event.relation.data[bucket].get("data", "{}"))
+    # Retrieve the new data from the event relation databag.
+    new_data = {
+        key: value for key, value in event.relation.data[event.app].items() if key != "data"
+    }
+
+    # These are the keys that were added to the databag and triggered this event.
+    added = new_data.keys() - old_data.keys()
+    # These are the keys that were removed from the databag and triggered this event.
+    deleted = old_data.keys() - new_data.keys()
+    # These are the keys that already existed in the databag,
+    # but had their values changed.
+    changed = {key for key in old_data.keys() & new_data.keys() if old_data[key] != new_data[key]}
+    # Convert the new_data to a serializable format and save it for a next diff check.
+    event.relation.data[bucket].update({"data": json.dumps(new_data)})
+
+    # Return the diff with all possible changes.
+    return Diff(added, changed, deleted)
+
+
+# Base DataProvides and DataRequires
+
+
+class DataProvides(Object, ABC):
+    """Base provides-side of the data products relation."""
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+        self.charm = charm
+        self.local_app = self.charm.model.app
+        self.local_unit = self.charm.unit
+        self.relation_name = relation_name
+        self.framework.observe(
+            charm.on[relation_name].relation_changed,
+            self._on_relation_changed,
+        )
+
+    def _diff(self, event: RelationChangedEvent) -> Diff:
+        """Retrieves the diff of the data in the relation changed databag.
+
+        Args:
+            event: relation changed event.
+
+        Returns:
+            a Diff instance containing the added, deleted and changed
+                keys from the event relation databag.
+        """
+        return diff(event, self.local_app)
+
+    @abstractmethod
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation data has changed."""
+        raise NotImplementedError
+
+    def fetch_relation_data(self) -> dict:
+        """Retrieves data from relation.
+
+        This function can be used to retrieve data from a relation
+        in the charm code when outside an event callback.
+
+        Returns:
+            a dict of the values stored in the relation data bag
+                for all relation instances (indexed by the relation id).
+        """
+        data = {}
+        for relation in self.relations:
+            data[relation.id] = {
+                key: value for key, value in relation.data[relation.app].items() if key != "data"
+            }
+        return data
+
+    def _update_relation_data(self, relation_id: int, data: dict) -> None:
+        """Updates a set of key-value pairs in the relation.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            data: dict containing the key-value pairs
+                that should be updated in the relation.
+        """
+        if self.local_unit.is_leader():
+            relation = self.charm.model.get_relation(self.relation_name, relation_id)
+            relation.data[self.local_app].update(data)
+
+    @property
+    def relations(self) -> List[Relation]:
+        """The list of Relation instances associated with this relation_name."""
+        return list(self.charm.model.relations[self.relation_name])
+
+    def set_credentials(self, relation_id: int, username: str, password: str) -> None:
+        """Set credentials.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            username: user that was created.
+            password: password of the created user.
+        """
+        self._update_relation_data(
+            relation_id,
+            {
+                "username": username,
+                "password": password,
+            },
+        )
+
+    def set_tls(self, relation_id: int, tls: str) -> None:
+        """Set whether TLS is enabled.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            tls: whether tls is enabled (True or False).
+        """
+        self._update_relation_data(relation_id, {"tls": tls})
+
+    def set_tls_ca(self, relation_id: int, tls_ca: str) -> None:
+        """Set the TLS CA in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            tls_ca: TLS certification authority.
+        """
+        self._update_relation_data(relation_id, {"tls_ca": tls_ca})
+
+
+class DataRequires(Object, ABC):
+    """Requires-side of the relation."""
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        extra_user_roles: str = None,
+    ):
+        """Manager of base client relations."""
+        super().__init__(charm, relation_name)
+        self.charm = charm
+        self.extra_user_roles = extra_user_roles
+        self.local_app = self.charm.model.app
+        self.local_unit = self.charm.unit
+        self.relation_name = relation_name
+        self.framework.observe(
+            self.charm.on[relation_name].relation_joined, self._on_relation_joined_event
+        )
+        self.framework.observe(
+            self.charm.on[relation_name].relation_changed, self._on_relation_changed_event
+        )
+
+    @abstractmethod
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the relation."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        raise NotImplementedError
+
+    def fetch_relation_data(self) -> dict:
+        """Retrieves data from relation.
+
+        This function can be used to retrieve data from a relation
+        in the charm code when outside an event callback.
+        Function cannot be used in `*-relation-broken` events and will raise an exception.
+
+        Returns:
+            a dict of the values stored in the relation data bag
+                for all relation instances (indexed by the relation ID).
+        """
+        data = {}
+        for relation in self.relations:
+            data[relation.id] = {
+                key: value for key, value in relation.data[relation.app].items() if key != "data"
+            }
+        return data
+
+    def _update_relation_data(self, relation_id: int, data: dict) -> None:
+        """Updates a set of key-value pairs in the relation.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            data: dict containing the key-value pairs
+                that should be updated in the relation.
+        """
+        if self.local_unit.is_leader():
+            relation = self.charm.model.get_relation(self.relation_name, relation_id)
+            relation.data[self.local_app].update(data)
+
+    def _diff(self, event: RelationChangedEvent) -> Diff:
+        """Retrieves the diff of the data in the relation changed databag.
+
+        Args:
+            event: relation changed event.
+
+        Returns:
+            a Diff instance containing the added, deleted and changed
+                keys from the event relation databag.
+        """
+        return diff(event, self.local_unit)
+
+    @property
+    def relations(self) -> List[Relation]:
+        """The list of Relation instances associated with this relation_name."""
+        return [
+            relation
+            for relation in self.charm.model.relations[self.relation_name]
+            if self._is_relation_active(relation)
+        ]
+
+    @staticmethod
+    def _is_relation_active(relation: Relation):
+        try:
+            _ = repr(relation.data)
+            return True
+        except RuntimeError:
+            return False
+
+    @staticmethod
+    def _is_resource_created_for_relation(relation: Relation):
+        return (
+            "username" in relation.data[relation.app] and "password" in relation.data[relation.app]
+        )
+
+    def is_resource_created(self, relation_id: Optional[int] = None) -> bool:
+        """Check if the resource has been created.
+
+        This function can be used to check if the Provider answered with data in the charm code
+        when outside an event callback.
+
+        Args:
+            relation_id (int, optional): When provided the check is done only for the relation id
+                provided, otherwise the check is done for all relations
+
+        Returns:
+            True or False
+
+        Raises:
+            IndexError: If relation_id is provided but that relation does not exist
+        """
+        if relation_id is not None:
+            try:
+                relation = [relation for relation in self.relations if relation.id == relation_id][
+                    0
+                ]
+                return self._is_resource_created_for_relation(relation)
+            except IndexError:
+                raise IndexError(f"relation id {relation_id} cannot be accessed")
+        else:
+            return (
+                all(
+                    [
+                        self._is_resource_created_for_relation(relation)
+                        for relation in self.relations
+                    ]
+                )
+                if self.relations
+                else False
+            )
+
+
+# General events
+
+
+class ExtraRoleEvent(RelationEvent):
+    """Base class for data events."""
+
+    @property
+    def extra_user_roles(self) -> Optional[str]:
+        """Returns the extra user roles that were requested."""
+        return self.relation.data[self.relation.app].get("extra-user-roles")
+
+
+class AuthenticationEvent(RelationEvent):
+    """Base class for authentication fields for events."""
+
+    @property
+    def username(self) -> Optional[str]:
+        """Returns the created username."""
+        return self.relation.data[self.relation.app].get("username")
+
+    @property
+    def password(self) -> Optional[str]:
+        """Returns the password for the created user."""
+        return self.relation.data[self.relation.app].get("password")
+
+    @property
+    def tls(self) -> Optional[str]:
+        """Returns whether TLS is configured."""
+        return self.relation.data[self.relation.app].get("tls")
+
+    @property
+    def tls_ca(self) -> Optional[str]:
+        """Returns TLS CA."""
+        return self.relation.data[self.relation.app].get("tls-ca")
+
+
+# Database related events and fields
+
+
+class DatabaseProvidesEvent(RelationEvent):
+    """Base class for database events."""
+
+    @property
+    def database(self) -> Optional[str]:
+        """Returns the database that was requested."""
+        return self.relation.data[self.relation.app].get("database")
+
+
+class DatabaseRequestedEvent(DatabaseProvidesEvent, ExtraRoleEvent):
+    """Event emitted when a new database is requested for use on this relation."""
+
+
+class DatabaseProvidesEvents(CharmEvents):
+    """Database events.
+
+    This class defines the events that the database can emit.
+    """
+
+    database_requested = EventSource(DatabaseRequestedEvent)
+
+
+class DatabaseRequiresEvent(RelationEvent):
+    """Base class for database events."""
+
+    @property
+    def endpoints(self) -> Optional[str]:
+        """Returns a comma separated list of read/write endpoints."""
+        return self.relation.data[self.relation.app].get("endpoints")
+
+    @property
+    def read_only_endpoints(self) -> Optional[str]:
+        """Returns a comma separated list of read only endpoints."""
+        return self.relation.data[self.relation.app].get("read-only-endpoints")
+
+    @property
+    def replset(self) -> Optional[str]:
+        """Returns the replicaset name.
+
+        MongoDB only.
+        """
+        return self.relation.data[self.relation.app].get("replset")
+
+    @property
+    def uris(self) -> Optional[str]:
+        """Returns the connection URIs.
+
+        MongoDB, Redis, OpenSearch.
+        """
+        return self.relation.data[self.relation.app].get("uris")
+
+    @property
+    def version(self) -> Optional[str]:
+        """Returns the version of the database.
+
+        Version as informed by the database daemon.
+        """
+        return self.relation.data[self.relation.app].get("version")
+
+
+class DatabaseCreatedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when a new database is created for use on this relation."""
+
+
+class DatabaseEndpointsChangedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when the read/write endpoints are changed."""
+
+
+class DatabaseReadOnlyEndpointsChangedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when the read only endpoints are changed."""
+
+
+class DatabaseRequiresEvents(CharmEvents):
+    """Database events.
+
+    This class defines the events that the database can emit.
+    """
+
+    database_created = EventSource(DatabaseCreatedEvent)
+    endpoints_changed = EventSource(DatabaseEndpointsChangedEvent)
+    read_only_endpoints_changed = EventSource(DatabaseReadOnlyEndpointsChangedEvent)
+
+
+# Database Provider and Requires
+
+
+class DatabaseProvides(DataProvides):
+    """Provider-side of the database relations."""
+
+    on = DatabaseProvidesEvents()
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Only the leader should handle this event.
+        if not self.local_unit.is_leader():
+            return
+
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Emit a database requested event if the setup key (database name and optional
+        # extra user roles) was added to the relation databag by the application.
+        if "database" in diff.added:
+            self.on.database_requested.emit(event.relation, app=event.app, unit=event.unit)
+
+    def set_endpoints(self, relation_id: int, connection_strings: str) -> None:
+        """Set database primary connections.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            connection_strings: database hosts and ports comma separated list.
+        """
+        self._update_relation_data(relation_id, {"endpoints": connection_strings})
+
+    def set_read_only_endpoints(self, relation_id: int, connection_strings: str) -> None:
+        """Set database replicas connection strings.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            connection_strings: database hosts and ports comma separated list.
+        """
+        self._update_relation_data(relation_id, {"read-only-endpoints": connection_strings})
+
+    def set_replset(self, relation_id: int, replset: str) -> None:
+        """Set replica set name in the application relation databag.
+
+        MongoDB only.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            replset: replica set name.
+        """
+        self._update_relation_data(relation_id, {"replset": replset})
+
+    def set_uris(self, relation_id: int, uris: str) -> None:
+        """Set the database connection URIs in the application relation databag.
+
+        MongoDB, Redis, and OpenSearch only.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            uris: connection URIs.
+        """
+        self._update_relation_data(relation_id, {"uris": uris})
+
+    def set_version(self, relation_id: int, version: str) -> None:
+        """Set the database version in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            version: database version.
+        """
+        self._update_relation_data(relation_id, {"version": version})
+
+
+class DatabaseRequires(DataRequires):
+    """Requires-side of the database relation."""
+
+    on = DatabaseRequiresEvents()
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        database_name: str,
+        extra_user_roles: str = None,
+        relations_aliases: List[str] = None,
+    ):
+        """Manager of database client relations."""
+        super().__init__(charm, relation_name, extra_user_roles)
+        self.database = database_name
+        self.relations_aliases = relations_aliases
+
+        # Define custom event names for each alias.
+        if relations_aliases:
+            # Ensure the number of aliases does not exceed the maximum
+            # of connections allowed in the specific relation.
+            relation_connection_limit = self.charm.meta.requires[relation_name].limit
+            if len(relations_aliases) != relation_connection_limit:
+                raise ValueError(
+                    f"The number of aliases must match the maximum number of connections allowed in the relation. "
+                    f"Expected {relation_connection_limit}, got {len(relations_aliases)}"
+                )
+
+            for relation_alias in relations_aliases:
+                self.on.define_event(f"{relation_alias}_database_created", DatabaseCreatedEvent)
+                self.on.define_event(
+                    f"{relation_alias}_endpoints_changed", DatabaseEndpointsChangedEvent
+                )
+                self.on.define_event(
+                    f"{relation_alias}_read_only_endpoints_changed",
+                    DatabaseReadOnlyEndpointsChangedEvent,
+                )
+
+    def _assign_relation_alias(self, relation_id: int) -> None:
+        """Assigns an alias to a relation.
+
+        This function writes in the unit data bag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+        """
+        # If no aliases were provided, return immediately.
+        if not self.relations_aliases:
+            return
+
+        # Return if an alias was already assigned to this relation
+        # (like when there are more than one unit joining the relation).
+        if (
+            self.charm.model.get_relation(self.relation_name, relation_id)
+            .data[self.local_unit]
+            .get("alias")
+        ):
+            return
+
+        # Retrieve the available aliases (the ones that weren't assigned to any relation).
+        available_aliases = self.relations_aliases[:]
+        for relation in self.charm.model.relations[self.relation_name]:
+            alias = relation.data[self.local_unit].get("alias")
+            if alias:
+                logger.debug("Alias %s was already assigned to relation %d", alias, relation.id)
+                available_aliases.remove(alias)
+
+        # Set the alias in the unit relation databag of the specific relation.
+        relation = self.charm.model.get_relation(self.relation_name, relation_id)
+        relation.data[self.local_unit].update({"alias": available_aliases[0]})
+
+    def _emit_aliased_event(self, event: RelationChangedEvent, event_name: str) -> None:
+        """Emit an aliased event to a particular relation if it has an alias.
+
+        Args:
+            event: the relation changed event that was received.
+            event_name: the name of the event to emit.
+        """
+        alias = self._get_relation_alias(event.relation.id)
+        if alias:
+            getattr(self.on, f"{alias}_{event_name}").emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+    def _get_relation_alias(self, relation_id: int) -> Optional[str]:
+        """Returns the relation alias.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+
+        Returns:
+            the relation alias or None if the relation was not found.
+        """
+        for relation in self.charm.model.relations[self.relation_name]:
+            if relation.id == relation_id:
+                return relation.data[self.local_unit].get("alias")
+        return None
+
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the database relation."""
+        # If relations aliases were provided, assign one to the relation.
+        self._assign_relation_alias(event.relation.id)
+
+        # Sets both database and extra user roles in the relation
+        # if the roles are provided. Otherwise, sets only the database.
+        if self.extra_user_roles:
+            self._update_relation_data(
+                event.relation.id,
+                {
+                    "database": self.database,
+                    "extra-user-roles": self.extra_user_roles,
+                },
+            )
+        else:
+            self._update_relation_data(event.relation.id, {"database": self.database})
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the database relation has changed."""
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Check if the database is created
+        # (the database charm shared the credentials).
+        if "username" in diff.added and "password" in diff.added:
+            # Emit the default event (the one without an alias).
+            logger.info("database created at %s", datetime.now())
+            self.on.database_created.emit(event.relation, app=event.app, unit=event.unit)
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "database_created")
+
+            # To avoid unnecessary application restarts do not trigger
+            # “endpoints_changed“ event if “database_created“ is triggered.
+            return
+
+        # Emit an endpoints changed event if the database
+        # added or changed this info in the relation databag.
+        if "endpoints" in diff.added or "endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("endpoints changed on %s", datetime.now())
+            self.on.endpoints_changed.emit(event.relation, app=event.app, unit=event.unit)
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "endpoints_changed")
+
+            # To avoid unnecessary application restarts do not trigger
+            # “read_only_endpoints_changed“ event if “endpoints_changed“ is triggered.
+            return
+
+        # Emit a read only endpoints changed event if the database
+        # added or changed this info in the relation databag.
+        if "read-only-endpoints" in diff.added or "read-only-endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("read-only-endpoints changed on %s", datetime.now())
+            self.on.read_only_endpoints_changed.emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "read_only_endpoints_changed")
+
+
+# Kafka related events
+
+
+class KafkaProvidesEvent(RelationEvent):
+    """Base class for Kafka events."""
+
+    @property
+    def topic(self) -> Optional[str]:
+        """Returns the topic that was requested."""
+        return self.relation.data[self.relation.app].get("topic")
+
+
+class TopicRequestedEvent(KafkaProvidesEvent, ExtraRoleEvent):
+    """Event emitted when a new topic is requested for use on this relation."""
+
+
+class KafkaProvidesEvents(CharmEvents):
+    """Kafka events.
+
+    This class defines the events that the Kafka can emit.
+    """
+
+    topic_requested = EventSource(TopicRequestedEvent)
+
+
+class KafkaRequiresEvent(RelationEvent):
+    """Base class for Kafka events."""
+
+    @property
+    def bootstrap_server(self) -> Optional[str]:
+        """Returns a a comma-seperated list of broker uris."""
+        return self.relation.data[self.relation.app].get("endpoints")
+
+    @property
+    def consumer_group_prefix(self) -> Optional[str]:
+        """Returns the consumer-group-prefix."""
+        return self.relation.data[self.relation.app].get("consumer-group-prefix")
+
+    @property
+    def zookeeper_uris(self) -> Optional[str]:
+        """Returns a comma separated list of Zookeeper uris."""
+        return self.relation.data[self.relation.app].get("zookeeper-uris")
+
+
+class TopicCreatedEvent(AuthenticationEvent, KafkaRequiresEvent):
+    """Event emitted when a new topic is created for use on this relation."""
+
+
+class BootstrapServerChangedEvent(AuthenticationEvent, KafkaRequiresEvent):
+    """Event emitted when the bootstrap server is changed."""
+
+
+class KafkaRequiresEvents(CharmEvents):
+    """Kafka events.
+
+    This class defines the events that the Kafka can emit.
+    """
+
+    topic_created = EventSource(TopicCreatedEvent)
+    bootstrap_server_changed = EventSource(BootstrapServerChangedEvent)
+
+
+# Kafka Provides and Requires
+
+
+class KafkaProvides(DataProvides):
+    """Provider-side of the Kafka relation."""
+
+    on = KafkaProvidesEvents()
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Only the leader should handle this event.
+        if not self.local_unit.is_leader():
+            return
+
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Emit a topic requested event if the setup key (topic name and optional
+        # extra user roles) was added to the relation databag by the application.
+        if "topic" in diff.added:
+            self.on.topic_requested.emit(event.relation, app=event.app, unit=event.unit)
+
+    def set_bootstrap_server(self, relation_id: int, bootstrap_server: str) -> None:
+        """Set the bootstrap server in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            bootstrap_server: the bootstrap server address.
+        """
+        self._update_relation_data(relation_id, {"endpoints": bootstrap_server})
+
+    def set_consumer_group_prefix(self, relation_id: int, consumer_group_prefix: str) -> None:
+        """Set the consumer group prefix in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            consumer_group_prefix: the consumer group prefix string.
+        """
+        self._update_relation_data(relation_id, {"consumer-group-prefix": consumer_group_prefix})
+
+    def set_zookeeper_uris(self, relation_id: int, zookeeper_uris: str) -> None:
+        """Set the zookeeper uris in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            zookeeper_uris: comma-seperated list of ZooKeeper server uris.
+        """
+        self._update_relation_data(relation_id, {"zookeeper-uris": zookeeper_uris})
+
+
+class KafkaRequires(DataRequires):
+    """Requires-side of the Kafka relation."""
+
+    on = KafkaRequiresEvents()
+
+    def __init__(self, charm, relation_name: str, topic: str, extra_user_roles: str = None):
+        """Manager of Kafka client relations."""
+        # super().__init__(charm, relation_name)
+        super().__init__(charm, relation_name, extra_user_roles)
+        self.charm = charm
+        self.topic = topic
+
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the Kafka relation."""
+        # Sets both topic and extra user roles in the relation
+        # if the roles are provided. Otherwise, sets only the topic.
+        self._update_relation_data(
+            event.relation.id,
+            {
+                "topic": self.topic,
+                "extra-user-roles": self.extra_user_roles,
+            }
+            if self.extra_user_roles is not None
+            else {"topic": self.topic},
+        )
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the Kafka relation has changed."""
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Check if the topic is created
+        # (the Kafka charm shared the credentials).
+        if "username" in diff.added and "password" in diff.added:
+            # Emit the default event (the one without an alias).
+            logger.info("topic created at %s", datetime.now())
+            self.on.topic_created.emit(event.relation, app=event.app, unit=event.unit)
+
+            # To avoid unnecessary application restarts do not trigger
+            # “endpoints_changed“ event if “topic_created“ is triggered.
+            return
+
+        # Emit an endpoints (bootstap-server) changed event if the Kakfa endpoints
+        # added or changed this info in the relation databag.
+        if "endpoints" in diff.added or "endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("endpoints changed on %s", datetime.now())
+            self.on.bootstrap_server_changed.emit(
+                event.relation, app=event.app, unit=event.unit
+            )  # here check if this is the right design
+            return
diff --git a/installers/charm/osm-nbi/metadata.yaml b/installers/charm/osm-nbi/metadata.yaml
index 7da2b4b..8a336c8 100644
--- a/installers/charm/osm-nbi/metadata.yaml
+++ b/installers/charm/osm-nbi/metadata.yaml
@@ -62,7 +62,7 @@
     interface: kafka
     limit: 1
   mongodb:
-    interface: mongodb
+    interface: mongodb_client
     limit: 1
   keystone:
     interface: keystone
diff --git a/installers/charm/osm-nbi/src/charm.py b/installers/charm/osm-nbi/src/charm.py
index 23ab054..8855de2 100755
--- a/installers/charm/osm-nbi/src/charm.py
+++ b/installers/charm/osm-nbi/src/charm.py
@@ -30,6 +30,7 @@
 import logging
 from typing import Any, Dict
 
+from charms.data_platform_libs.v0.data_interfaces import DatabaseRequires
 from charms.kafka_k8s.v0.kafka import KafkaEvents, KafkaRequires
 from charms.nginx_ingress_integrator.v0.ingress import IngressRequires
 from charms.observability_libs.v1.kubernetes_service_patch import KubernetesServicePatch
@@ -47,7 +48,7 @@
 from ops.main import main
 from ops.model import ActiveStatus, Container
 
-from legacy_interfaces import KeystoneClient, MongoClient, PrometheusClient
+from legacy_interfaces import KeystoneClient, PrometheusClient
 
 HOSTPATHS = [
     HostPath(
@@ -82,7 +83,9 @@
         )
         self.kafka = KafkaRequires(self)
         self.nbi = NbiProvides(self)
-        self.mongodb_client = MongoClient(self, "mongodb")
+        self.mongodb_client = DatabaseRequires(
+            self, "mongodb", database_name="osm", extra_user_roles="admin"
+        )
         self.prometheus_client = PrometheusClient(self, "prometheus")
         self.keystone_client = KeystoneClient(self, "keystone")
         self._observe_charm_events()
@@ -179,17 +182,25 @@
             # Relation events
             self.on.kafka_available: self._on_config_changed,
             self.on["kafka"].relation_broken: self._on_required_relation_broken,
+            self.mongodb_client.on.database_created: self._on_config_changed,
+            self.on["mongodb"].relation_broken: self._on_required_relation_broken,
             # Action events
             self.on.get_debug_mode_information_action: self._on_get_debug_mode_information_action,
             self.on.nbi_relation_joined: self._update_nbi_relation,
         }
-        for relation in [self.on[rel_name] for rel_name in ["mongodb", "prometheus", "keystone"]]:
+        for relation in [self.on[rel_name] for rel_name in ["prometheus", "keystone"]]:
             event_handler_mapping[relation.relation_changed] = self._on_config_changed
             event_handler_mapping[relation.relation_broken] = self._on_required_relation_broken
 
         for event, handler in event_handler_mapping.items():
             self.framework.observe(event, handler)
 
+    def _is_database_available(self) -> bool:
+        try:
+            return self.mongodb_client.is_resource_created()
+        except KeyError:
+            return False
+
     def _validate_config(self) -> None:
         """Validate charm configuration.
 
@@ -209,7 +220,7 @@
 
         if not self.kafka.host or not self.kafka.port:
             missing_relations.append("kafka")
-        if self.mongodb_client.is_missing_data_in_unit():
+        if not self._is_database_available():
             missing_relations.append("mongodb")
         if self.prometheus_client.is_missing_data_in_app():
             missing_relations.append("prometheus")
@@ -263,13 +274,13 @@
                         "OSMNBI_MESSAGE_DRIVER": "kafka",
                         # Database configuration
                         "OSMNBI_DATABASE_DRIVER": "mongo",
-                        "OSMNBI_DATABASE_URI": self.mongodb_client.connection_string,
+                        "OSMNBI_DATABASE_URI": self._get_mongodb_uri(),
                         "OSMNBI_DATABASE_COMMONKEY": self.config["database-commonkey"],
                         # Storage configuration
                         "OSMNBI_STORAGE_DRIVER": "mongo",
                         "OSMNBI_STORAGE_PATH": "/app/storage",
                         "OSMNBI_STORAGE_COLLECTION": "files",
-                        "OSMNBI_STORAGE_URI": self.mongodb_client.connection_string,
+                        "OSMNBI_STORAGE_URI": self._get_mongodb_uri(),
                         # Prometheus configuration
                         "OSMNBI_PROMETHEUS_HOST": self.prometheus_client.hostname,
                         "OSMNBI_PROMETHEUS_PORT": self.prometheus_client.port,
@@ -294,6 +305,9 @@
             },
         }
 
+    def _get_mongodb_uri(self):
+        return list(self.mongodb_client.fetch_relation_data().values())[0]["uris"]
+
 
 if __name__ == "__main__":  # pragma: no cover
     main(OsmNbiCharm)
diff --git a/installers/charm/osm-nbi/tests/integration/test_charm.py b/installers/charm/osm-nbi/tests/integration/test_charm.py
index ac35ea6..a714638 100644
--- a/installers/charm/osm-nbi/tests/integration/test_charm.py
+++ b/installers/charm/osm-nbi/tests/integration/test_charm.py
@@ -61,7 +61,7 @@
             charm, resources=resources, application_name=NBI_APP, series="focal"
         ),
         ops_test.model.deploy(KAFKA_CHARM, application_name=KAFKA_APP, channel="stable"),
-        ops_test.model.deploy(MONGO_DB_CHARM, application_name=MONGO_DB_APP, channel="stable"),
+        ops_test.model.deploy(MONGO_DB_CHARM, application_name=MONGO_DB_APP, channel="edge"),
         ops_test.model.deploy(MARIADB_CHARM, application_name=MARIADB_APP, channel="stable"),
         ops_test.model.deploy(ZOOKEEPER_CHARM, application_name=ZOOKEEPER_APP, channel="stable"),
         ops_test.model.deploy(PROMETHEUS_CHARM, application_name=PROMETHEUS_APP, channel="stable"),
diff --git a/installers/charm/osm-nbi/tests/unit/test_charm.py b/installers/charm/osm-nbi/tests/unit/test_charm.py
index 87afafa..b160419 100644
--- a/installers/charm/osm-nbi/tests/unit/test_charm.py
+++ b/installers/charm/osm-nbi/tests/unit/test_charm.py
@@ -37,6 +37,7 @@
     mocker.patch("charm.KubernetesServicePatch", lambda x, y: None)
     harness = Harness(OsmNbiCharm)
     harness.begin()
+    harness.container_pebble_ready(container_name)
     yield harness
     harness.cleanup()
 
@@ -81,7 +82,9 @@
     relation_id = harness.add_relation("mongodb", "mongodb")
     harness.add_relation_unit(relation_id, "mongodb/0")
     harness.update_relation_data(
-        relation_id, "mongodb/0", {"connection_string": "mongodb://:1234"}
+        relation_id,
+        "mongodb",
+        {"uris": "mongodb://:1234", "username": "user", "password": "password"},
     )
     relation_ids.append(relation_id)
     # Add kafka relation
diff --git a/installers/charm/osm-nbi/tox.ini b/installers/charm/osm-nbi/tox.ini
index c1bada0..5200e6c 100644
--- a/installers/charm/osm-nbi/tox.ini
+++ b/installers/charm/osm-nbi/tox.ini
@@ -88,7 +88,7 @@
 description = Run integration tests
 deps =
     pytest
-    juju
+    juju<3
     pytest-operator
     -r{toxinidir}/requirements.txt
 commands =
diff --git a/installers/charm/osm-pol/lib/charms/data_platform_libs/v0/data_interfaces.py b/installers/charm/osm-pol/lib/charms/data_platform_libs/v0/data_interfaces.py
new file mode 100644
index 0000000..b3da5aa
--- /dev/null
+++ b/installers/charm/osm-pol/lib/charms/data_platform_libs/v0/data_interfaces.py
@@ -0,0 +1,1130 @@
+# Copyright 2023 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.
+
+"""Library to manage the relation for the data-platform products.
+
+This library contains the Requires and Provides classes for handling the relation
+between an application and multiple managed application supported by the data-team:
+MySQL, Postgresql, MongoDB, Redis,  and Kakfa.
+
+### Database (MySQL, Postgresql, MongoDB, and Redis)
+
+#### Requires Charm
+This library is a uniform interface to a selection of common database
+metadata, with added custom events that add convenience to database management,
+and methods to consume the application related data.
+
+
+Following an example of using the DatabaseCreatedEvent, in the context of the
+application charm code:
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    DatabaseCreatedEvent,
+    DatabaseRequires,
+)
+
+class ApplicationCharm(CharmBase):
+    # Application charm that connects to database charms.
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Charm events defined in the database requires charm library.
+        self.database = DatabaseRequires(self, relation_name="database", database_name="database")
+        self.framework.observe(self.database.on.database_created, self._on_database_created)
+
+    def _on_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+
+        # Start application with rendered configuration
+        self._start_application(config_file)
+
+        # Set active status
+        self.unit.status = ActiveStatus("received database credentials")
+```
+
+As shown above, the library provides some custom events to handle specific situations,
+which are listed below:
+
+-  database_created: event emitted when the requested database is created.
+-  endpoints_changed: event emitted when the read/write endpoints of the database have changed.
+-  read_only_endpoints_changed: event emitted when the read-only endpoints of the database
+  have changed. Event is not triggered if read/write endpoints changed too.
+
+If it is needed to connect multiple database clusters to the same relation endpoint
+the application charm can implement the same code as if it would connect to only
+one database cluster (like the above code example).
+
+To differentiate multiple clusters connected to the same relation endpoint
+the application charm can use the name of the remote application:
+
+```python
+
+def _on_database_created(self, event: DatabaseCreatedEvent) -> None:
+    # Get the remote app name of the cluster that triggered this event
+    cluster = event.relation.app.name
+```
+
+It is also possible to provide an alias for each different database cluster/relation.
+
+So, it is possible to differentiate the clusters in two ways.
+The first is to use the remote application name, i.e., `event.relation.app.name`, as above.
+
+The second way is to use different event handlers to handle each cluster events.
+The implementation would be something like the following code:
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    DatabaseCreatedEvent,
+    DatabaseRequires,
+)
+
+class ApplicationCharm(CharmBase):
+    # Application charm that connects to database charms.
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Define the cluster aliases and one handler for each cluster database created event.
+        self.database = DatabaseRequires(
+            self,
+            relation_name="database",
+            database_name="database",
+            relations_aliases = ["cluster1", "cluster2"],
+        )
+        self.framework.observe(
+            self.database.on.cluster1_database_created, self._on_cluster1_database_created
+        )
+        self.framework.observe(
+            self.database.on.cluster2_database_created, self._on_cluster2_database_created
+        )
+
+    def _on_cluster1_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database on the cluster named cluster1
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+        ...
+
+    def _on_cluster2_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database on the cluster named cluster2
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+        ...
+
+```
+
+### Provider Charm
+
+Following an example of using the DatabaseRequestedEvent, in the context of the
+database charm code:
+
+```python
+from charms.data_platform_libs.v0.data_interfaces import DatabaseProvides
+
+class SampleCharm(CharmBase):
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        # Charm events defined in the database provides charm library.
+        self.provided_database = DatabaseProvides(self, relation_name="database")
+        self.framework.observe(self.provided_database.on.database_requested,
+            self._on_database_requested)
+        # Database generic helper
+        self.database = DatabaseHelper()
+
+    def _on_database_requested(self, event: DatabaseRequestedEvent) -> None:
+        # Handle the event triggered by a new database requested in the relation
+        # Retrieve the database name using the charm library.
+        db_name = event.database
+        # generate a new user credential
+        username = self.database.generate_user()
+        password = self.database.generate_password()
+        # set the credentials for the relation
+        self.provided_database.set_credentials(event.relation.id, username, password)
+        # set other variables for the relation event.set_tls("False")
+```
+As shown above, the library provides a custom event (database_requested) to handle
+the situation when an application charm requests a new database to be created.
+It's preferred to subscribe to this event instead of relation changed event to avoid
+creating a new database when other information other than a database name is
+exchanged in the relation databag.
+
+### Kafka
+
+This library is the interface to use and interact with the Kafka charm. This library contains
+custom events that add convenience to manage Kafka, and provides methods to consume the
+application related data.
+
+#### Requirer Charm
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    BootstrapServerChangedEvent,
+    KafkaRequires,
+    TopicCreatedEvent,
+)
+
+class ApplicationCharm(CharmBase):
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.kafka = KafkaRequires(self, "kafka_client", "test-topic")
+        self.framework.observe(
+            self.kafka.on.bootstrap_server_changed, self._on_kafka_bootstrap_server_changed
+        )
+        self.framework.observe(
+            self.kafka.on.topic_created, self._on_kafka_topic_created
+        )
+
+    def _on_kafka_bootstrap_server_changed(self, event: BootstrapServerChangedEvent):
+        # Event triggered when a bootstrap server was changed for this application
+
+        new_bootstrap_server = event.bootstrap_server
+        ...
+
+    def _on_kafka_topic_created(self, event: TopicCreatedEvent):
+        # Event triggered when a topic was created for this application
+        username = event.username
+        password = event.password
+        tls = event.tls
+        tls_ca= event.tls_ca
+        bootstrap_server event.bootstrap_server
+        consumer_group_prefic = event.consumer_group_prefix
+        zookeeper_uris = event.zookeeper_uris
+        ...
+
+```
+
+As shown above, the library provides some custom events to handle specific situations,
+which are listed below:
+
+- topic_created: event emitted when the requested topic is created.
+- bootstrap_server_changed: event emitted when the bootstrap server have changed.
+- credential_changed: event emitted when the credentials of Kafka changed.
+
+### Provider Charm
+
+Following the previous example, this is an example of the provider charm.
+
+```python
+class SampleCharm(CharmBase):
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    KafkaProvides,
+    TopicRequestedEvent,
+)
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Default charm events.
+        self.framework.observe(self.on.start, self._on_start)
+
+        # Charm events defined in the Kafka Provides charm library.
+        self.kafka_provider = KafkaProvides(self, relation_name="kafka_client")
+        self.framework.observe(self.kafka_provider.on.topic_requested, self._on_topic_requested)
+        # Kafka generic helper
+        self.kafka = KafkaHelper()
+
+    def _on_topic_requested(self, event: TopicRequestedEvent):
+        # Handle the on_topic_requested event.
+
+        topic = event.topic
+        relation_id = event.relation.id
+        # set connection info in the databag relation
+        self.kafka_provider.set_bootstrap_server(relation_id, self.kafka.get_bootstrap_server())
+        self.kafka_provider.set_credentials(relation_id, username=username, password=password)
+        self.kafka_provider.set_consumer_group_prefix(relation_id, ...)
+        self.kafka_provider.set_tls(relation_id, "False")
+        self.kafka_provider.set_zookeeper_uris(relation_id, ...)
+
+```
+As shown above, the library provides a custom event (topic_requested) to handle
+the situation when an application charm requests a new topic to be created.
+It is preferred to subscribe to this event instead of relation changed event to avoid
+creating a new topic when other information other than a topic name is
+exchanged in the relation databag.
+"""
+
+import json
+import logging
+from abc import ABC, abstractmethod
+from collections import namedtuple
+from datetime import datetime
+from typing import List, Optional
+
+from ops.charm import (
+    CharmBase,
+    CharmEvents,
+    RelationChangedEvent,
+    RelationEvent,
+    RelationJoinedEvent,
+)
+from ops.framework import EventSource, Object
+from ops.model import Relation
+
+# The unique Charmhub library identifier, never change it
+LIBID = "6c3e6b6680d64e9c89e611d1a15f65be"
+
+# Increment this major API version when introducing breaking changes
+LIBAPI = 0
+
+# Increment this PATCH version before using `charmcraft publish-lib` or reset
+# to 0 if you are raising the major API version
+LIBPATCH = 7
+
+PYDEPS = ["ops>=2.0.0"]
+
+logger = logging.getLogger(__name__)
+
+Diff = namedtuple("Diff", "added changed deleted")
+Diff.__doc__ = """
+A tuple for storing the diff between two data mappings.
+
+added - keys that were added
+changed - keys that still exist but have new values
+deleted - key that were deleted"""
+
+
+def diff(event: RelationChangedEvent, bucket: str) -> Diff:
+    """Retrieves the diff of the data in the relation changed databag.
+
+    Args:
+        event: relation changed event.
+        bucket: bucket of the databag (app or unit)
+
+    Returns:
+        a Diff instance containing the added, deleted and changed
+            keys from the event relation databag.
+    """
+    # Retrieve the old data from the data key in the application relation databag.
+    old_data = json.loads(event.relation.data[bucket].get("data", "{}"))
+    # Retrieve the new data from the event relation databag.
+    new_data = {
+        key: value for key, value in event.relation.data[event.app].items() if key != "data"
+    }
+
+    # These are the keys that were added to the databag and triggered this event.
+    added = new_data.keys() - old_data.keys()
+    # These are the keys that were removed from the databag and triggered this event.
+    deleted = old_data.keys() - new_data.keys()
+    # These are the keys that already existed in the databag,
+    # but had their values changed.
+    changed = {key for key in old_data.keys() & new_data.keys() if old_data[key] != new_data[key]}
+    # Convert the new_data to a serializable format and save it for a next diff check.
+    event.relation.data[bucket].update({"data": json.dumps(new_data)})
+
+    # Return the diff with all possible changes.
+    return Diff(added, changed, deleted)
+
+
+# Base DataProvides and DataRequires
+
+
+class DataProvides(Object, ABC):
+    """Base provides-side of the data products relation."""
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+        self.charm = charm
+        self.local_app = self.charm.model.app
+        self.local_unit = self.charm.unit
+        self.relation_name = relation_name
+        self.framework.observe(
+            charm.on[relation_name].relation_changed,
+            self._on_relation_changed,
+        )
+
+    def _diff(self, event: RelationChangedEvent) -> Diff:
+        """Retrieves the diff of the data in the relation changed databag.
+
+        Args:
+            event: relation changed event.
+
+        Returns:
+            a Diff instance containing the added, deleted and changed
+                keys from the event relation databag.
+        """
+        return diff(event, self.local_app)
+
+    @abstractmethod
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation data has changed."""
+        raise NotImplementedError
+
+    def fetch_relation_data(self) -> dict:
+        """Retrieves data from relation.
+
+        This function can be used to retrieve data from a relation
+        in the charm code when outside an event callback.
+
+        Returns:
+            a dict of the values stored in the relation data bag
+                for all relation instances (indexed by the relation id).
+        """
+        data = {}
+        for relation in self.relations:
+            data[relation.id] = {
+                key: value for key, value in relation.data[relation.app].items() if key != "data"
+            }
+        return data
+
+    def _update_relation_data(self, relation_id: int, data: dict) -> None:
+        """Updates a set of key-value pairs in the relation.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            data: dict containing the key-value pairs
+                that should be updated in the relation.
+        """
+        if self.local_unit.is_leader():
+            relation = self.charm.model.get_relation(self.relation_name, relation_id)
+            relation.data[self.local_app].update(data)
+
+    @property
+    def relations(self) -> List[Relation]:
+        """The list of Relation instances associated with this relation_name."""
+        return list(self.charm.model.relations[self.relation_name])
+
+    def set_credentials(self, relation_id: int, username: str, password: str) -> None:
+        """Set credentials.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            username: user that was created.
+            password: password of the created user.
+        """
+        self._update_relation_data(
+            relation_id,
+            {
+                "username": username,
+                "password": password,
+            },
+        )
+
+    def set_tls(self, relation_id: int, tls: str) -> None:
+        """Set whether TLS is enabled.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            tls: whether tls is enabled (True or False).
+        """
+        self._update_relation_data(relation_id, {"tls": tls})
+
+    def set_tls_ca(self, relation_id: int, tls_ca: str) -> None:
+        """Set the TLS CA in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            tls_ca: TLS certification authority.
+        """
+        self._update_relation_data(relation_id, {"tls_ca": tls_ca})
+
+
+class DataRequires(Object, ABC):
+    """Requires-side of the relation."""
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        extra_user_roles: str = None,
+    ):
+        """Manager of base client relations."""
+        super().__init__(charm, relation_name)
+        self.charm = charm
+        self.extra_user_roles = extra_user_roles
+        self.local_app = self.charm.model.app
+        self.local_unit = self.charm.unit
+        self.relation_name = relation_name
+        self.framework.observe(
+            self.charm.on[relation_name].relation_joined, self._on_relation_joined_event
+        )
+        self.framework.observe(
+            self.charm.on[relation_name].relation_changed, self._on_relation_changed_event
+        )
+
+    @abstractmethod
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the relation."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        raise NotImplementedError
+
+    def fetch_relation_data(self) -> dict:
+        """Retrieves data from relation.
+
+        This function can be used to retrieve data from a relation
+        in the charm code when outside an event callback.
+        Function cannot be used in `*-relation-broken` events and will raise an exception.
+
+        Returns:
+            a dict of the values stored in the relation data bag
+                for all relation instances (indexed by the relation ID).
+        """
+        data = {}
+        for relation in self.relations:
+            data[relation.id] = {
+                key: value for key, value in relation.data[relation.app].items() if key != "data"
+            }
+        return data
+
+    def _update_relation_data(self, relation_id: int, data: dict) -> None:
+        """Updates a set of key-value pairs in the relation.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            data: dict containing the key-value pairs
+                that should be updated in the relation.
+        """
+        if self.local_unit.is_leader():
+            relation = self.charm.model.get_relation(self.relation_name, relation_id)
+            relation.data[self.local_app].update(data)
+
+    def _diff(self, event: RelationChangedEvent) -> Diff:
+        """Retrieves the diff of the data in the relation changed databag.
+
+        Args:
+            event: relation changed event.
+
+        Returns:
+            a Diff instance containing the added, deleted and changed
+                keys from the event relation databag.
+        """
+        return diff(event, self.local_unit)
+
+    @property
+    def relations(self) -> List[Relation]:
+        """The list of Relation instances associated with this relation_name."""
+        return [
+            relation
+            for relation in self.charm.model.relations[self.relation_name]
+            if self._is_relation_active(relation)
+        ]
+
+    @staticmethod
+    def _is_relation_active(relation: Relation):
+        try:
+            _ = repr(relation.data)
+            return True
+        except RuntimeError:
+            return False
+
+    @staticmethod
+    def _is_resource_created_for_relation(relation: Relation):
+        return (
+            "username" in relation.data[relation.app] and "password" in relation.data[relation.app]
+        )
+
+    def is_resource_created(self, relation_id: Optional[int] = None) -> bool:
+        """Check if the resource has been created.
+
+        This function can be used to check if the Provider answered with data in the charm code
+        when outside an event callback.
+
+        Args:
+            relation_id (int, optional): When provided the check is done only for the relation id
+                provided, otherwise the check is done for all relations
+
+        Returns:
+            True or False
+
+        Raises:
+            IndexError: If relation_id is provided but that relation does not exist
+        """
+        if relation_id is not None:
+            try:
+                relation = [relation for relation in self.relations if relation.id == relation_id][
+                    0
+                ]
+                return self._is_resource_created_for_relation(relation)
+            except IndexError:
+                raise IndexError(f"relation id {relation_id} cannot be accessed")
+        else:
+            return (
+                all(
+                    [
+                        self._is_resource_created_for_relation(relation)
+                        for relation in self.relations
+                    ]
+                )
+                if self.relations
+                else False
+            )
+
+
+# General events
+
+
+class ExtraRoleEvent(RelationEvent):
+    """Base class for data events."""
+
+    @property
+    def extra_user_roles(self) -> Optional[str]:
+        """Returns the extra user roles that were requested."""
+        return self.relation.data[self.relation.app].get("extra-user-roles")
+
+
+class AuthenticationEvent(RelationEvent):
+    """Base class for authentication fields for events."""
+
+    @property
+    def username(self) -> Optional[str]:
+        """Returns the created username."""
+        return self.relation.data[self.relation.app].get("username")
+
+    @property
+    def password(self) -> Optional[str]:
+        """Returns the password for the created user."""
+        return self.relation.data[self.relation.app].get("password")
+
+    @property
+    def tls(self) -> Optional[str]:
+        """Returns whether TLS is configured."""
+        return self.relation.data[self.relation.app].get("tls")
+
+    @property
+    def tls_ca(self) -> Optional[str]:
+        """Returns TLS CA."""
+        return self.relation.data[self.relation.app].get("tls-ca")
+
+
+# Database related events and fields
+
+
+class DatabaseProvidesEvent(RelationEvent):
+    """Base class for database events."""
+
+    @property
+    def database(self) -> Optional[str]:
+        """Returns the database that was requested."""
+        return self.relation.data[self.relation.app].get("database")
+
+
+class DatabaseRequestedEvent(DatabaseProvidesEvent, ExtraRoleEvent):
+    """Event emitted when a new database is requested for use on this relation."""
+
+
+class DatabaseProvidesEvents(CharmEvents):
+    """Database events.
+
+    This class defines the events that the database can emit.
+    """
+
+    database_requested = EventSource(DatabaseRequestedEvent)
+
+
+class DatabaseRequiresEvent(RelationEvent):
+    """Base class for database events."""
+
+    @property
+    def endpoints(self) -> Optional[str]:
+        """Returns a comma separated list of read/write endpoints."""
+        return self.relation.data[self.relation.app].get("endpoints")
+
+    @property
+    def read_only_endpoints(self) -> Optional[str]:
+        """Returns a comma separated list of read only endpoints."""
+        return self.relation.data[self.relation.app].get("read-only-endpoints")
+
+    @property
+    def replset(self) -> Optional[str]:
+        """Returns the replicaset name.
+
+        MongoDB only.
+        """
+        return self.relation.data[self.relation.app].get("replset")
+
+    @property
+    def uris(self) -> Optional[str]:
+        """Returns the connection URIs.
+
+        MongoDB, Redis, OpenSearch.
+        """
+        return self.relation.data[self.relation.app].get("uris")
+
+    @property
+    def version(self) -> Optional[str]:
+        """Returns the version of the database.
+
+        Version as informed by the database daemon.
+        """
+        return self.relation.data[self.relation.app].get("version")
+
+
+class DatabaseCreatedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when a new database is created for use on this relation."""
+
+
+class DatabaseEndpointsChangedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when the read/write endpoints are changed."""
+
+
+class DatabaseReadOnlyEndpointsChangedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when the read only endpoints are changed."""
+
+
+class DatabaseRequiresEvents(CharmEvents):
+    """Database events.
+
+    This class defines the events that the database can emit.
+    """
+
+    database_created = EventSource(DatabaseCreatedEvent)
+    endpoints_changed = EventSource(DatabaseEndpointsChangedEvent)
+    read_only_endpoints_changed = EventSource(DatabaseReadOnlyEndpointsChangedEvent)
+
+
+# Database Provider and Requires
+
+
+class DatabaseProvides(DataProvides):
+    """Provider-side of the database relations."""
+
+    on = DatabaseProvidesEvents()
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Only the leader should handle this event.
+        if not self.local_unit.is_leader():
+            return
+
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Emit a database requested event if the setup key (database name and optional
+        # extra user roles) was added to the relation databag by the application.
+        if "database" in diff.added:
+            self.on.database_requested.emit(event.relation, app=event.app, unit=event.unit)
+
+    def set_endpoints(self, relation_id: int, connection_strings: str) -> None:
+        """Set database primary connections.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            connection_strings: database hosts and ports comma separated list.
+        """
+        self._update_relation_data(relation_id, {"endpoints": connection_strings})
+
+    def set_read_only_endpoints(self, relation_id: int, connection_strings: str) -> None:
+        """Set database replicas connection strings.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            connection_strings: database hosts and ports comma separated list.
+        """
+        self._update_relation_data(relation_id, {"read-only-endpoints": connection_strings})
+
+    def set_replset(self, relation_id: int, replset: str) -> None:
+        """Set replica set name in the application relation databag.
+
+        MongoDB only.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            replset: replica set name.
+        """
+        self._update_relation_data(relation_id, {"replset": replset})
+
+    def set_uris(self, relation_id: int, uris: str) -> None:
+        """Set the database connection URIs in the application relation databag.
+
+        MongoDB, Redis, and OpenSearch only.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            uris: connection URIs.
+        """
+        self._update_relation_data(relation_id, {"uris": uris})
+
+    def set_version(self, relation_id: int, version: str) -> None:
+        """Set the database version in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            version: database version.
+        """
+        self._update_relation_data(relation_id, {"version": version})
+
+
+class DatabaseRequires(DataRequires):
+    """Requires-side of the database relation."""
+
+    on = DatabaseRequiresEvents()
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        database_name: str,
+        extra_user_roles: str = None,
+        relations_aliases: List[str] = None,
+    ):
+        """Manager of database client relations."""
+        super().__init__(charm, relation_name, extra_user_roles)
+        self.database = database_name
+        self.relations_aliases = relations_aliases
+
+        # Define custom event names for each alias.
+        if relations_aliases:
+            # Ensure the number of aliases does not exceed the maximum
+            # of connections allowed in the specific relation.
+            relation_connection_limit = self.charm.meta.requires[relation_name].limit
+            if len(relations_aliases) != relation_connection_limit:
+                raise ValueError(
+                    f"The number of aliases must match the maximum number of connections allowed in the relation. "
+                    f"Expected {relation_connection_limit}, got {len(relations_aliases)}"
+                )
+
+            for relation_alias in relations_aliases:
+                self.on.define_event(f"{relation_alias}_database_created", DatabaseCreatedEvent)
+                self.on.define_event(
+                    f"{relation_alias}_endpoints_changed", DatabaseEndpointsChangedEvent
+                )
+                self.on.define_event(
+                    f"{relation_alias}_read_only_endpoints_changed",
+                    DatabaseReadOnlyEndpointsChangedEvent,
+                )
+
+    def _assign_relation_alias(self, relation_id: int) -> None:
+        """Assigns an alias to a relation.
+
+        This function writes in the unit data bag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+        """
+        # If no aliases were provided, return immediately.
+        if not self.relations_aliases:
+            return
+
+        # Return if an alias was already assigned to this relation
+        # (like when there are more than one unit joining the relation).
+        if (
+            self.charm.model.get_relation(self.relation_name, relation_id)
+            .data[self.local_unit]
+            .get("alias")
+        ):
+            return
+
+        # Retrieve the available aliases (the ones that weren't assigned to any relation).
+        available_aliases = self.relations_aliases[:]
+        for relation in self.charm.model.relations[self.relation_name]:
+            alias = relation.data[self.local_unit].get("alias")
+            if alias:
+                logger.debug("Alias %s was already assigned to relation %d", alias, relation.id)
+                available_aliases.remove(alias)
+
+        # Set the alias in the unit relation databag of the specific relation.
+        relation = self.charm.model.get_relation(self.relation_name, relation_id)
+        relation.data[self.local_unit].update({"alias": available_aliases[0]})
+
+    def _emit_aliased_event(self, event: RelationChangedEvent, event_name: str) -> None:
+        """Emit an aliased event to a particular relation if it has an alias.
+
+        Args:
+            event: the relation changed event that was received.
+            event_name: the name of the event to emit.
+        """
+        alias = self._get_relation_alias(event.relation.id)
+        if alias:
+            getattr(self.on, f"{alias}_{event_name}").emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+    def _get_relation_alias(self, relation_id: int) -> Optional[str]:
+        """Returns the relation alias.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+
+        Returns:
+            the relation alias or None if the relation was not found.
+        """
+        for relation in self.charm.model.relations[self.relation_name]:
+            if relation.id == relation_id:
+                return relation.data[self.local_unit].get("alias")
+        return None
+
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the database relation."""
+        # If relations aliases were provided, assign one to the relation.
+        self._assign_relation_alias(event.relation.id)
+
+        # Sets both database and extra user roles in the relation
+        # if the roles are provided. Otherwise, sets only the database.
+        if self.extra_user_roles:
+            self._update_relation_data(
+                event.relation.id,
+                {
+                    "database": self.database,
+                    "extra-user-roles": self.extra_user_roles,
+                },
+            )
+        else:
+            self._update_relation_data(event.relation.id, {"database": self.database})
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the database relation has changed."""
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Check if the database is created
+        # (the database charm shared the credentials).
+        if "username" in diff.added and "password" in diff.added:
+            # Emit the default event (the one without an alias).
+            logger.info("database created at %s", datetime.now())
+            self.on.database_created.emit(event.relation, app=event.app, unit=event.unit)
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "database_created")
+
+            # To avoid unnecessary application restarts do not trigger
+            # “endpoints_changed“ event if “database_created“ is triggered.
+            return
+
+        # Emit an endpoints changed event if the database
+        # added or changed this info in the relation databag.
+        if "endpoints" in diff.added or "endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("endpoints changed on %s", datetime.now())
+            self.on.endpoints_changed.emit(event.relation, app=event.app, unit=event.unit)
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "endpoints_changed")
+
+            # To avoid unnecessary application restarts do not trigger
+            # “read_only_endpoints_changed“ event if “endpoints_changed“ is triggered.
+            return
+
+        # Emit a read only endpoints changed event if the database
+        # added or changed this info in the relation databag.
+        if "read-only-endpoints" in diff.added or "read-only-endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("read-only-endpoints changed on %s", datetime.now())
+            self.on.read_only_endpoints_changed.emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "read_only_endpoints_changed")
+
+
+# Kafka related events
+
+
+class KafkaProvidesEvent(RelationEvent):
+    """Base class for Kafka events."""
+
+    @property
+    def topic(self) -> Optional[str]:
+        """Returns the topic that was requested."""
+        return self.relation.data[self.relation.app].get("topic")
+
+
+class TopicRequestedEvent(KafkaProvidesEvent, ExtraRoleEvent):
+    """Event emitted when a new topic is requested for use on this relation."""
+
+
+class KafkaProvidesEvents(CharmEvents):
+    """Kafka events.
+
+    This class defines the events that the Kafka can emit.
+    """
+
+    topic_requested = EventSource(TopicRequestedEvent)
+
+
+class KafkaRequiresEvent(RelationEvent):
+    """Base class for Kafka events."""
+
+    @property
+    def bootstrap_server(self) -> Optional[str]:
+        """Returns a a comma-seperated list of broker uris."""
+        return self.relation.data[self.relation.app].get("endpoints")
+
+    @property
+    def consumer_group_prefix(self) -> Optional[str]:
+        """Returns the consumer-group-prefix."""
+        return self.relation.data[self.relation.app].get("consumer-group-prefix")
+
+    @property
+    def zookeeper_uris(self) -> Optional[str]:
+        """Returns a comma separated list of Zookeeper uris."""
+        return self.relation.data[self.relation.app].get("zookeeper-uris")
+
+
+class TopicCreatedEvent(AuthenticationEvent, KafkaRequiresEvent):
+    """Event emitted when a new topic is created for use on this relation."""
+
+
+class BootstrapServerChangedEvent(AuthenticationEvent, KafkaRequiresEvent):
+    """Event emitted when the bootstrap server is changed."""
+
+
+class KafkaRequiresEvents(CharmEvents):
+    """Kafka events.
+
+    This class defines the events that the Kafka can emit.
+    """
+
+    topic_created = EventSource(TopicCreatedEvent)
+    bootstrap_server_changed = EventSource(BootstrapServerChangedEvent)
+
+
+# Kafka Provides and Requires
+
+
+class KafkaProvides(DataProvides):
+    """Provider-side of the Kafka relation."""
+
+    on = KafkaProvidesEvents()
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Only the leader should handle this event.
+        if not self.local_unit.is_leader():
+            return
+
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Emit a topic requested event if the setup key (topic name and optional
+        # extra user roles) was added to the relation databag by the application.
+        if "topic" in diff.added:
+            self.on.topic_requested.emit(event.relation, app=event.app, unit=event.unit)
+
+    def set_bootstrap_server(self, relation_id: int, bootstrap_server: str) -> None:
+        """Set the bootstrap server in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            bootstrap_server: the bootstrap server address.
+        """
+        self._update_relation_data(relation_id, {"endpoints": bootstrap_server})
+
+    def set_consumer_group_prefix(self, relation_id: int, consumer_group_prefix: str) -> None:
+        """Set the consumer group prefix in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            consumer_group_prefix: the consumer group prefix string.
+        """
+        self._update_relation_data(relation_id, {"consumer-group-prefix": consumer_group_prefix})
+
+    def set_zookeeper_uris(self, relation_id: int, zookeeper_uris: str) -> None:
+        """Set the zookeeper uris in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            zookeeper_uris: comma-seperated list of ZooKeeper server uris.
+        """
+        self._update_relation_data(relation_id, {"zookeeper-uris": zookeeper_uris})
+
+
+class KafkaRequires(DataRequires):
+    """Requires-side of the Kafka relation."""
+
+    on = KafkaRequiresEvents()
+
+    def __init__(self, charm, relation_name: str, topic: str, extra_user_roles: str = None):
+        """Manager of Kafka client relations."""
+        # super().__init__(charm, relation_name)
+        super().__init__(charm, relation_name, extra_user_roles)
+        self.charm = charm
+        self.topic = topic
+
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the Kafka relation."""
+        # Sets both topic and extra user roles in the relation
+        # if the roles are provided. Otherwise, sets only the topic.
+        self._update_relation_data(
+            event.relation.id,
+            {
+                "topic": self.topic,
+                "extra-user-roles": self.extra_user_roles,
+            }
+            if self.extra_user_roles is not None
+            else {"topic": self.topic},
+        )
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the Kafka relation has changed."""
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Check if the topic is created
+        # (the Kafka charm shared the credentials).
+        if "username" in diff.added and "password" in diff.added:
+            # Emit the default event (the one without an alias).
+            logger.info("topic created at %s", datetime.now())
+            self.on.topic_created.emit(event.relation, app=event.app, unit=event.unit)
+
+            # To avoid unnecessary application restarts do not trigger
+            # “endpoints_changed“ event if “topic_created“ is triggered.
+            return
+
+        # Emit an endpoints (bootstap-server) changed event if the Kakfa endpoints
+        # added or changed this info in the relation databag.
+        if "endpoints" in diff.added or "endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("endpoints changed on %s", datetime.now())
+            self.on.bootstrap_server_changed.emit(
+                event.relation, app=event.app, unit=event.unit
+            )  # here check if this is the right design
+            return
diff --git a/installers/charm/osm-pol/metadata.yaml b/installers/charm/osm-pol/metadata.yaml
index d6bb35a..adf189a 100644
--- a/installers/charm/osm-pol/metadata.yaml
+++ b/installers/charm/osm-pol/metadata.yaml
@@ -56,7 +56,7 @@
     interface: kafka
     limit: 1
   mongodb:
-    interface: mongodb
+    interface: mongodb_client
     limit: 1
   mysql:
     interface: mysql
diff --git a/installers/charm/osm-pol/src/charm.py b/installers/charm/osm-pol/src/charm.py
index 2749ddb..07bf87e 100755
--- a/installers/charm/osm-pol/src/charm.py
+++ b/installers/charm/osm-pol/src/charm.py
@@ -30,6 +30,7 @@
 import logging
 from typing import Any, Dict
 
+from charms.data_platform_libs.v0.data_interfaces import DatabaseRequires
 from charms.kafka_k8s.v0.kafka import KafkaEvents, KafkaRequires
 from charms.osm_libs.v0.utils import (
     CharmError,
@@ -43,7 +44,7 @@
 from ops.main import main
 from ops.model import ActiveStatus, Container
 
-from legacy_interfaces import MongoClient, MysqlClient
+from legacy_interfaces import MysqlClient
 
 HOSTPATHS = [
     HostPath(
@@ -71,7 +72,7 @@
         super().__init__(*args)
 
         self.kafka = KafkaRequires(self)
-        self.mongodb_client = MongoClient(self, "mongodb")
+        self.mongodb_client = DatabaseRequires(self, "mongodb", database_name="osm")
         self.mysql_client = MysqlClient(self, "mysql")
         self._observe_charm_events()
         self.container: Container = self.unit.get_container(self.container_name)
@@ -145,16 +146,23 @@
             # Relation events
             self.on.kafka_available: self._on_config_changed,
             self.on["kafka"].relation_broken: self._on_required_relation_broken,
+            self.on["mysql"].relation_changed: self._on_config_changed,
+            self.on["mysql"].relation_broken: self._on_config_changed,
+            self.mongodb_client.on.database_created: self._on_config_changed,
+            self.on["mongodb"].relation_broken: self._on_required_relation_broken,
             # Action events
             self.on.get_debug_mode_information_action: self._on_get_debug_mode_information_action,
         }
-        for relation in [self.on[rel_name] for rel_name in ["mongodb", "mysql"]]:
-            event_handler_mapping[relation.relation_changed] = self._on_config_changed
-            event_handler_mapping[relation.relation_broken] = self._on_required_relation_broken
 
         for event, handler in event_handler_mapping.items():
             self.framework.observe(event, handler)
 
+    def _is_database_available(self) -> bool:
+        try:
+            return self.mongodb_client.is_resource_created()
+        except KeyError:
+            return False
+
     def _validate_config(self) -> None:
         """Validate charm configuration.
 
@@ -174,7 +182,7 @@
 
         if not self.kafka.host or not self.kafka.port:
             missing_relations.append("kafka")
-        if self.mongodb_client.is_missing_data_in_unit():
+        if not self._is_database_available():
             missing_relations.append("mongodb")
         if not self.config.get("mysql-uri") and self.mysql_client.is_missing_data_in_unit():
             missing_relations.append("mysql")
@@ -214,7 +222,7 @@
                         "OSMPOL_MESSAGE_DRIVER": "kafka",
                         # Database Mongodb configuration
                         "OSMPOL_DATABASE_DRIVER": "mongo",
-                        "OSMPOL_DATABASE_URI": self.mongodb_client.connection_string,
+                        "OSMPOL_DATABASE_URI": self._get_mongodb_uri(),
                         # Database MySQL configuration
                         "OSMPOL_SQL_DATABASE_URI": self._get_mysql_uri(),
                     },
@@ -225,6 +233,9 @@
     def _get_mysql_uri(self):
         return self.config.get("mysql-uri") or self.mysql_client.get_root_uri("pol")
 
+    def _get_mongodb_uri(self):
+        return list(self.mongodb_client.fetch_relation_data().values())[0]["uris"]
+
 
 if __name__ == "__main__":  # pragma: no cover
     main(OsmPolCharm)
diff --git a/installers/charm/osm-pol/tests/integration/test_charm.py b/installers/charm/osm-pol/tests/integration/test_charm.py
index e2f3307..87132d1 100644
--- a/installers/charm/osm-pol/tests/integration/test_charm.py
+++ b/installers/charm/osm-pol/tests/integration/test_charm.py
@@ -54,7 +54,7 @@
             charm, resources=resources, application_name=POL_APP, series="focal"
         ),
         ops_test.model.deploy(KAFKA_CHARM, application_name=KAFKA_APP, channel="stable"),
-        ops_test.model.deploy(MONGO_DB_CHARM, application_name=MONGO_DB_APP, channel="stable"),
+        ops_test.model.deploy(MONGO_DB_CHARM, application_name=MONGO_DB_APP, channel="edge"),
         ops_test.model.deploy(MARIADB_CHARM, application_name=MARIADB_APP, channel="stable"),
         ops_test.model.deploy(ZOOKEEPER_CHARM, application_name=ZOOKEEPER_APP, channel="stable"),
     )
diff --git a/installers/charm/osm-pol/tests/unit/test_charm.py b/installers/charm/osm-pol/tests/unit/test_charm.py
index 916a73b..1b5013a 100644
--- a/installers/charm/osm-pol/tests/unit/test_charm.py
+++ b/installers/charm/osm-pol/tests/unit/test_charm.py
@@ -70,7 +70,9 @@
     relation_id = harness.add_relation("mongodb", "mongodb")
     harness.add_relation_unit(relation_id, "mongodb/0")
     harness.update_relation_data(
-        relation_id, "mongodb/0", {"connection_string": "mongodb://:1234"}
+        relation_id,
+        "mongodb",
+        {"uris": "mongodb://:1234", "username": "user", "password": "password"},
     )
     relation_ids.append(relation_id)
     # Add kafka relation
diff --git a/installers/charm/osm-ro/lib/charms/data_platform_libs/v0/data_interfaces.py b/installers/charm/osm-ro/lib/charms/data_platform_libs/v0/data_interfaces.py
new file mode 100644
index 0000000..b3da5aa
--- /dev/null
+++ b/installers/charm/osm-ro/lib/charms/data_platform_libs/v0/data_interfaces.py
@@ -0,0 +1,1130 @@
+# Copyright 2023 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.
+
+"""Library to manage the relation for the data-platform products.
+
+This library contains the Requires and Provides classes for handling the relation
+between an application and multiple managed application supported by the data-team:
+MySQL, Postgresql, MongoDB, Redis,  and Kakfa.
+
+### Database (MySQL, Postgresql, MongoDB, and Redis)
+
+#### Requires Charm
+This library is a uniform interface to a selection of common database
+metadata, with added custom events that add convenience to database management,
+and methods to consume the application related data.
+
+
+Following an example of using the DatabaseCreatedEvent, in the context of the
+application charm code:
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    DatabaseCreatedEvent,
+    DatabaseRequires,
+)
+
+class ApplicationCharm(CharmBase):
+    # Application charm that connects to database charms.
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Charm events defined in the database requires charm library.
+        self.database = DatabaseRequires(self, relation_name="database", database_name="database")
+        self.framework.observe(self.database.on.database_created, self._on_database_created)
+
+    def _on_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+
+        # Start application with rendered configuration
+        self._start_application(config_file)
+
+        # Set active status
+        self.unit.status = ActiveStatus("received database credentials")
+```
+
+As shown above, the library provides some custom events to handle specific situations,
+which are listed below:
+
+-  database_created: event emitted when the requested database is created.
+-  endpoints_changed: event emitted when the read/write endpoints of the database have changed.
+-  read_only_endpoints_changed: event emitted when the read-only endpoints of the database
+  have changed. Event is not triggered if read/write endpoints changed too.
+
+If it is needed to connect multiple database clusters to the same relation endpoint
+the application charm can implement the same code as if it would connect to only
+one database cluster (like the above code example).
+
+To differentiate multiple clusters connected to the same relation endpoint
+the application charm can use the name of the remote application:
+
+```python
+
+def _on_database_created(self, event: DatabaseCreatedEvent) -> None:
+    # Get the remote app name of the cluster that triggered this event
+    cluster = event.relation.app.name
+```
+
+It is also possible to provide an alias for each different database cluster/relation.
+
+So, it is possible to differentiate the clusters in two ways.
+The first is to use the remote application name, i.e., `event.relation.app.name`, as above.
+
+The second way is to use different event handlers to handle each cluster events.
+The implementation would be something like the following code:
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    DatabaseCreatedEvent,
+    DatabaseRequires,
+)
+
+class ApplicationCharm(CharmBase):
+    # Application charm that connects to database charms.
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Define the cluster aliases and one handler for each cluster database created event.
+        self.database = DatabaseRequires(
+            self,
+            relation_name="database",
+            database_name="database",
+            relations_aliases = ["cluster1", "cluster2"],
+        )
+        self.framework.observe(
+            self.database.on.cluster1_database_created, self._on_cluster1_database_created
+        )
+        self.framework.observe(
+            self.database.on.cluster2_database_created, self._on_cluster2_database_created
+        )
+
+    def _on_cluster1_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database on the cluster named cluster1
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+        ...
+
+    def _on_cluster2_database_created(self, event: DatabaseCreatedEvent) -> None:
+        # Handle the created database on the cluster named cluster2
+
+        # Create configuration file for app
+        config_file = self._render_app_config_file(
+            event.username,
+            event.password,
+            event.endpoints,
+        )
+        ...
+
+```
+
+### Provider Charm
+
+Following an example of using the DatabaseRequestedEvent, in the context of the
+database charm code:
+
+```python
+from charms.data_platform_libs.v0.data_interfaces import DatabaseProvides
+
+class SampleCharm(CharmBase):
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        # Charm events defined in the database provides charm library.
+        self.provided_database = DatabaseProvides(self, relation_name="database")
+        self.framework.observe(self.provided_database.on.database_requested,
+            self._on_database_requested)
+        # Database generic helper
+        self.database = DatabaseHelper()
+
+    def _on_database_requested(self, event: DatabaseRequestedEvent) -> None:
+        # Handle the event triggered by a new database requested in the relation
+        # Retrieve the database name using the charm library.
+        db_name = event.database
+        # generate a new user credential
+        username = self.database.generate_user()
+        password = self.database.generate_password()
+        # set the credentials for the relation
+        self.provided_database.set_credentials(event.relation.id, username, password)
+        # set other variables for the relation event.set_tls("False")
+```
+As shown above, the library provides a custom event (database_requested) to handle
+the situation when an application charm requests a new database to be created.
+It's preferred to subscribe to this event instead of relation changed event to avoid
+creating a new database when other information other than a database name is
+exchanged in the relation databag.
+
+### Kafka
+
+This library is the interface to use and interact with the Kafka charm. This library contains
+custom events that add convenience to manage Kafka, and provides methods to consume the
+application related data.
+
+#### Requirer Charm
+
+```python
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    BootstrapServerChangedEvent,
+    KafkaRequires,
+    TopicCreatedEvent,
+)
+
+class ApplicationCharm(CharmBase):
+
+    def __init__(self, *args):
+        super().__init__(*args)
+        self.kafka = KafkaRequires(self, "kafka_client", "test-topic")
+        self.framework.observe(
+            self.kafka.on.bootstrap_server_changed, self._on_kafka_bootstrap_server_changed
+        )
+        self.framework.observe(
+            self.kafka.on.topic_created, self._on_kafka_topic_created
+        )
+
+    def _on_kafka_bootstrap_server_changed(self, event: BootstrapServerChangedEvent):
+        # Event triggered when a bootstrap server was changed for this application
+
+        new_bootstrap_server = event.bootstrap_server
+        ...
+
+    def _on_kafka_topic_created(self, event: TopicCreatedEvent):
+        # Event triggered when a topic was created for this application
+        username = event.username
+        password = event.password
+        tls = event.tls
+        tls_ca= event.tls_ca
+        bootstrap_server event.bootstrap_server
+        consumer_group_prefic = event.consumer_group_prefix
+        zookeeper_uris = event.zookeeper_uris
+        ...
+
+```
+
+As shown above, the library provides some custom events to handle specific situations,
+which are listed below:
+
+- topic_created: event emitted when the requested topic is created.
+- bootstrap_server_changed: event emitted when the bootstrap server have changed.
+- credential_changed: event emitted when the credentials of Kafka changed.
+
+### Provider Charm
+
+Following the previous example, this is an example of the provider charm.
+
+```python
+class SampleCharm(CharmBase):
+
+from charms.data_platform_libs.v0.data_interfaces import (
+    KafkaProvides,
+    TopicRequestedEvent,
+)
+
+    def __init__(self, *args):
+        super().__init__(*args)
+
+        # Default charm events.
+        self.framework.observe(self.on.start, self._on_start)
+
+        # Charm events defined in the Kafka Provides charm library.
+        self.kafka_provider = KafkaProvides(self, relation_name="kafka_client")
+        self.framework.observe(self.kafka_provider.on.topic_requested, self._on_topic_requested)
+        # Kafka generic helper
+        self.kafka = KafkaHelper()
+
+    def _on_topic_requested(self, event: TopicRequestedEvent):
+        # Handle the on_topic_requested event.
+
+        topic = event.topic
+        relation_id = event.relation.id
+        # set connection info in the databag relation
+        self.kafka_provider.set_bootstrap_server(relation_id, self.kafka.get_bootstrap_server())
+        self.kafka_provider.set_credentials(relation_id, username=username, password=password)
+        self.kafka_provider.set_consumer_group_prefix(relation_id, ...)
+        self.kafka_provider.set_tls(relation_id, "False")
+        self.kafka_provider.set_zookeeper_uris(relation_id, ...)
+
+```
+As shown above, the library provides a custom event (topic_requested) to handle
+the situation when an application charm requests a new topic to be created.
+It is preferred to subscribe to this event instead of relation changed event to avoid
+creating a new topic when other information other than a topic name is
+exchanged in the relation databag.
+"""
+
+import json
+import logging
+from abc import ABC, abstractmethod
+from collections import namedtuple
+from datetime import datetime
+from typing import List, Optional
+
+from ops.charm import (
+    CharmBase,
+    CharmEvents,
+    RelationChangedEvent,
+    RelationEvent,
+    RelationJoinedEvent,
+)
+from ops.framework import EventSource, Object
+from ops.model import Relation
+
+# The unique Charmhub library identifier, never change it
+LIBID = "6c3e6b6680d64e9c89e611d1a15f65be"
+
+# Increment this major API version when introducing breaking changes
+LIBAPI = 0
+
+# Increment this PATCH version before using `charmcraft publish-lib` or reset
+# to 0 if you are raising the major API version
+LIBPATCH = 7
+
+PYDEPS = ["ops>=2.0.0"]
+
+logger = logging.getLogger(__name__)
+
+Diff = namedtuple("Diff", "added changed deleted")
+Diff.__doc__ = """
+A tuple for storing the diff between two data mappings.
+
+added - keys that were added
+changed - keys that still exist but have new values
+deleted - key that were deleted"""
+
+
+def diff(event: RelationChangedEvent, bucket: str) -> Diff:
+    """Retrieves the diff of the data in the relation changed databag.
+
+    Args:
+        event: relation changed event.
+        bucket: bucket of the databag (app or unit)
+
+    Returns:
+        a Diff instance containing the added, deleted and changed
+            keys from the event relation databag.
+    """
+    # Retrieve the old data from the data key in the application relation databag.
+    old_data = json.loads(event.relation.data[bucket].get("data", "{}"))
+    # Retrieve the new data from the event relation databag.
+    new_data = {
+        key: value for key, value in event.relation.data[event.app].items() if key != "data"
+    }
+
+    # These are the keys that were added to the databag and triggered this event.
+    added = new_data.keys() - old_data.keys()
+    # These are the keys that were removed from the databag and triggered this event.
+    deleted = old_data.keys() - new_data.keys()
+    # These are the keys that already existed in the databag,
+    # but had their values changed.
+    changed = {key for key in old_data.keys() & new_data.keys() if old_data[key] != new_data[key]}
+    # Convert the new_data to a serializable format and save it for a next diff check.
+    event.relation.data[bucket].update({"data": json.dumps(new_data)})
+
+    # Return the diff with all possible changes.
+    return Diff(added, changed, deleted)
+
+
+# Base DataProvides and DataRequires
+
+
+class DataProvides(Object, ABC):
+    """Base provides-side of the data products relation."""
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+        self.charm = charm
+        self.local_app = self.charm.model.app
+        self.local_unit = self.charm.unit
+        self.relation_name = relation_name
+        self.framework.observe(
+            charm.on[relation_name].relation_changed,
+            self._on_relation_changed,
+        )
+
+    def _diff(self, event: RelationChangedEvent) -> Diff:
+        """Retrieves the diff of the data in the relation changed databag.
+
+        Args:
+            event: relation changed event.
+
+        Returns:
+            a Diff instance containing the added, deleted and changed
+                keys from the event relation databag.
+        """
+        return diff(event, self.local_app)
+
+    @abstractmethod
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation data has changed."""
+        raise NotImplementedError
+
+    def fetch_relation_data(self) -> dict:
+        """Retrieves data from relation.
+
+        This function can be used to retrieve data from a relation
+        in the charm code when outside an event callback.
+
+        Returns:
+            a dict of the values stored in the relation data bag
+                for all relation instances (indexed by the relation id).
+        """
+        data = {}
+        for relation in self.relations:
+            data[relation.id] = {
+                key: value for key, value in relation.data[relation.app].items() if key != "data"
+            }
+        return data
+
+    def _update_relation_data(self, relation_id: int, data: dict) -> None:
+        """Updates a set of key-value pairs in the relation.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            data: dict containing the key-value pairs
+                that should be updated in the relation.
+        """
+        if self.local_unit.is_leader():
+            relation = self.charm.model.get_relation(self.relation_name, relation_id)
+            relation.data[self.local_app].update(data)
+
+    @property
+    def relations(self) -> List[Relation]:
+        """The list of Relation instances associated with this relation_name."""
+        return list(self.charm.model.relations[self.relation_name])
+
+    def set_credentials(self, relation_id: int, username: str, password: str) -> None:
+        """Set credentials.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            username: user that was created.
+            password: password of the created user.
+        """
+        self._update_relation_data(
+            relation_id,
+            {
+                "username": username,
+                "password": password,
+            },
+        )
+
+    def set_tls(self, relation_id: int, tls: str) -> None:
+        """Set whether TLS is enabled.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            tls: whether tls is enabled (True or False).
+        """
+        self._update_relation_data(relation_id, {"tls": tls})
+
+    def set_tls_ca(self, relation_id: int, tls_ca: str) -> None:
+        """Set the TLS CA in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            tls_ca: TLS certification authority.
+        """
+        self._update_relation_data(relation_id, {"tls_ca": tls_ca})
+
+
+class DataRequires(Object, ABC):
+    """Requires-side of the relation."""
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        extra_user_roles: str = None,
+    ):
+        """Manager of base client relations."""
+        super().__init__(charm, relation_name)
+        self.charm = charm
+        self.extra_user_roles = extra_user_roles
+        self.local_app = self.charm.model.app
+        self.local_unit = self.charm.unit
+        self.relation_name = relation_name
+        self.framework.observe(
+            self.charm.on[relation_name].relation_joined, self._on_relation_joined_event
+        )
+        self.framework.observe(
+            self.charm.on[relation_name].relation_changed, self._on_relation_changed_event
+        )
+
+    @abstractmethod
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the relation."""
+        raise NotImplementedError
+
+    @abstractmethod
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        raise NotImplementedError
+
+    def fetch_relation_data(self) -> dict:
+        """Retrieves data from relation.
+
+        This function can be used to retrieve data from a relation
+        in the charm code when outside an event callback.
+        Function cannot be used in `*-relation-broken` events and will raise an exception.
+
+        Returns:
+            a dict of the values stored in the relation data bag
+                for all relation instances (indexed by the relation ID).
+        """
+        data = {}
+        for relation in self.relations:
+            data[relation.id] = {
+                key: value for key, value in relation.data[relation.app].items() if key != "data"
+            }
+        return data
+
+    def _update_relation_data(self, relation_id: int, data: dict) -> None:
+        """Updates a set of key-value pairs in the relation.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            data: dict containing the key-value pairs
+                that should be updated in the relation.
+        """
+        if self.local_unit.is_leader():
+            relation = self.charm.model.get_relation(self.relation_name, relation_id)
+            relation.data[self.local_app].update(data)
+
+    def _diff(self, event: RelationChangedEvent) -> Diff:
+        """Retrieves the diff of the data in the relation changed databag.
+
+        Args:
+            event: relation changed event.
+
+        Returns:
+            a Diff instance containing the added, deleted and changed
+                keys from the event relation databag.
+        """
+        return diff(event, self.local_unit)
+
+    @property
+    def relations(self) -> List[Relation]:
+        """The list of Relation instances associated with this relation_name."""
+        return [
+            relation
+            for relation in self.charm.model.relations[self.relation_name]
+            if self._is_relation_active(relation)
+        ]
+
+    @staticmethod
+    def _is_relation_active(relation: Relation):
+        try:
+            _ = repr(relation.data)
+            return True
+        except RuntimeError:
+            return False
+
+    @staticmethod
+    def _is_resource_created_for_relation(relation: Relation):
+        return (
+            "username" in relation.data[relation.app] and "password" in relation.data[relation.app]
+        )
+
+    def is_resource_created(self, relation_id: Optional[int] = None) -> bool:
+        """Check if the resource has been created.
+
+        This function can be used to check if the Provider answered with data in the charm code
+        when outside an event callback.
+
+        Args:
+            relation_id (int, optional): When provided the check is done only for the relation id
+                provided, otherwise the check is done for all relations
+
+        Returns:
+            True or False
+
+        Raises:
+            IndexError: If relation_id is provided but that relation does not exist
+        """
+        if relation_id is not None:
+            try:
+                relation = [relation for relation in self.relations if relation.id == relation_id][
+                    0
+                ]
+                return self._is_resource_created_for_relation(relation)
+            except IndexError:
+                raise IndexError(f"relation id {relation_id} cannot be accessed")
+        else:
+            return (
+                all(
+                    [
+                        self._is_resource_created_for_relation(relation)
+                        for relation in self.relations
+                    ]
+                )
+                if self.relations
+                else False
+            )
+
+
+# General events
+
+
+class ExtraRoleEvent(RelationEvent):
+    """Base class for data events."""
+
+    @property
+    def extra_user_roles(self) -> Optional[str]:
+        """Returns the extra user roles that were requested."""
+        return self.relation.data[self.relation.app].get("extra-user-roles")
+
+
+class AuthenticationEvent(RelationEvent):
+    """Base class for authentication fields for events."""
+
+    @property
+    def username(self) -> Optional[str]:
+        """Returns the created username."""
+        return self.relation.data[self.relation.app].get("username")
+
+    @property
+    def password(self) -> Optional[str]:
+        """Returns the password for the created user."""
+        return self.relation.data[self.relation.app].get("password")
+
+    @property
+    def tls(self) -> Optional[str]:
+        """Returns whether TLS is configured."""
+        return self.relation.data[self.relation.app].get("tls")
+
+    @property
+    def tls_ca(self) -> Optional[str]:
+        """Returns TLS CA."""
+        return self.relation.data[self.relation.app].get("tls-ca")
+
+
+# Database related events and fields
+
+
+class DatabaseProvidesEvent(RelationEvent):
+    """Base class for database events."""
+
+    @property
+    def database(self) -> Optional[str]:
+        """Returns the database that was requested."""
+        return self.relation.data[self.relation.app].get("database")
+
+
+class DatabaseRequestedEvent(DatabaseProvidesEvent, ExtraRoleEvent):
+    """Event emitted when a new database is requested for use on this relation."""
+
+
+class DatabaseProvidesEvents(CharmEvents):
+    """Database events.
+
+    This class defines the events that the database can emit.
+    """
+
+    database_requested = EventSource(DatabaseRequestedEvent)
+
+
+class DatabaseRequiresEvent(RelationEvent):
+    """Base class for database events."""
+
+    @property
+    def endpoints(self) -> Optional[str]:
+        """Returns a comma separated list of read/write endpoints."""
+        return self.relation.data[self.relation.app].get("endpoints")
+
+    @property
+    def read_only_endpoints(self) -> Optional[str]:
+        """Returns a comma separated list of read only endpoints."""
+        return self.relation.data[self.relation.app].get("read-only-endpoints")
+
+    @property
+    def replset(self) -> Optional[str]:
+        """Returns the replicaset name.
+
+        MongoDB only.
+        """
+        return self.relation.data[self.relation.app].get("replset")
+
+    @property
+    def uris(self) -> Optional[str]:
+        """Returns the connection URIs.
+
+        MongoDB, Redis, OpenSearch.
+        """
+        return self.relation.data[self.relation.app].get("uris")
+
+    @property
+    def version(self) -> Optional[str]:
+        """Returns the version of the database.
+
+        Version as informed by the database daemon.
+        """
+        return self.relation.data[self.relation.app].get("version")
+
+
+class DatabaseCreatedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when a new database is created for use on this relation."""
+
+
+class DatabaseEndpointsChangedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when the read/write endpoints are changed."""
+
+
+class DatabaseReadOnlyEndpointsChangedEvent(AuthenticationEvent, DatabaseRequiresEvent):
+    """Event emitted when the read only endpoints are changed."""
+
+
+class DatabaseRequiresEvents(CharmEvents):
+    """Database events.
+
+    This class defines the events that the database can emit.
+    """
+
+    database_created = EventSource(DatabaseCreatedEvent)
+    endpoints_changed = EventSource(DatabaseEndpointsChangedEvent)
+    read_only_endpoints_changed = EventSource(DatabaseReadOnlyEndpointsChangedEvent)
+
+
+# Database Provider and Requires
+
+
+class DatabaseProvides(DataProvides):
+    """Provider-side of the database relations."""
+
+    on = DatabaseProvidesEvents()
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Only the leader should handle this event.
+        if not self.local_unit.is_leader():
+            return
+
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Emit a database requested event if the setup key (database name and optional
+        # extra user roles) was added to the relation databag by the application.
+        if "database" in diff.added:
+            self.on.database_requested.emit(event.relation, app=event.app, unit=event.unit)
+
+    def set_endpoints(self, relation_id: int, connection_strings: str) -> None:
+        """Set database primary connections.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            connection_strings: database hosts and ports comma separated list.
+        """
+        self._update_relation_data(relation_id, {"endpoints": connection_strings})
+
+    def set_read_only_endpoints(self, relation_id: int, connection_strings: str) -> None:
+        """Set database replicas connection strings.
+
+        This function writes in the application data bag, therefore,
+        only the leader unit can call it.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            connection_strings: database hosts and ports comma separated list.
+        """
+        self._update_relation_data(relation_id, {"read-only-endpoints": connection_strings})
+
+    def set_replset(self, relation_id: int, replset: str) -> None:
+        """Set replica set name in the application relation databag.
+
+        MongoDB only.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            replset: replica set name.
+        """
+        self._update_relation_data(relation_id, {"replset": replset})
+
+    def set_uris(self, relation_id: int, uris: str) -> None:
+        """Set the database connection URIs in the application relation databag.
+
+        MongoDB, Redis, and OpenSearch only.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            uris: connection URIs.
+        """
+        self._update_relation_data(relation_id, {"uris": uris})
+
+    def set_version(self, relation_id: int, version: str) -> None:
+        """Set the database version in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            version: database version.
+        """
+        self._update_relation_data(relation_id, {"version": version})
+
+
+class DatabaseRequires(DataRequires):
+    """Requires-side of the database relation."""
+
+    on = DatabaseRequiresEvents()
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        database_name: str,
+        extra_user_roles: str = None,
+        relations_aliases: List[str] = None,
+    ):
+        """Manager of database client relations."""
+        super().__init__(charm, relation_name, extra_user_roles)
+        self.database = database_name
+        self.relations_aliases = relations_aliases
+
+        # Define custom event names for each alias.
+        if relations_aliases:
+            # Ensure the number of aliases does not exceed the maximum
+            # of connections allowed in the specific relation.
+            relation_connection_limit = self.charm.meta.requires[relation_name].limit
+            if len(relations_aliases) != relation_connection_limit:
+                raise ValueError(
+                    f"The number of aliases must match the maximum number of connections allowed in the relation. "
+                    f"Expected {relation_connection_limit}, got {len(relations_aliases)}"
+                )
+
+            for relation_alias in relations_aliases:
+                self.on.define_event(f"{relation_alias}_database_created", DatabaseCreatedEvent)
+                self.on.define_event(
+                    f"{relation_alias}_endpoints_changed", DatabaseEndpointsChangedEvent
+                )
+                self.on.define_event(
+                    f"{relation_alias}_read_only_endpoints_changed",
+                    DatabaseReadOnlyEndpointsChangedEvent,
+                )
+
+    def _assign_relation_alias(self, relation_id: int) -> None:
+        """Assigns an alias to a relation.
+
+        This function writes in the unit data bag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+        """
+        # If no aliases were provided, return immediately.
+        if not self.relations_aliases:
+            return
+
+        # Return if an alias was already assigned to this relation
+        # (like when there are more than one unit joining the relation).
+        if (
+            self.charm.model.get_relation(self.relation_name, relation_id)
+            .data[self.local_unit]
+            .get("alias")
+        ):
+            return
+
+        # Retrieve the available aliases (the ones that weren't assigned to any relation).
+        available_aliases = self.relations_aliases[:]
+        for relation in self.charm.model.relations[self.relation_name]:
+            alias = relation.data[self.local_unit].get("alias")
+            if alias:
+                logger.debug("Alias %s was already assigned to relation %d", alias, relation.id)
+                available_aliases.remove(alias)
+
+        # Set the alias in the unit relation databag of the specific relation.
+        relation = self.charm.model.get_relation(self.relation_name, relation_id)
+        relation.data[self.local_unit].update({"alias": available_aliases[0]})
+
+    def _emit_aliased_event(self, event: RelationChangedEvent, event_name: str) -> None:
+        """Emit an aliased event to a particular relation if it has an alias.
+
+        Args:
+            event: the relation changed event that was received.
+            event_name: the name of the event to emit.
+        """
+        alias = self._get_relation_alias(event.relation.id)
+        if alias:
+            getattr(self.on, f"{alias}_{event_name}").emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+    def _get_relation_alias(self, relation_id: int) -> Optional[str]:
+        """Returns the relation alias.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+
+        Returns:
+            the relation alias or None if the relation was not found.
+        """
+        for relation in self.charm.model.relations[self.relation_name]:
+            if relation.id == relation_id:
+                return relation.data[self.local_unit].get("alias")
+        return None
+
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the database relation."""
+        # If relations aliases were provided, assign one to the relation.
+        self._assign_relation_alias(event.relation.id)
+
+        # Sets both database and extra user roles in the relation
+        # if the roles are provided. Otherwise, sets only the database.
+        if self.extra_user_roles:
+            self._update_relation_data(
+                event.relation.id,
+                {
+                    "database": self.database,
+                    "extra-user-roles": self.extra_user_roles,
+                },
+            )
+        else:
+            self._update_relation_data(event.relation.id, {"database": self.database})
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the database relation has changed."""
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Check if the database is created
+        # (the database charm shared the credentials).
+        if "username" in diff.added and "password" in diff.added:
+            # Emit the default event (the one without an alias).
+            logger.info("database created at %s", datetime.now())
+            self.on.database_created.emit(event.relation, app=event.app, unit=event.unit)
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "database_created")
+
+            # To avoid unnecessary application restarts do not trigger
+            # “endpoints_changed“ event if “database_created“ is triggered.
+            return
+
+        # Emit an endpoints changed event if the database
+        # added or changed this info in the relation databag.
+        if "endpoints" in diff.added or "endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("endpoints changed on %s", datetime.now())
+            self.on.endpoints_changed.emit(event.relation, app=event.app, unit=event.unit)
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "endpoints_changed")
+
+            # To avoid unnecessary application restarts do not trigger
+            # “read_only_endpoints_changed“ event if “endpoints_changed“ is triggered.
+            return
+
+        # Emit a read only endpoints changed event if the database
+        # added or changed this info in the relation databag.
+        if "read-only-endpoints" in diff.added or "read-only-endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("read-only-endpoints changed on %s", datetime.now())
+            self.on.read_only_endpoints_changed.emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+            # Emit the aliased event (if any).
+            self._emit_aliased_event(event, "read_only_endpoints_changed")
+
+
+# Kafka related events
+
+
+class KafkaProvidesEvent(RelationEvent):
+    """Base class for Kafka events."""
+
+    @property
+    def topic(self) -> Optional[str]:
+        """Returns the topic that was requested."""
+        return self.relation.data[self.relation.app].get("topic")
+
+
+class TopicRequestedEvent(KafkaProvidesEvent, ExtraRoleEvent):
+    """Event emitted when a new topic is requested for use on this relation."""
+
+
+class KafkaProvidesEvents(CharmEvents):
+    """Kafka events.
+
+    This class defines the events that the Kafka can emit.
+    """
+
+    topic_requested = EventSource(TopicRequestedEvent)
+
+
+class KafkaRequiresEvent(RelationEvent):
+    """Base class for Kafka events."""
+
+    @property
+    def bootstrap_server(self) -> Optional[str]:
+        """Returns a a comma-seperated list of broker uris."""
+        return self.relation.data[self.relation.app].get("endpoints")
+
+    @property
+    def consumer_group_prefix(self) -> Optional[str]:
+        """Returns the consumer-group-prefix."""
+        return self.relation.data[self.relation.app].get("consumer-group-prefix")
+
+    @property
+    def zookeeper_uris(self) -> Optional[str]:
+        """Returns a comma separated list of Zookeeper uris."""
+        return self.relation.data[self.relation.app].get("zookeeper-uris")
+
+
+class TopicCreatedEvent(AuthenticationEvent, KafkaRequiresEvent):
+    """Event emitted when a new topic is created for use on this relation."""
+
+
+class BootstrapServerChangedEvent(AuthenticationEvent, KafkaRequiresEvent):
+    """Event emitted when the bootstrap server is changed."""
+
+
+class KafkaRequiresEvents(CharmEvents):
+    """Kafka events.
+
+    This class defines the events that the Kafka can emit.
+    """
+
+    topic_created = EventSource(TopicCreatedEvent)
+    bootstrap_server_changed = EventSource(BootstrapServerChangedEvent)
+
+
+# Kafka Provides and Requires
+
+
+class KafkaProvides(DataProvides):
+    """Provider-side of the Kafka relation."""
+
+    on = KafkaProvidesEvents()
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        super().__init__(charm, relation_name)
+
+    def _on_relation_changed(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Only the leader should handle this event.
+        if not self.local_unit.is_leader():
+            return
+
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Emit a topic requested event if the setup key (topic name and optional
+        # extra user roles) was added to the relation databag by the application.
+        if "topic" in diff.added:
+            self.on.topic_requested.emit(event.relation, app=event.app, unit=event.unit)
+
+    def set_bootstrap_server(self, relation_id: int, bootstrap_server: str) -> None:
+        """Set the bootstrap server in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            bootstrap_server: the bootstrap server address.
+        """
+        self._update_relation_data(relation_id, {"endpoints": bootstrap_server})
+
+    def set_consumer_group_prefix(self, relation_id: int, consumer_group_prefix: str) -> None:
+        """Set the consumer group prefix in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            consumer_group_prefix: the consumer group prefix string.
+        """
+        self._update_relation_data(relation_id, {"consumer-group-prefix": consumer_group_prefix})
+
+    def set_zookeeper_uris(self, relation_id: int, zookeeper_uris: str) -> None:
+        """Set the zookeeper uris in the application relation databag.
+
+        Args:
+            relation_id: the identifier for a particular relation.
+            zookeeper_uris: comma-seperated list of ZooKeeper server uris.
+        """
+        self._update_relation_data(relation_id, {"zookeeper-uris": zookeeper_uris})
+
+
+class KafkaRequires(DataRequires):
+    """Requires-side of the Kafka relation."""
+
+    on = KafkaRequiresEvents()
+
+    def __init__(self, charm, relation_name: str, topic: str, extra_user_roles: str = None):
+        """Manager of Kafka client relations."""
+        # super().__init__(charm, relation_name)
+        super().__init__(charm, relation_name, extra_user_roles)
+        self.charm = charm
+        self.topic = topic
+
+    def _on_relation_joined_event(self, event: RelationJoinedEvent) -> None:
+        """Event emitted when the application joins the Kafka relation."""
+        # Sets both topic and extra user roles in the relation
+        # if the roles are provided. Otherwise, sets only the topic.
+        self._update_relation_data(
+            event.relation.id,
+            {
+                "topic": self.topic,
+                "extra-user-roles": self.extra_user_roles,
+            }
+            if self.extra_user_roles is not None
+            else {"topic": self.topic},
+        )
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the Kafka relation has changed."""
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Check if the topic is created
+        # (the Kafka charm shared the credentials).
+        if "username" in diff.added and "password" in diff.added:
+            # Emit the default event (the one without an alias).
+            logger.info("topic created at %s", datetime.now())
+            self.on.topic_created.emit(event.relation, app=event.app, unit=event.unit)
+
+            # To avoid unnecessary application restarts do not trigger
+            # “endpoints_changed“ event if “topic_created“ is triggered.
+            return
+
+        # Emit an endpoints (bootstap-server) changed event if the Kakfa endpoints
+        # added or changed this info in the relation databag.
+        if "endpoints" in diff.added or "endpoints" in diff.changed:
+            # Emit the default event (the one without an alias).
+            logger.info("endpoints changed on %s", datetime.now())
+            self.on.bootstrap_server_changed.emit(
+                event.relation, app=event.app, unit=event.unit
+            )  # here check if this is the right design
+            return
diff --git a/installers/charm/osm-ro/metadata.yaml b/installers/charm/osm-ro/metadata.yaml
index 4336ded..a94036a 100644
--- a/installers/charm/osm-ro/metadata.yaml
+++ b/installers/charm/osm-ro/metadata.yaml
@@ -58,7 +58,7 @@
     interface: kafka
     limit: 1
   mongodb:
-    interface: mongodb
+    interface: mongodb_client
     limit: 1
 
 provides:
diff --git a/installers/charm/osm-ro/src/charm.py b/installers/charm/osm-ro/src/charm.py
index e112d4c..84c0ee3 100755
--- a/installers/charm/osm-ro/src/charm.py
+++ b/installers/charm/osm-ro/src/charm.py
@@ -31,6 +31,7 @@
 import logging
 from typing import Any, Dict
 
+from charms.data_platform_libs.v0.data_interfaces import DatabaseRequires
 from charms.kafka_k8s.v0.kafka import KafkaEvents, KafkaRequires
 from charms.observability_libs.v1.kubernetes_service_patch import KubernetesServicePatch
 from charms.osm_libs.v0.utils import (
@@ -47,8 +48,6 @@
 from ops.main import main
 from ops.model import ActiveStatus, Container
 
-from legacy_interfaces import MongoClient
-
 ro_host_paths = {
     "NG-RO": "/usr/lib/python3/dist-packages/osm_ng_ro",
     "RO-plugin": "/usr/lib/python3/dist-packages/osm_ro_plugin",
@@ -101,7 +100,7 @@
         super().__init__(*args)
         self._stored.set_default(certificates=set())
         self.kafka = KafkaRequires(self)
-        self.mongodb_client = MongoClient(self, "mongodb")
+        self.mongodb_client = DatabaseRequires(self, "mongodb", database_name="osm")
         self._observe_charm_events()
         self._patch_k8s_service()
         self.ro = RoProvides(self)
@@ -197,7 +196,7 @@
             # Relation events
             self.on.kafka_available: self._on_config_changed,
             self.on["kafka"].relation_broken: self._on_required_relation_broken,
-            self.on["mongodb"].relation_changed: self._on_config_changed,
+            self.mongodb_client.on.database_created: self._on_config_changed,
             self.on["mongodb"].relation_broken: self._on_required_relation_broken,
             self.on.ro_relation_joined: self._update_ro_relation,
             # Action events
@@ -207,6 +206,12 @@
         for event, handler in event_handler_mapping.items():
             self.framework.observe(event, handler)
 
+    def _is_database_available(self) -> bool:
+        try:
+            return self.mongodb_client.is_resource_created()
+        except KeyError:
+            return False
+
     def _validate_config(self) -> None:
         """Validate charm configuration.
 
@@ -241,7 +246,7 @@
 
         if not self.kafka.host or not self.kafka.port:
             missing_relations.append("kafka")
-        if self.mongodb_client.is_missing_data_in_unit():
+        if not self._is_database_available():
             missing_relations.append("mongodb")
 
         if missing_relations:
@@ -310,13 +315,13 @@
                         "OSMRO_MESSAGE_DRIVER": "kafka",
                         # Database configuration
                         "OSMRO_DATABASE_DRIVER": "mongo",
-                        "OSMRO_DATABASE_URI": self.mongodb_client.connection_string,
+                        "OSMRO_DATABASE_URI": self._get_mongodb_uri(),
                         "OSMRO_DATABASE_COMMONKEY": self.config["database-commonkey"],
                         # Storage configuration
                         "OSMRO_STORAGE_DRIVER": "mongo",
                         "OSMRO_STORAGE_PATH": "/app/storage",
                         "OSMRO_STORAGE_COLLECTION": "files",
-                        "OSMRO_STORAGE_URI": self.mongodb_client.connection_string,
+                        "OSMRO_STORAGE_URI": self._get_mongodb_uri(),
                         "OSMRO_PERIOD_REFRESH_ACTIVE": self.config.get("period_refresh_active")
                         or 60,
                     },
@@ -324,6 +329,9 @@
             },
         }
 
+    def _get_mongodb_uri(self):
+        return list(self.mongodb_client.fetch_relation_data().values())[0]["uris"]
+
 
 if __name__ == "__main__":  # pragma: no cover
     main(OsmRoCharm)
diff --git a/installers/charm/osm-ro/tests/integration/test_charm.py b/installers/charm/osm-ro/tests/integration/test_charm.py
index c39c47a..38dc40f 100644
--- a/installers/charm/osm-ro/tests/integration/test_charm.py
+++ b/installers/charm/osm-ro/tests/integration/test_charm.py
@@ -51,7 +51,7 @@
         ops_test.model.deploy(charm, resources=resources, application_name=RO_APP),
         ops_test.model.deploy(ZOOKEEPER_CHARM, application_name=ZOOKEEPER_APP, channel="stable"),
         ops_test.model.deploy(KAFKA_CHARM, application_name=KAFKA_APP, channel="stable"),
-        ops_test.model.deploy(MONGO_DB_CHARM, application_name=MONGO_DB_APP, channel="stable"),
+        ops_test.model.deploy(MONGO_DB_CHARM, application_name=MONGO_DB_APP, channel="edge"),
     )
 
     async with ops_test.fast_forward():
diff --git a/installers/charm/osm-ro/tests/unit/test_charm.py b/installers/charm/osm-ro/tests/unit/test_charm.py
index 05206d0..d0353ab 100644
--- a/installers/charm/osm-ro/tests/unit/test_charm.py
+++ b/installers/charm/osm-ro/tests/unit/test_charm.py
@@ -37,6 +37,7 @@
     mocker.patch("charm.KubernetesServicePatch", lambda x, y: None)
     harness = Harness(OsmRoCharm)
     harness.begin()
+    harness.container_pebble_ready(container_name)
     yield harness
     harness.cleanup()
 
@@ -88,7 +89,9 @@
     relation_id = harness.add_relation("mongodb", "mongodb")
     harness.add_relation_unit(relation_id, "mongodb/0")
     harness.update_relation_data(
-        relation_id, "mongodb/0", {"connection_string": "mongodb://:1234"}
+        relation_id,
+        "mongodb",
+        {"uris": "mongodb://:1234", "username": "user", "password": "password"},
     )
     relation_ids.append(relation_id)
     # Add kafka relation
diff --git a/installers/charm/osm-ro/tox.ini b/installers/charm/osm-ro/tox.ini
index 0083afe..c974509 100644
--- a/installers/charm/osm-ro/tox.ini
+++ b/installers/charm/osm-ro/tox.ini
@@ -88,7 +88,7 @@
 description = Run integration tests
 deps =
     pytest
-    juju
+    juju<3
     pytest-operator
     -r{toxinidir}/requirements.txt
 commands =