diff --git a/.github/workflows/integrate.yaml b/.github/workflows/integrate.yaml
index e76d4c3..a308871 100644
--- a/.github/workflows/integrate.yaml
+++ b/.github/workflows/integrate.yaml
@@ -72,6 +72,7 @@ jobs:
         uses: charmed-kubernetes/actions-operator@1.1.0
         with:
           juju-channel: 3.1/stable
+          lxd-channel: 5.20/stable
           provider: microk8s
           channel: 1.28-strict/stable
           microk8s-addons: "dns storage rbac metallb:10.64.140.43-10.64.140.49"
diff --git a/lib/charms/data_platform_libs/v0/data_interfaces.py b/lib/charms/data_platform_libs/v0/data_interfaces.py
index a266a74..4a2ee5a 100644
--- a/lib/charms/data_platform_libs/v0/data_interfaces.py
+++ b/lib/charms/data_platform_libs/v0/data_interfaces.py
@@ -295,12 +295,23 @@ def _on_topic_requested(self, event: TopicRequestedEvent):
 import json
 import logging
 from abc import ABC, abstractmethod
-from collections import namedtuple
+from collections import UserDict, namedtuple
 from datetime import datetime
 from enum import Enum
-from typing import Callable, Dict, List, Optional, Set, Tuple, Union
+from typing import (
+    Callable,
+    Dict,
+    ItemsView,
+    KeysView,
+    List,
+    Optional,
+    Set,
+    Tuple,
+    Union,
+    ValuesView,
+)
 
-from ops import JujuVersion, Secret, SecretInfo, SecretNotFoundError
+from ops import JujuVersion, Model, Secret, SecretInfo, SecretNotFoundError
 from ops.charm import (
     CharmBase,
     CharmEvents,
@@ -320,7 +331,7 @@ def _on_topic_requested(self, event: TopicRequestedEvent):
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 28
+LIBPATCH = 33
 
 PYDEPS = ["ops>=2.0.0"]
 
@@ -337,21 +348,46 @@ def _on_topic_requested(self, event: TopicRequestedEvent):
 
 PROV_SECRET_PREFIX = "secret-"
 REQ_SECRET_FIELDS = "requested-secrets"
+GROUP_MAPPING_FIELD = "secret_group_mapping"
+GROUP_SEPARATOR = "@"
+
+
+class SecretGroup(str):
+    """Secret groups specific type."""
+
+
+class SecretGroupsAggregate(str):
+    """Secret groups with option to extend with additional constants."""
+
+    def __init__(self):
+        self.USER = SecretGroup("user")
+        self.TLS = SecretGroup("tls")
+        self.EXTRA = SecretGroup("extra")
+
+    def __setattr__(self, name, value):
+        """Setting internal constants."""
+        if name in self.__dict__:
+            raise RuntimeError("Can't set constant!")
+        else:
+            super().__setattr__(name, SecretGroup(value))
 
+    def groups(self) -> list:
+        """Return the list of stored SecretGroups."""
+        return list(self.__dict__.values())
 
-class SecretGroup(Enum):
-    """Secret groups as constants."""
+    def get_group(self, group: str) -> Optional[SecretGroup]:
+        """If the input str translates to a group name, return that."""
+        return SecretGroup(group) if group in self.groups() else None
 
-    USER = "user"
-    TLS = "tls"
-    EXTRA = "extra"
+
+SECRET_GROUPS = SecretGroupsAggregate()
 
 
 class DataInterfacesError(Exception):
     """Common ancestor for DataInterfaces related exceptions."""
 
 
-class SecretError(Exception):
+class SecretError(DataInterfacesError):
     """Common ancestor for Secrets related exceptions."""
 
 
@@ -367,6 +403,10 @@ class SecretsIllegalUpdateError(SecretError):
     """Secrets aren't yet available for Juju version used."""
 
 
+class IllegalOperationError(DataInterfacesError):
+    """To be used when an operation is not allowed to be performed."""
+
+
 def get_encoded_dict(
     relation: Relation, member: Union[Unit, Application], field: str
 ) -> Optional[Dict[str, str]]:
@@ -397,7 +437,7 @@ def set_encoded_field(
     relation.data[member].update({field: json.dumps(value)})
 
 
-def diff(event: RelationChangedEvent, bucket: Union[Unit, Application]) -> Diff:
+def diff(event: RelationChangedEvent, bucket: Optional[Union[Unit, Application]]) -> Diff:
     """Retrieves the diff of the data in the relation changed databag.
 
     Args:
@@ -409,6 +449,9 @@ def diff(event: RelationChangedEvent, bucket: Union[Unit, Application]) -> Diff:
             keys from the event relation databag.
     """
     # Retrieve the old data from the data key in the application relation databag.
+    if not bucket:
+        return Diff([], [], [])
+
     old_data = get_encoded_dict(event.relation, bucket, "data")
 
     if not old_data:
@@ -464,6 +507,34 @@ def wrapper(self, *args, **kwargs):
     return wrapper
 
 
+def dynamic_secrets_only(f):
+    """Decorator to ensure that certain operations would be only executed when NO static secrets are defined."""
+
+    def wrapper(self, *args, **kwargs):
+        if self.static_secret_fields:
+            raise IllegalOperationError(
+                "Unsafe usage of statically and dynamically defined secrets, aborting."
+            )
+        return f(self, *args, **kwargs)
+
+    return wrapper
+
+
+def either_static_or_dynamic_secrets(f):
+    """Decorator to ensure that static and dynamic secrets won't be used in parallel."""
+
+    def wrapper(self, *args, **kwargs):
+        if self.static_secret_fields and set(self.current_secret_fields) - set(
+            self.static_secret_fields
+        ):
+            raise IllegalOperationError(
+                "Unsafe usage of statically and dynamically defined secrets, aborting."
+            )
+        return f(self, *args, **kwargs)
+
+    return wrapper
+
+
 class Scope(Enum):
     """Peer relations scope."""
 
@@ -471,6 +542,11 @@ class Scope(Enum):
     UNIT = "unit"
 
 
+################################################################################
+# Secrets internal caching
+################################################################################
+
+
 class CachedSecret:
     """Locally cache a secret.
 
@@ -479,7 +555,7 @@ class CachedSecret:
 
     def __init__(
         self,
-        charm: CharmBase,
+        model: Model,
         component: Union[Application, Unit],
         label: str,
         secret_uri: Optional[str] = None,
@@ -488,7 +564,7 @@ def __init__(
         self._secret_content = {}
         self._secret_uri = secret_uri
         self.label = label
-        self.charm = charm
+        self._model = model
         self.component = component
 
     def add_secret(self, content: Dict[str, str], relation: Relation) -> Secret:
@@ -499,7 +575,7 @@ def add_secret(self, content: Dict[str, str], relation: Relation) -> Secret:
             )
 
         secret = self.component.add_secret(content, label=self.label)
-        if relation.app != self.charm.app:
+        if relation.app != self._model.app:
             # If it's not a peer relation, grant is to be applied
             secret.grant(relation)
         self._secret_uri = secret.id
@@ -513,10 +589,10 @@ def meta(self) -> Optional[Secret]:
             if not (self._secret_uri or self.label):
                 return
             try:
-                self._secret_meta = self.charm.model.get_secret(label=self.label)
+                self._secret_meta = self._model.get_secret(label=self.label)
             except SecretNotFoundError:
                 if self._secret_uri:
-                    self._secret_meta = self.charm.model.get_secret(
+                    self._secret_meta = self._model.get_secret(
                         id=self._secret_uri, label=self.label
                     )
         return self._secret_meta
@@ -558,19 +634,31 @@ def get_info(self) -> Optional[SecretInfo]:
         if self.meta:
             return self.meta.get_info()
 
+    def remove(self) -> None:
+        """Remove secret."""
+        if not self.meta:
+            raise SecretsUnavailableError("Non-existent secret was attempted to be removed.")
+        try:
+            self.meta.remove_all_revisions()
+        except SecretNotFoundError:
+            pass
+        self._secret_content = {}
+        self._secret_meta = None
+        self._secret_uri = None
+
 
 class SecretCache:
     """A data structure storing CachedSecret objects."""
 
-    def __init__(self, charm: CharmBase, component: Union[Application, Unit]):
-        self.charm = charm
+    def __init__(self, model: Model, component: Union[Application, Unit]):
+        self._model = model
         self.component = component
         self._secrets: Dict[str, CachedSecret] = {}
 
     def get(self, label: str, uri: Optional[str] = None) -> Optional[CachedSecret]:
         """Getting a secret from Juju Secret store or cache."""
         if not self._secrets.get(label):
-            secret = CachedSecret(self.charm, self.component, label, uri)
+            secret = CachedSecret(self._model, self.component, label, uri)
             if secret.meta:
                 self._secrets[label] = secret
         return self._secrets.get(label)
@@ -580,49 +668,158 @@ def add(self, label: str, content: Dict[str, str], relation: Relation) -> Cached
         if self._secrets.get(label):
             raise SecretAlreadyExistsError(f"Secret {label} already exists")
 
-        secret = CachedSecret(self.charm, self.component, label)
+        secret = CachedSecret(self._model, self.component, label)
         secret.add_secret(content, relation)
         self._secrets[label] = secret
         return self._secrets[label]
 
+    def remove(self, label: str) -> None:
+        """Remove a secret from the cache."""
+        if secret := self.get(label):
+            secret.remove()
+            self._secrets.pop(label)
+        else:
+            logging.error("Non-existing Juju Secret was attempted to be removed %s", label)
+
+
+################################################################################
+# Relation Data base/abstract ancestors (i.e. parent classes)
+################################################################################
+
+
+# Base Data
+
+
+class DataDict(UserDict):
+    """Python Standard Library 'dict' - like representation of Relation Data."""
+
+    def __init__(self, relation_data: "Data", relation_id: int):
+        self.relation_data = relation_data
+        self.relation_id = relation_id
+
+    @property
+    def data(self) -> Dict[str, str]:
+        """Return the full content of the Abstract Relation Data dictionary."""
+        result = self.relation_data.fetch_my_relation_data([self.relation_id])
+        try:
+            result_remote = self.relation_data.fetch_relation_data([self.relation_id])
+        except NotImplementedError:
+            result_remote = {self.relation_id: {}}
+        if result:
+            result_remote[self.relation_id].update(result[self.relation_id])
+        return result_remote.get(self.relation_id, {})
+
+    def __setitem__(self, key: str, item: str) -> None:
+        """Set an item of the Abstract Relation Data dictionary."""
+        self.relation_data.update_relation_data(self.relation_id, {key: item})
+
+    def __getitem__(self, key: str) -> str:
+        """Get an item of the Abstract Relation Data dictionary."""
+        result = None
+        if not (result := self.relation_data.fetch_my_relation_field(self.relation_id, key)):
+            try:
+                result = self.relation_data.fetch_relation_field(self.relation_id, key)
+            except NotImplementedError:
+                pass
+        if not result:
+            raise KeyError
+        return result
+
+    def __eq__(self, d: dict) -> bool:
+        """Equality."""
+        return self.data == d
+
+    def __repr__(self) -> str:
+        """String representation Abstract Relation Data dictionary."""
+        return repr(self.data)
+
+    def __len__(self) -> int:
+        """Length of the Abstract Relation Data dictionary."""
+        return len(self.data)
+
+    def __delitem__(self, key: str) -> None:
+        """Delete an item of the Abstract Relation Data dictionary."""
+        self.relation_data.delete_relation_data(self.relation_id, [key])
+
+    def has_key(self, key: str) -> bool:
+        """Does the key exist in the Abstract Relation Data dictionary?"""
+        return key in self.data
+
+    def update(self, items: Dict[str, str]):
+        """Update the Abstract Relation Data dictionary."""
+        self.relation_data.update_relation_data(self.relation_id, items)
+
+    def keys(self) -> KeysView[str]:
+        """Keys of the Abstract Relation Data dictionary."""
+        return self.data.keys()
+
+    def values(self) -> ValuesView[str]:
+        """Values of the Abstract Relation Data dictionary."""
+        return self.data.values()
+
+    def items(self) -> ItemsView[str, str]:
+        """Items of the Abstract Relation Data dictionary."""
+        return self.data.items()
+
+    def pop(self, item: str) -> str:
+        """Pop an item of the Abstract Relation Data dictionary."""
+        result = self.relation_data.fetch_my_relation_field(self.relation_id, item)
+        if not result:
+            raise KeyError(f"Item {item} doesn't exist.")
+        self.relation_data.delete_relation_data(self.relation_id, [item])
+        return result
+
+    def __contains__(self, item: str) -> bool:
+        """Does the Abstract Relation Data dictionary contain item?"""
+        return item in self.data.values()
 
-# Base DataRelation
+    def __iter__(self):
+        """Iterate through the Abstract Relation Data dictionary."""
+        return iter(self.data)
+
+    def get(self, key: str, default: Optional[str] = None) -> Optional[str]:
+        """Safely get an item of the Abstract Relation Data dictionary."""
+        try:
+            if result := self[key]:
+                return result
+        except KeyError:
+            return default
 
 
-class DataRelation(Object, ABC):
+class Data(ABC):
     """Base relation data mainpulation (abstract) class."""
 
     SCOPE = Scope.APP
 
     # Local map to associate mappings with secrets potentially as a group
     SECRET_LABEL_MAP = {
-        "username": SecretGroup.USER,
-        "password": SecretGroup.USER,
-        "uris": SecretGroup.USER,
-        "tls": SecretGroup.TLS,
-        "tls-ca": SecretGroup.TLS,
+        "username": SECRET_GROUPS.USER,
+        "password": SECRET_GROUPS.USER,
+        "uris": SECRET_GROUPS.USER,
+        "tls": SECRET_GROUPS.TLS,
+        "tls-ca": SECRET_GROUPS.TLS,
     }
 
-    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
+    def __init__(
+        self,
+        model: Model,
+        relation_name: str,
+    ) -> None:
+        self._model = model
+        self.local_app = self._model.app
+        self.local_unit = self._model.unit
         self.relation_name = relation_name
-        self.framework.observe(
-            charm.on[relation_name].relation_changed,
-            self._on_relation_changed_event,
-        )
         self._jujuversion = None
         self.component = self.local_app if self.SCOPE == Scope.APP else self.local_unit
-        self.secrets = SecretCache(self.charm, self.component)
+        self.secrets = SecretCache(self._model, self.component)
+        self.data_component = None
 
     @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]
+            for relation in self._model.relations[self.relation_name]
             if self._is_relation_active(relation)
         ]
 
@@ -633,12 +830,12 @@ def secrets_enabled(self):
             self._jujuversion = JujuVersion.from_environ()
         return self._jujuversion.has_secrets
 
-    # Mandatory overrides for internal/helper methods
+    @property
+    def secret_label_map(self):
+        """Exposing secret-label map via a property -- could be overridden in descendants!"""
+        return self.SECRET_LABEL_MAP
 
-    @abstractmethod
-    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
-        """Event emitted when the relation data has changed."""
-        raise NotImplementedError
+    # Mandatory overrides for internal/helper methods
 
     @abstractmethod
     def _get_relation_secret(
@@ -692,11 +889,11 @@ def _generate_secret_label(
         relation_name: str, relation_id: int, group_mapping: SecretGroup
     ) -> str:
         """Generate unique group_mappings for secrets within a relation context."""
-        return f"{relation_name}.{relation_id}.{group_mapping.value}.secret"
+        return f"{relation_name}.{relation_id}.{group_mapping}.secret"
 
     def _generate_secret_field_name(self, group_mapping: SecretGroup) -> str:
         """Generate unique group_mappings for secrets within a relation context."""
-        return f"{PROV_SECRET_PREFIX}{group_mapping.value}"
+        return f"{PROV_SECRET_PREFIX}{group_mapping}"
 
     def _relation_from_secret_label(self, secret_label: str) -> Optional[Relation]:
         """Retrieve the relation that belongs to a secret label."""
@@ -721,8 +918,7 @@ def _relation_from_secret_label(self, secret_label: str) -> Optional[Relation]:
         except ModelError:
             return
 
-    @classmethod
-    def _group_secret_fields(cls, secret_fields: List[str]) -> Dict[SecretGroup, List[str]]:
+    def _group_secret_fields(self, secret_fields: List[str]) -> Dict[SecretGroup, List[str]]:
         """Helper function to arrange secret mappings under their group.
 
         NOTE: All unrecognized items end up in the 'extra' secret bucket.
@@ -730,44 +926,42 @@ def _group_secret_fields(cls, secret_fields: List[str]) -> Dict[SecretGroup, Lis
         """
         secret_fieldnames_grouped = {}
         for key in secret_fields:
-            if group := cls.SECRET_LABEL_MAP.get(key):
+            if group := self.secret_label_map.get(key):
                 secret_fieldnames_grouped.setdefault(group, []).append(key)
             else:
-                secret_fieldnames_grouped.setdefault(SecretGroup.EXTRA, []).append(key)
+                secret_fieldnames_grouped.setdefault(SECRET_GROUPS.EXTRA, []).append(key)
         return secret_fieldnames_grouped
 
     def _get_group_secret_contents(
         self,
         relation: Relation,
         group: SecretGroup,
-        secret_fields: Optional[Union[Set[str], List[str]]] = None,
+        secret_fields: Union[Set[str], List[str]] = [],
     ) -> Dict[str, str]:
         """Helper function to retrieve collective, requested contents of a secret."""
-        if not secret_fields:
-            secret_fields = []
-
         if (secret := self._get_relation_secret(relation.id, group)) and (
             secret_data := secret.get_content()
         ):
-            return {k: v for k, v in secret_data.items() if k in secret_fields}
+            return {
+                k: v for k, v in secret_data.items() if not secret_fields or k in secret_fields
+            }
         return {}
 
-    @classmethod
     def _content_for_secret_group(
-        cls, content: Dict[str, str], secret_fields: Set[str], group_mapping: SecretGroup
+        self, content: Dict[str, str], secret_fields: Set[str], group_mapping: SecretGroup
     ) -> Dict[str, str]:
         """Select <field>: <value> pairs from input, that belong to this particular Secret group."""
-        if group_mapping == SecretGroup.EXTRA:
+        if group_mapping == SECRET_GROUPS.EXTRA:
             return {
                 k: v
                 for k, v in content.items()
-                if k in secret_fields and k not in cls.SECRET_LABEL_MAP.keys()
+                if k in secret_fields and k not in self.secret_label_map.keys()
             }
 
         return {
             k: v
             for k, v in content.items()
-            if k in secret_fields and cls.SECRET_LABEL_MAP.get(k) == group_mapping
+            if k in secret_fields and self.secret_label_map.get(k) == group_mapping
         }
 
     @juju_secrets_only
@@ -796,10 +990,10 @@ def _process_secret_fields(
 
         # If the relation started on a databag, we just stay on the databag
         # (Rolling upgrades may result in a relation starting on databag, getting secrets enabled on-the-fly)
-        # self.local_app is sufficient to check (ignored if Requires, never has secrets -- works if Provides)
+        # self.local_app is sufficient to check (ignored if Requires, never has secrets -- works if Provider)
         fallback_to_databag = (
             req_secret_fields
-            and self.local_unit.is_leader()
+            and (self.local_unit == self._model.unit and self.local_unit.is_leader())
             and set(req_secret_fields) & set(relation.data[self.component])
         )
 
@@ -829,7 +1023,7 @@ def _fetch_relation_data_without_secrets(
 
         Since the Provider's databag is the only one holding secrest, we can apply
         a simplified workflow to read the Require's side's databag.
-        This is used typically when the Provides side wants to read the Requires side's data,
+        This is used typically when the Provider side wants to read the Requires side's data,
         or when the Requires side may want to read its own data.
         """
         if component not in relation.data or not relation.data[component]:
@@ -853,23 +1047,19 @@ def _fetch_relation_data_with_secrets(
 
         This function has internal logic to resolve if a requested field may be "hidden"
         within a Relation Secret, or directly available as a databag field. Typically
-        used to read the Provides side's databag (eigher by the Requires side, or by
-        Provides side itself).
+        used to read the Provider side's databag (eigher by the Requires side, or by
+        Provider side itself).
         """
         result = {}
         normal_fields = []
 
         if not fields:
-            if component not in relation.data or not relation.data[component]:
+            if component not in relation.data:
                 return {}
 
             all_fields = list(relation.data[component].keys())
             normal_fields = [field for field in all_fields if not self._is_secret_field(field)]
-
-            # There must have been secrets there
-            if all_fields != normal_fields and req_secret_fields:
-                # So we assemble the full fields list (without 'secret-<X>' fields)
-                fields = normal_fields + req_secret_fields
+            fields = normal_fields + req_secret_fields if req_secret_fields else normal_fields
 
         if fields:
             result, normal_fields = self._process_secret_fields(
@@ -877,7 +1067,7 @@ def _fetch_relation_data_with_secrets(
             )
 
         # Processing "normal" fields. May include leftover from what we couldn't retrieve as a secret.
-        # (Typically when Juju3 Requires meets Juju2 Provides)
+        # (Typically when Juju3 Requires meets Juju2 Provider)
         if normal_fields:
             result.update(
                 self._fetch_relation_data_without_secrets(component, relation, list(normal_fields))
@@ -915,9 +1105,13 @@ def _delete_relation_data_without_secrets(
     # Public interface methods
     # Handling Relation Fields seamlessly, regardless if in databag or a Juju Secret
 
+    def as_dict(self, relation_id: int) -> UserDict:
+        """Dict behavior representation of the Abstract Data."""
+        return DataDict(self, relation_id)
+
     def get_relation(self, relation_name, relation_id) -> Relation:
         """Safe way of retrieving a relation."""
-        relation = self.charm.model.get_relation(relation_name, relation_id)
+        relation = self._model.get_relation(relation_name, relation_id)
 
         if not relation:
             raise DataInterfacesError(
@@ -1023,14 +1217,22 @@ def delete_relation_data(self, relation_id: int, fields: List[str]) -> None:
         return self._delete_relation_data(relation, fields)
 
 
-# Base DataProvides and DataRequires
+class EventHandlers(Object):
+    """Requires-side of the relation."""
 
+    def __init__(self, charm: CharmBase, relation_data: Data, unique_key: str = ""):
+        """Manager of base client relations."""
+        if not unique_key:
+            unique_key = relation_data.relation_name
+        super().__init__(charm, unique_key)
 
-class DataProvides(DataRelation):
-    """Base provides-side of the data products relation."""
+        self.charm = charm
+        self.relation_data = relation_data
 
-    def __init__(self, charm: CharmBase, relation_name: str) -> None:
-        super().__init__(charm, relation_name)
+        self.framework.observe(
+            charm.on[self.relation_data.relation_name].relation_changed,
+            self._on_relation_changed_event,
+        )
 
     def _diff(self, event: RelationChangedEvent) -> Diff:
         """Retrieves the diff of the data in the relation changed databag.
@@ -1042,7 +1244,27 @@ def _diff(self, event: RelationChangedEvent) -> Diff:
             a Diff instance containing the added, deleted and changed
                 keys from the event relation databag.
         """
-        return diff(event, self.local_app)
+        return diff(event, self.relation_data.data_component)
+
+    @abstractmethod
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation data has changed."""
+        raise NotImplementedError
+
+
+# Base ProviderData and RequiresData
+
+
+class ProviderData(Data):
+    """Base provides-side of the data products relation."""
+
+    def __init__(
+        self,
+        model: Model,
+        relation_name: str,
+    ) -> None:
+        super().__init__(model, relation_name)
+        self.data_component = self.local_app
 
     # Private methods handling secrets
 
@@ -1136,8 +1358,6 @@ def _delete_relation_secret(
                 )
                 return False
 
-        secret.set_content(new_content)
-
         # Remove secret from the relation if it's fully gone
         if not new_content:
             field = self._generate_secret_field_name(group)
@@ -1145,6 +1365,10 @@ def _delete_relation_secret(
                 relation.data[self.component].pop(field)
             except KeyError:
                 pass
+            label = self._generate_secret_label(self.relation_name, relation.id, group)
+            self.secrets.remove(label)
+        else:
+            secret.set_content(new_content)
 
         # Return the content that was removed
         return True
@@ -1163,7 +1387,7 @@ def _get_relation_secret(
         if secret := self.secrets.get(label):
             return secret
 
-        relation = self.charm.model.get_relation(relation_name, relation_id)
+        relation = self._model.get_relation(relation_name, relation_id)
         if not relation:
             return
 
@@ -1174,9 +1398,9 @@ def _get_relation_secret(
     def _fetch_specific_relation_data(
         self, relation: Relation, fields: Optional[List[str]]
     ) -> Dict[str, str]:
-        """Fetching relation data for Provides.
+        """Fetching relation data for Provider.
 
-        NOTE: Since all secret fields are in the Provides side of the databag, we don't need to worry about that
+        NOTE: Since all secret fields are in the Provider side of the databag, we don't need to worry about that
         """
         if not relation.app:
             return {}
@@ -1261,36 +1485,29 @@ def set_tls_ca(self, relation_id: int, tls_ca: str) -> None:
 
     # Public functions -- inherited
 
-    fetch_my_relation_data = leader_only(DataRelation.fetch_my_relation_data)
-    fetch_my_relation_field = leader_only(DataRelation.fetch_my_relation_field)
+    fetch_my_relation_data = leader_only(Data.fetch_my_relation_data)
+    fetch_my_relation_field = leader_only(Data.fetch_my_relation_field)
 
 
-class DataRequires(DataRelation):
-    """Requires-side of the relation."""
+class RequirerData(Data):
+    """Requirer-side of the relation."""
 
     SECRET_FIELDS = ["username", "password", "tls", "tls-ca", "uris"]
 
     def __init__(
         self,
-        charm,
+        model,
         relation_name: str,
         extra_user_roles: Optional[str] = None,
         additional_secret_fields: Optional[List[str]] = [],
     ):
         """Manager of base client relations."""
-        super().__init__(charm, relation_name)
+        super().__init__(model, relation_name)
         self.extra_user_roles = extra_user_roles
         self._secret_fields = list(self.SECRET_FIELDS)
         if additional_secret_fields:
             self._secret_fields += additional_secret_fields
-
-        self.framework.observe(
-            self.charm.on[relation_name].relation_created, self._on_relation_created_event
-        )
-        self.framework.observe(
-            charm.on.secret_changed,
-            self._on_secret_changed_event,
-        )
+        self.data_component = self.local_unit
 
     @property
     def secret_fields(self) -> Optional[List[str]]:
@@ -1298,18 +1515,6 @@ def secret_fields(self) -> Optional[List[str]]:
         if self.secrets_enabled:
             return self._secret_fields
 
-    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)
-
     # Internal helper functions
 
     def _register_secret_to_relation(
@@ -1322,13 +1527,13 @@ def _register_secret_to_relation(
         then <arbitraty_label> will be "stuck" on the Secret object, whenever it may
         appear (i.e. as an event attribute, or fetched manually) on future occasions.
 
-        This will allow us to uniquely identify the secret on Provides side (typically on
+        This will allow us to uniquely identify the secret on Provider side (typically on
         'secret-changed' events), and map it to the corresponding relation.
         """
         label = self._generate_secret_label(relation_name, relation_id, group)
 
         # Fetchin the Secret's meta information ensuring that it's locally getting registered with
-        CachedSecret(self.charm, self.component, label, secret_id).meta
+        CachedSecret(self._model, self.component, label, secret_id).meta
 
     def _register_secrets_to_relation(self, relation: Relation, params_name_list: List[str]):
         """Make sure that secrets of the provided list are locally 'registered' from the databag.
@@ -1338,7 +1543,7 @@ def _register_secrets_to_relation(self, relation: Relation, params_name_list: Li
         if not relation.app:
             return
 
-        for group in SecretGroup:
+        for group in SECRET_GROUPS.groups():
             secret_field = self._generate_secret_field_name(group)
             if secret_field in params_name_list:
                 if secret_uri := relation.data[relation.app].get(secret_field):
@@ -1388,23 +1593,6 @@ def is_resource_created(self, relation_id: Optional[int] = None) -> bool:
                 else False
             )
 
-    # Event handlers
-
-    def _on_relation_created_event(self, event: RelationCreatedEvent) -> None:
-        """Event emitted when the relation is created."""
-        if not self.local_unit.is_leader():
-            return
-
-        if self.secret_fields:
-            set_encoded_field(
-                event.relation, self.charm.app, REQ_SECRET_FIELDS, self.secret_fields
-            )
-
-    @abstractmethod
-    def _on_secret_changed_event(self, event: RelationChangedEvent) -> None:
-        """Event emitted when the relation data has changed."""
-        raise NotImplementedError
-
     # Mandatory internal overrides
 
     @juju_secrets_only
@@ -1421,7 +1609,7 @@ def _get_relation_secret(
     def _fetch_specific_relation_data(
         self, relation, fields: Optional[List[str]] = None
     ) -> Dict[str, str]:
-        """Fetching Requires data -- that may include secrets."""
+        """Fetching Requirer data -- that may include secrets."""
         if not relation.app:
             return {}
         return self._fetch_relation_data_with_secrets(
@@ -1459,35 +1647,91 @@ def _delete_relation_data(self, relation: Relation, fields: List[str]) -> None:
 
     # Public functions -- inherited
 
-    fetch_my_relation_data = leader_only(DataRelation.fetch_my_relation_data)
-    fetch_my_relation_field = leader_only(DataRelation.fetch_my_relation_field)
+    fetch_my_relation_data = leader_only(Data.fetch_my_relation_data)
+    fetch_my_relation_field = leader_only(Data.fetch_my_relation_field)
 
 
-# Base DataPeer
+class RequirerEventHandlers(EventHandlers):
+    """Requires-side of the relation."""
 
+    def __init__(self, charm: CharmBase, relation_data: RequirerData, unique_key: str = ""):
+        """Manager of base client relations."""
+        super().__init__(charm, relation_data, unique_key)
+
+        self.framework.observe(
+            self.charm.on[relation_data.relation_name].relation_created,
+            self._on_relation_created_event,
+        )
+        self.framework.observe(
+            charm.on.secret_changed,
+            self._on_secret_changed_event,
+        )
+
+    # Event handlers
+
+    def _on_relation_created_event(self, event: RelationCreatedEvent) -> None:
+        """Event emitted when the relation is created."""
+        if not self.relation_data.local_unit.is_leader():
+            return
+
+        if self.relation_data.secret_fields:  # pyright: ignore [reportAttributeAccessIssue]
+            set_encoded_field(
+                event.relation,
+                self.relation_data.component,
+                REQ_SECRET_FIELDS,
+                self.relation_data.secret_fields,  # pyright: ignore [reportAttributeAccessIssue]
+            )
+
+    @abstractmethod
+    def _on_secret_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation data has changed."""
+        raise NotImplementedError
 
-class DataPeer(DataRequires, DataProvides):
-    """Represents peer relations."""
 
-    SECRET_FIELDS = ["operator-password"]
+################################################################################
+# Peer Relation Data
+################################################################################
+
+
+class DataPeerData(RequirerData, ProviderData):
+    """Represents peer relations data."""
+
+    SECRET_FIELDS = []
     SECRET_FIELD_NAME = "internal_secret"
     SECRET_LABEL_MAP = {}
 
     def __init__(
         self,
-        charm,
+        model,
         relation_name: str,
         extra_user_roles: Optional[str] = None,
         additional_secret_fields: Optional[List[str]] = [],
+        additional_secret_group_mapping: Dict[str, str] = {},
         secret_field_name: Optional[str] = None,
         deleted_label: Optional[str] = None,
     ):
         """Manager of base client relations."""
-        DataRequires.__init__(
-            self, charm, relation_name, extra_user_roles, additional_secret_fields
+        RequirerData.__init__(
+            self,
+            model,
+            relation_name,
+            extra_user_roles,
+            additional_secret_fields,
         )
         self.secret_field_name = secret_field_name if secret_field_name else self.SECRET_FIELD_NAME
         self.deleted_label = deleted_label
+        self._secret_label_map = {}
+        # Secrets that are being dynamically added within the scope of this event handler run
+        self._new_secrets = []
+
+        for group, fields in additional_secret_group_mapping.items():
+            if group not in SECRET_GROUPS.groups():
+                setattr(SECRET_GROUPS, group, group)
+            for field in fields:
+                secret_group = SECRET_GROUPS.get_group(group)
+                internal_field = self._field_to_internal_name(field, secret_group)
+                self._secret_label_map.setdefault(group, []).append(internal_field)
+                self._secret_fields.append(internal_field)
 
     @property
     def scope(self) -> Optional[Scope]:
@@ -1497,6 +1741,137 @@ def scope(self) -> Optional[Scope]:
         if isinstance(self.component, Unit):
             return Scope.UNIT
 
+    @property
+    def secret_label_map(self) -> Dict[str, str]:
+        """Property storing secret mappings."""
+        return self._secret_label_map
+
+    @property
+    def static_secret_fields(self) -> List[str]:
+        """Re-definition of the property in a way that dynamically extended list is retrieved."""
+        return self._secret_fields
+
+    @property
+    def secret_fields(self) -> List[str]:
+        """Re-definition of the property in a way that dynamically extended list is retrieved."""
+        return (
+            self.static_secret_fields if self.static_secret_fields else self.current_secret_fields
+        )
+
+    @property
+    def current_secret_fields(self) -> List[str]:
+        """Helper method to get all currently existing secret fields (added statically or dynamically)."""
+        if not self.secrets_enabled:
+            return []
+
+        if len(self._model.relations[self.relation_name]) > 1:
+            raise ValueError(f"More than one peer relation on {self.relation_name}")
+
+        relation = self._model.relations[self.relation_name][0]
+        fields = []
+        for group in SECRET_GROUPS.groups():
+            if content := self._get_group_secret_contents(relation, group):
+                fields += [self._field_to_internal_name(field, group) for field in content]
+        return list(set(fields) | set(self._new_secrets))
+
+    @juju_secrets_only
+    @dynamic_secrets_only
+    def set_secret(
+        self,
+        relation_id: int,
+        field: str,
+        value: str,
+        group_mapping: Optional[SecretGroup] = None,
+    ) -> None:
+        """Public interface method to add a Relation Data field specifically as a Juju Secret.
+
+        Args:
+            relation_id: ID of the relation
+            field: The secret field that is to be added
+            value: The string value of the secret
+            group_mapping: The name of the "secret group", in case the field is to be added to an existing secret
+        """
+        full_field = self._field_to_internal_name(field, group_mapping)
+        if full_field not in self.current_secret_fields:
+            self._new_secrets.append(full_field)
+        self.update_relation_data(relation_id, {full_field: value})
+
+    # Unlike for set_secret(), there's no harm using this operation with static secrets
+    # The restricion is only added to keep the concept clear
+    @juju_secrets_only
+    @dynamic_secrets_only
+    def get_secret(
+        self,
+        relation_id: int,
+        field: str,
+        group_mapping: Optional[SecretGroup] = None,
+    ) -> Optional[str]:
+        """Public interface method to fetch secrets only."""
+        full_field = self._field_to_internal_name(field, group_mapping)
+        if full_field not in self.current_secret_fields:
+            raise SecretsUnavailableError(
+                f"Secret {field} from group {group_mapping} was not found"
+            )
+        return self.fetch_my_relation_field(relation_id, full_field)
+
+    @juju_secrets_only
+    @dynamic_secrets_only
+    def delete_secret(
+        self,
+        relation_id: int,
+        field: str,
+        group_mapping: Optional[SecretGroup] = None,
+    ) -> Optional[str]:
+        """Public interface method to delete secrets only."""
+        full_field = self._field_to_internal_name(field, group_mapping)
+        if full_field not in self.current_secret_fields:
+            logger.warning(f"Secret {field} from group {group_mapping} was not found")
+        self.delete_relation_data(relation_id, [full_field])
+
+    # Helpers
+
+    @staticmethod
+    def _field_to_internal_name(field: str, group: Optional[SecretGroup]) -> str:
+        if not group or group == SECRET_GROUPS.EXTRA:
+            return field
+        return f"{field}{GROUP_SEPARATOR}{group}"
+
+    @staticmethod
+    def _internal_name_to_field(name: str) -> Tuple[str, SecretGroup]:
+        parts = name.split(GROUP_SEPARATOR)
+        if not len(parts) > 1:
+            return (parts[0], SECRET_GROUPS.EXTRA)
+        secret_group = SECRET_GROUPS.get_group(parts[1])
+        if not secret_group:
+            raise ValueError(f"Invalid secret field {name}")
+        return (parts[0], secret_group)
+
+    def _group_secret_fields(self, secret_fields: List[str]) -> Dict[SecretGroup, List[str]]:
+        """Helper function to arrange secret mappings under their group.
+
+        NOTE: All unrecognized items end up in the 'extra' secret bucket.
+        Make sure only secret fields are passed!
+        """
+        secret_fieldnames_grouped = {}
+        for key in secret_fields:
+            field, group = self._internal_name_to_field(key)
+            secret_fieldnames_grouped.setdefault(group, []).append(field)
+        return secret_fieldnames_grouped
+
+    def _content_for_secret_group(
+        self, content: Dict[str, str], secret_fields: Set[str], group_mapping: SecretGroup
+    ) -> Dict[str, str]:
+        """Select <field>: <value> pairs from input, that belong to this particular Secret group."""
+        if group_mapping == SECRET_GROUPS.EXTRA:
+            return {k: v for k, v in content.items() if k in self.secret_fields}
+        return {
+            self._internal_name_to_field(k)[0]: v
+            for k, v in content.items()
+            if k in self.secret_fields
+        }
+
+    # Event handlers
+
     def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
         """Event emitted when the relation has changed."""
         pass
@@ -1505,15 +1880,19 @@ def _on_secret_changed_event(self, event: SecretChangedEvent) -> None:
         """Event emitted when the secret has changed."""
         pass
 
+    # Overrides of Relation Data handling functions
+
     def _generate_secret_label(
         self, relation_name: str, relation_id: int, group_mapping: SecretGroup
     ) -> str:
-        members = [self.charm.app.name]
+        members = [self._model.app.name]
         if self.scope:
             members.append(self.scope.value)
+        if group_mapping != SECRET_GROUPS.EXTRA:
+            members.append(group_mapping)
         return f"{'.'.join(members)}"
 
-    def _generate_secret_field_name(self, group_mapping: SecretGroup = SecretGroup.EXTRA) -> str:
+    def _generate_secret_field_name(self, group_mapping: SecretGroup = SECRET_GROUPS.EXTRA) -> str:
         """Generate unique group_mappings for secrets within a relation context."""
         return f"{self.secret_field_name}"
 
@@ -1521,7 +1900,7 @@ def _generate_secret_field_name(self, group_mapping: SecretGroup = SecretGroup.E
     def _get_relation_secret(
         self,
         relation_id: int,
-        group_mapping: SecretGroup = SecretGroup.EXTRA,
+        group_mapping: SecretGroup = SECRET_GROUPS.EXTRA,
         relation_name: Optional[str] = None,
     ) -> Optional[CachedSecret]:
         """Retrieve a Juju Secret specifically for peer relations.
@@ -1533,7 +1912,7 @@ def _get_relation_secret(
         if not relation_name:
             relation_name = self.relation_name
 
-        relation = self.charm.model.get_relation(relation_name, relation_id)
+        relation = self._model.get_relation(relation_name, relation_id)
         if not relation:
             return
 
@@ -1555,13 +1934,18 @@ def _get_group_secret_contents(
         self,
         relation: Relation,
         group: SecretGroup,
-        secret_fields: Optional[Union[Set[str], List[str]]] = None,
+        secret_fields: Union[Set[str], List[str]] = [],
     ) -> Dict[str, str]:
         """Helper function to retrieve collective, requested contents of a secret."""
+        secret_fields = [self._internal_name_to_field(k)[0] for k in secret_fields]
         result = super()._get_group_secret_contents(relation, group, secret_fields)
         if not self.deleted_label:
             return result
-        return {key: result[key] for key in result if result[key] != self.deleted_label}
+        return {
+            self._field_to_internal_name(key, group): result[key]
+            for key in result
+            if result[key] != self.deleted_label
+        }
 
     def _remove_secret_from_databag(self, relation, fields: List[str]) -> None:
         """For Rolling Upgrades -- when moving from databag to secrets usage.
@@ -1577,14 +1961,7 @@ def _remove_secret_from_databag(self, relation, fields: List[str]) -> None:
             if self._fetch_relation_data_without_secrets(self.component, relation, [field]):
                 self._delete_relation_data_without_secrets(self.component, relation, [field])
 
-    def _fetch_specific_relation_data(
-        self, relation: Relation, fields: Optional[List[str]]
-    ) -> Dict[str, str]:
-        """Fetch data available (directily or indirectly -- i.e. secrets) from the relation."""
-        return self._fetch_relation_data_with_secrets(
-            self.component, self.secret_fields, relation, fields
-        )
-
+    @either_static_or_dynamic_secrets
     def _fetch_my_specific_relation_data(
         self, relation: Relation, fields: Optional[List[str]]
     ) -> Dict[str, str]:
@@ -1593,6 +1970,7 @@ def _fetch_my_specific_relation_data(
             self.component, self.secret_fields, relation, fields
         )
 
+    @either_static_or_dynamic_secrets
     def _update_relation_data(self, relation: Relation, data: Dict[str, str]) -> None:
         """Update data available (directily or indirectly -- i.e. secrets) from the relation for owner/this_app."""
         self._remove_secret_from_databag(relation, list(data.keys()))
@@ -1608,6 +1986,7 @@ def _update_relation_data(self, relation: Relation, data: Dict[str, str]) -> Non
         normal_content = {k: v for k, v in data.items() if k in normal_fields}
         self._update_relation_data_without_secrets(self.component, relation, normal_content)
 
+    @either_static_or_dynamic_secrets
     def _delete_relation_data(self, relation: Relation, fields: List[str]) -> None:
         """Delete data available (directily or indirectly -- i.e. secrets) from the relation for owner/this_app."""
         if self.secret_fields and self.deleted_label:
@@ -1657,22 +2036,159 @@ def fetch_relation_field(
             "fetch_my_relation_data() and fetch_my_relation_field()"
         )
 
-    # Public functions -- inherited
+    def fetch_my_relation_field(
+        self, relation_id: int, field: str, relation_name: Optional[str] = None
+    ) -> Optional[str]:
+        """Get a single field from the relation data -- owner side.
 
-    fetch_my_relation_data = DataRelation.fetch_my_relation_data
-    fetch_my_relation_field = DataRelation.fetch_my_relation_field
+        Re-implementing the inherited function due to field@group conversion
+        """
+        if relation_data := self.fetch_my_relation_data([relation_id], [field], relation_name):
+            return relation_data.get(relation_id, {}).get(self._internal_name_to_field(field)[0])
 
+    # Public functions -- inherited
 
-class DataPeerUnit(DataPeer):
-    """Unit databag representation."""
+    fetch_my_relation_data = Data.fetch_my_relation_data
 
-    SCOPE = Scope.UNIT
 
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
+class DataPeerEventHandlers(RequirerEventHandlers):
+    """Requires-side of the relation."""
 
+    def __init__(self, charm: CharmBase, relation_data: RequirerData, unique_key: str = ""):
+        """Manager of base client relations."""
+        super().__init__(charm, relation_data, unique_key)
 
-# General events
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        pass
+
+    def _on_secret_changed_event(self, event: SecretChangedEvent) -> None:
+        """Event emitted when the secret has changed."""
+        pass
+
+
+class DataPeer(DataPeerData, DataPeerEventHandlers):
+    """Represents peer relations."""
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        extra_user_roles: Optional[str] = None,
+        additional_secret_fields: Optional[List[str]] = [],
+        additional_secret_group_mapping: Dict[str, str] = {},
+        secret_field_name: Optional[str] = None,
+        deleted_label: Optional[str] = None,
+        unique_key: str = "",
+    ):
+        DataPeerData.__init__(
+            self,
+            charm.model,
+            relation_name,
+            extra_user_roles,
+            additional_secret_fields,
+            additional_secret_group_mapping,
+            secret_field_name,
+            deleted_label,
+        )
+        DataPeerEventHandlers.__init__(self, charm, self, unique_key)
+
+
+class DataPeerUnitData(DataPeerData):
+    """Unit data abstraction representation."""
+
+    SCOPE = Scope.UNIT
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+
+class DataPeerUnit(DataPeerUnitData, DataPeerEventHandlers):
+    """Unit databag representation."""
+
+    def __init__(
+        self,
+        charm,
+        relation_name: str,
+        extra_user_roles: Optional[str] = None,
+        additional_secret_fields: Optional[List[str]] = [],
+        additional_secret_group_mapping: Dict[str, str] = {},
+        secret_field_name: Optional[str] = None,
+        deleted_label: Optional[str] = None,
+        unique_key: str = "",
+    ):
+        DataPeerData.__init__(
+            self,
+            charm.model,
+            relation_name,
+            extra_user_roles,
+            additional_secret_fields,
+            additional_secret_group_mapping,
+            secret_field_name,
+            deleted_label,
+        )
+        DataPeerEventHandlers.__init__(self, charm, self, unique_key)
+
+
+class DataPeerOtherUnitData(DataPeerUnitData):
+    """Unit data abstraction representation."""
+
+    def __init__(self, unit: Unit, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.local_unit = unit
+        self.component = unit
+
+    def update_relation_data(self, relation_id: int, data: dict) -> None:
+        """This method makes no sense for a Other Peer Relation."""
+        raise NotImplementedError("It's not possible to update data of another unit.")
+
+    def delete_relation_data(self, relation_id: int, fields: List[str]) -> None:
+        """This method makes no sense for a Other Peer Relation."""
+        raise NotImplementedError("It's not possible to delete data of another unit.")
+
+
+class DataPeerOtherUnitEventHandlers(DataPeerEventHandlers):
+    """Requires-side of the relation."""
+
+    def __init__(self, charm: CharmBase, relation_data: DataPeerUnitData):
+        """Manager of base client relations."""
+        unique_key = f"{relation_data.relation_name}-{relation_data.local_unit.name}"
+        super().__init__(charm, relation_data, unique_key=unique_key)
+
+
+class DataPeerOtherUnit(DataPeerOtherUnitData, DataPeerOtherUnitEventHandlers):
+    """Unit databag representation for another unit than the executor."""
+
+    def __init__(
+        self,
+        unit: Unit,
+        charm: CharmBase,
+        relation_name: str,
+        extra_user_roles: Optional[str] = None,
+        additional_secret_fields: Optional[List[str]] = [],
+        additional_secret_group_mapping: Dict[str, str] = {},
+        secret_field_name: Optional[str] = None,
+        deleted_label: Optional[str] = None,
+    ):
+        DataPeerOtherUnitData.__init__(
+            self,
+            unit,
+            charm.model,
+            relation_name,
+            extra_user_roles,
+            additional_secret_fields,
+            additional_secret_group_mapping,
+            secret_field_name,
+            deleted_label,
+        )
+        DataPeerOtherUnitEventHandlers.__init__(self, charm, self)
+
+
+################################################################################
+# Cross-charm Relatoins Data Handling and Evenets
+################################################################################
+
+# Generic events
 
 
 class ExtraRoleEvent(RelationEvent):
@@ -1917,28 +2433,11 @@ class DatabaseRequiresEvents(CharmEvents):
 # Database Provider and Requires
 
 
-class DatabaseProvides(DataProvides):
-    """Provider-side of the database relations."""
-
-    on = DatabaseProvidesEvents()  # pyright: ignore [reportAssignmentType]
+class DatabaseProviderData(ProviderData):
+    """Provider-side data of the database relations."""
 
-    def __init__(self, charm: CharmBase, relation_name: str) -> None:
-        super().__init__(charm, relation_name)
-
-    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
-        """Event emitted when the relation has changed."""
-        # Leader only
-        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:
-            getattr(self.on, "database_requested").emit(
-                event.relation, app=event.app, unit=event.unit
-            )
+    def __init__(self, model: Model, relation_name: str) -> None:
+        super().__init__(model, relation_name)
 
     def set_database(self, relation_id: int, database_name: str) -> None:
         """Set database name.
@@ -2012,14 +2511,49 @@ def set_version(self, relation_id: int, version: str) -> None:
         self.update_relation_data(relation_id, {"version": version})
 
 
-class DatabaseRequires(DataRequires):
-    """Requires-side of the database relation."""
+class DatabaseProviderEventHandlers(EventHandlers):
+    """Provider-side of the database relation handlers."""
 
-    on = DatabaseRequiresEvents()  # pyright: ignore [reportAssignmentType]
+    on = DatabaseProvidesEvents()  # pyright: ignore [reportAssignmentType]
+
+    def __init__(
+        self, charm: CharmBase, relation_data: DatabaseProviderData, unique_key: str = ""
+    ):
+        """Manager of base client relations."""
+        super().__init__(charm, relation_data, unique_key)
+        # Just to calm down pyright, it can't parse that the same type is being used in the super() call above
+        self.relation_data = relation_data
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Leader only
+        if not self.relation_data.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:
+            getattr(self.on, "database_requested").emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+
+class DatabaseProvides(DatabaseProviderData, DatabaseProviderEventHandlers):
+    """Provider-side of the database relations."""
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        DatabaseProviderData.__init__(self, charm.model, relation_name)
+        DatabaseProviderEventHandlers.__init__(self, charm, self)
+
+
+class DatabaseRequirerData(RequirerData):
+    """Requirer-side of the database relation."""
 
     def __init__(
         self,
-        charm,
+        model: Model,
         relation_name: str,
         database_name: str,
         extra_user_roles: Optional[str] = None,
@@ -2028,23 +2562,89 @@ def __init__(
         external_node_connectivity: bool = False,
     ):
         """Manager of database client relations."""
-        super().__init__(charm, relation_name, extra_user_roles, additional_secret_fields)
+        super().__init__(model, relation_name, extra_user_roles, additional_secret_fields)
         self.database = database_name
         self.relations_aliases = relations_aliases
         self.external_node_connectivity = external_node_connectivity
 
+    def is_postgresql_plugin_enabled(self, plugin: str, relation_index: int = 0) -> bool:
+        """Returns whether a plugin is enabled in the database.
+
+        Args:
+            plugin: name of the plugin to check.
+            relation_index: optional relation index to check the database
+                (default: 0 - first relation).
+
+        PostgreSQL only.
+        """
+        # Psycopg 3 is imported locally to avoid the need of its package installation
+        # when relating to a database charm other than PostgreSQL.
+        import psycopg
+
+        # Return False if no relation is established.
+        if len(self.relations) == 0:
+            return False
+
+        relation_id = self.relations[relation_index].id
+        host = self.fetch_relation_field(relation_id, "endpoints")
+
+        # Return False if there is no endpoint available.
+        if host is None:
+            return False
+
+        host = host.split(":")[0]
+
+        content = self.fetch_relation_data([relation_id], ["username", "password"]).get(
+            relation_id, {}
+        )
+        user = content.get("username")
+        password = content.get("password")
+
+        connection_string = (
+            f"host='{host}' dbname='{self.database}' user='{user}' password='{password}'"
+        )
+        try:
+            with psycopg.connect(connection_string) as connection:
+                with connection.cursor() as cursor:
+                    cursor.execute(
+                        "SELECT TRUE FROM pg_extension WHERE extname=%s::text;", (plugin,)
+                    )
+                    return cursor.fetchone() is not None
+        except psycopg.Error as e:
+            logger.exception(
+                f"failed to check whether {plugin} plugin is enabled in the database: %s", str(e)
+            )
+            return False
+
+
+class DatabaseRequirerEventHandlers(RequirerEventHandlers):
+    """Requires-side of the relation."""
+
+    on = DatabaseRequiresEvents()  # pyright: ignore [reportAssignmentType]
+
+    def __init__(
+        self, charm: CharmBase, relation_data: DatabaseRequirerData, unique_key: str = ""
+    ):
+        """Manager of base client relations."""
+        super().__init__(charm, relation_data, unique_key)
+        # Just to keep lint quiet, can't resolve inheritance. The same happened in super().__init__() above
+        self.relation_data = relation_data
+
         # Define custom event names for each alias.
-        if relations_aliases:
+        if self.relation_data.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:
+            relation_connection_limit = self.charm.meta.requires[
+                self.relation_data.relation_name
+            ].limit
+            if len(self.relation_data.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)}"
+                    f"Expected {relation_connection_limit}, got {len(self.relation_data.relations_aliases)}"
                 )
 
-            for relation_alias in relations_aliases:
+        if self.relation_data.relations_aliases:
+            for relation_alias in self.relation_data.relations_aliases:
                 self.on.define_event(f"{relation_alias}_database_created", DatabaseCreatedEvent)
                 self.on.define_event(
                     f"{relation_alias}_endpoints_changed", DatabaseEndpointsChangedEvent
@@ -2067,32 +2667,32 @@ def _assign_relation_alias(self, relation_id: int) -> None:
             relation_id: the identifier for a particular relation.
         """
         # If no aliases were provided, return immediately.
-        if not self.relations_aliases:
+        if not self.relation_data.relations_aliases:
             return
 
         # Return if an alias was already assigned to this relation
         # (like when there are more than one unit joining the relation).
-        relation = self.charm.model.get_relation(self.relation_name, relation_id)
-        if relation and relation.data[self.local_unit].get("alias"):
+        relation = self.charm.model.get_relation(self.relation_data.relation_name, relation_id)
+        if relation and relation.data[self.relation_data.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")
+        available_aliases = self.relation_data.relations_aliases[:]
+        for relation in self.charm.model.relations[self.relation_data.relation_name]:
+            alias = relation.data[self.relation_data.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 = self.charm.model.get_relation(self.relation_data.relation_name, relation_id)
         if relation:
-            relation.data[self.local_unit].update({"alias": available_aliases[0]})
+            relation.data[self.relation_data.local_unit].update({"alias": available_aliases[0]})
 
         # We need to set relation alias also on the application level so,
         # it will be accessible in show-unit juju command, executed for a consumer application unit
-        if self.local_unit.is_leader():
-            self.update_relation_data(relation_id, {"alias": available_aliases[0]})
+        if self.relation_data.local_unit.is_leader():
+            self.relation_data.update_relation_data(relation_id, {"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.
@@ -2116,60 +2716,11 @@ def _get_relation_alias(self, relation_id: int) -> Optional[str]:
         Returns:
             the relation alias or None if the relation was not found.
         """
-        for relation in self.charm.model.relations[self.relation_name]:
+        for relation in self.charm.model.relations[self.relation_data.relation_name]:
             if relation.id == relation_id:
-                return relation.data[self.local_unit].get("alias")
+                return relation.data[self.relation_data.local_unit].get("alias")
         return None
 
-    def is_postgresql_plugin_enabled(self, plugin: str, relation_index: int = 0) -> bool:
-        """Returns whether a plugin is enabled in the database.
-
-        Args:
-            plugin: name of the plugin to check.
-            relation_index: optional relation index to check the database
-                (default: 0 - first relation).
-
-        PostgreSQL only.
-        """
-        # Psycopg 3 is imported locally to avoid the need of its package installation
-        # when relating to a database charm other than PostgreSQL.
-        import psycopg
-
-        # Return False if no relation is established.
-        if len(self.relations) == 0:
-            return False
-
-        relation_id = self.relations[relation_index].id
-        host = self.fetch_relation_field(relation_id, "endpoints")
-
-        # Return False if there is no endpoint available.
-        if host is None:
-            return False
-
-        host = host.split(":")[0]
-
-        content = self.fetch_relation_data([relation_id], ["username", "password"]).get(
-            relation_id, {}
-        )
-        user = content.get("username")
-        password = content.get("password")
-
-        connection_string = (
-            f"host='{host}' dbname='{self.database}' user='{user}' password='{password}'"
-        )
-        try:
-            with psycopg.connect(connection_string) as connection:
-                with connection.cursor() as cursor:
-                    cursor.execute(
-                        "SELECT TRUE FROM pg_extension WHERE extname=%s::text;", (plugin,)
-                    )
-                    return cursor.fetchone() is not None
-        except psycopg.Error as e:
-            logger.exception(
-                f"failed to check whether {plugin} plugin is enabled in the database: %s", str(e)
-            )
-            return False
-
     def _on_relation_created_event(self, event: RelationCreatedEvent) -> None:
         """Event emitted when the database relation is created."""
         super()._on_relation_created_event(event)
@@ -2179,19 +2730,19 @@ def _on_relation_created_event(self, event: RelationCreatedEvent) -> None:
 
         # Sets both database and extra user roles in the relation
         # if the roles are provided. Otherwise, sets only the database.
-        if not self.local_unit.is_leader():
+        if not self.relation_data.local_unit.is_leader():
             return
 
-        event_data = {"database": self.database}
+        event_data = {"database": self.relation_data.database}
 
-        if self.extra_user_roles:
-            event_data["extra-user-roles"] = self.extra_user_roles
+        if self.relation_data.extra_user_roles:
+            event_data["extra-user-roles"] = self.relation_data.extra_user_roles
 
         # set external-node-connectivity field
-        if self.external_node_connectivity:
+        if self.relation_data.external_node_connectivity:
             event_data["external-node-connectivity"] = "true"
 
-        self.update_relation_data(event.relation.id, event_data)
+        self.relation_data.update_relation_data(event.relation.id, event_data)
 
     def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
         """Event emitted when the database relation has changed."""
@@ -2199,12 +2750,12 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
         diff = self._diff(event)
 
         # Register all new secrets with their labels
-        if any(newval for newval in diff.added if self._is_secret_field(newval)):
-            self._register_secrets_to_relation(event.relation, diff.added)
+        if any(newval for newval in diff.added if self.relation_data._is_secret_field(newval)):
+            self.relation_data._register_secrets_to_relation(event.relation, diff.added)
 
         # Check if the database is created
         # (the database charm shared the credentials).
-        secret_field_user = self._generate_secret_field_name(SecretGroup.USER)
+        secret_field_user = self.relation_data._generate_secret_field_name(SECRET_GROUPS.USER)
         if (
             "username" in diff.added and "password" in diff.added
         ) or secret_field_user in diff.added:
@@ -2250,7 +2801,37 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
             self._emit_aliased_event(event, "read_only_endpoints_changed")
 
 
-# Kafka related events
+class DatabaseRequires(DatabaseRequirerData, DatabaseRequirerEventHandlers):
+    """Provider-side of the database relations."""
+
+    def __init__(
+        self,
+        charm: CharmBase,
+        relation_name: str,
+        database_name: str,
+        extra_user_roles: Optional[str] = None,
+        relations_aliases: Optional[List[str]] = None,
+        additional_secret_fields: Optional[List[str]] = [],
+        external_node_connectivity: bool = False,
+    ):
+        DatabaseRequirerData.__init__(
+            self,
+            charm.model,
+            relation_name,
+            database_name,
+            extra_user_roles,
+            relations_aliases,
+            additional_secret_fields,
+            external_node_connectivity,
+        )
+        DatabaseRequirerEventHandlers.__init__(self, charm, self)
+
+
+################################################################################
+# Charm-specific Relations Data and Events
+################################################################################
+
+# Kafka Events
 
 
 class KafkaProvidesEvent(RelationEvent):
@@ -2343,29 +2924,11 @@ class KafkaRequiresEvents(CharmEvents):
 # Kafka Provides and Requires
 
 
-class KafkaProvides(DataProvides):
+class KafkaProvidesData(ProviderData):
     """Provider-side of the Kafka relation."""
 
-    on = KafkaProvidesEvents()  # pyright: ignore [reportAssignmentType]
-
-    def __init__(self, charm: CharmBase, relation_name: str) -> None:
-        super().__init__(charm, relation_name)
-
-    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
-        """Event emitted when the relation has changed."""
-        # Leader only
-        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:
-            getattr(self.on, "topic_requested").emit(
-                event.relation, app=event.app, unit=event.unit
-            )
+    def __init__(self, model: Model, relation_name: str) -> None:
+        super().__init__(model, relation_name)
 
     def set_topic(self, relation_id: int, topic: str) -> None:
         """Set topic name in the application relation databag.
@@ -2404,14 +2967,47 @@ def set_zookeeper_uris(self, relation_id: int, zookeeper_uris: str) -> None:
         self.update_relation_data(relation_id, {"zookeeper-uris": zookeeper_uris})
 
 
-class KafkaRequires(DataRequires):
-    """Requires-side of the Kafka relation."""
+class KafkaProvidesEventHandlers(EventHandlers):
+    """Provider-side of the Kafka relation."""
 
-    on = KafkaRequiresEvents()  # pyright: ignore [reportAssignmentType]
+    on = KafkaProvidesEvents()  # pyright: ignore [reportAssignmentType]
+
+    def __init__(self, charm: CharmBase, relation_data: KafkaProvidesData) -> None:
+        super().__init__(charm, relation_data)
+        # Just to keep lint quiet, can't resolve inheritance. The same happened in super().__init__() above
+        self.relation_data = relation_data
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Leader only
+        if not self.relation_data.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:
+            getattr(self.on, "topic_requested").emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+
+class KafkaProvides(KafkaProvidesData, KafkaProvidesEventHandlers):
+    """Provider-side of the Kafka relation."""
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        KafkaProvidesData.__init__(self, charm.model, relation_name)
+        KafkaProvidesEventHandlers.__init__(self, charm, self)
+
+
+class KafkaRequiresData(RequirerData):
+    """Requirer-side of the Kafka relation."""
 
     def __init__(
         self,
-        charm,
+        model: Model,
         relation_name: str,
         topic: str,
         extra_user_roles: Optional[str] = None,
@@ -2419,9 +3015,7 @@ def __init__(
         additional_secret_fields: Optional[List[str]] = [],
     ):
         """Manager of Kafka client relations."""
-        # super().__init__(charm, relation_name)
-        super().__init__(charm, relation_name, extra_user_roles, additional_secret_fields)
-        self.charm = charm
+        super().__init__(model, relation_name, extra_user_roles, additional_secret_fields)
         self.topic = topic
         self.consumer_group_prefix = consumer_group_prefix or ""
 
@@ -2437,11 +3031,22 @@ def topic(self, value):
             raise ValueError(f"Error on topic '{value}', cannot be a wildcard.")
         self._topic = value
 
+
+class KafkaRequiresEventHandlers(RequirerEventHandlers):
+    """Requires-side of the Kafka relation."""
+
+    on = KafkaRequiresEvents()  # pyright: ignore [reportAssignmentType]
+
+    def __init__(self, charm: CharmBase, relation_data: KafkaRequiresData) -> None:
+        super().__init__(charm, relation_data)
+        # Just to keep lint quiet, can't resolve inheritance. The same happened in super().__init__() above
+        self.relation_data = relation_data
+
     def _on_relation_created_event(self, event: RelationCreatedEvent) -> None:
         """Event emitted when the Kafka relation is created."""
         super()._on_relation_created_event(event)
 
-        if not self.local_unit.is_leader():
+        if not self.relation_data.local_unit.is_leader():
             return
 
         # Sets topic, extra user roles, and "consumer-group-prefix" in the relation
@@ -2450,7 +3055,7 @@ def _on_relation_created_event(self, event: RelationCreatedEvent) -> None:
             for f in ["consumer-group-prefix", "extra-user-roles", "topic"]
         }
 
-        self.update_relation_data(event.relation.id, relation_data)
+        self.relation_data.update_relation_data(event.relation.id, relation_data)
 
     def _on_secret_changed_event(self, event: SecretChangedEvent):
         """Event notifying about a new value of a secret."""
@@ -2465,10 +3070,10 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
         # (the Kafka charm shared the credentials).
 
         # Register all new secrets with their labels
-        if any(newval for newval in diff.added if self._is_secret_field(newval)):
-            self._register_secrets_to_relation(event.relation, diff.added)
+        if any(newval for newval in diff.added if self.relation_data._is_secret_field(newval)):
+            self.relation_data._register_secrets_to_relation(event.relation, diff.added)
 
-        secret_field_user = self._generate_secret_field_name(SecretGroup.USER)
+        secret_field_user = self.relation_data._generate_secret_field_name(SECRET_GROUPS.USER)
         if (
             "username" in diff.added and "password" in diff.added
         ) or secret_field_user in diff.added:
@@ -2491,6 +3096,30 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
             return
 
 
+class KafkaRequires(KafkaRequiresData, KafkaRequiresEventHandlers):
+    """Provider-side of the Kafka relation."""
+
+    def __init__(
+        self,
+        charm: CharmBase,
+        relation_name: str,
+        topic: str,
+        extra_user_roles: Optional[str] = None,
+        consumer_group_prefix: Optional[str] = None,
+        additional_secret_fields: Optional[List[str]] = [],
+    ) -> None:
+        KafkaRequiresData.__init__(
+            self,
+            charm.model,
+            relation_name,
+            topic,
+            extra_user_roles,
+            consumer_group_prefix,
+            additional_secret_fields,
+        )
+        KafkaRequiresEventHandlers.__init__(self, charm, self)
+
+
 # Opensearch related events
 
 
@@ -2541,28 +3170,11 @@ class OpenSearchRequiresEvents(CharmEvents):
 # OpenSearch Provides and Requires Objects
 
 
-class OpenSearchProvides(DataProvides):
+class OpenSearchProvidesData(ProviderData):
     """Provider-side of the OpenSearch relation."""
 
-    on = OpenSearchProvidesEvents()  # pyright: ignore[reportAssignmentType]
-
-    def __init__(self, charm: CharmBase, relation_name: str) -> None:
-        super().__init__(charm, relation_name)
-
-    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
-        """Event emitted when the relation has changed."""
-        # Leader only
-        if not self.local_unit.is_leader():
-            return
-        # Check which data has changed to emit customs events.
-        diff = self._diff(event)
-
-        # Emit an index requested event if the setup key (index name and optional extra user roles)
-        # have been added to the relation databag by the application.
-        if "index" in diff.added:
-            getattr(self.on, "index_requested").emit(
-                event.relation, app=event.app, unit=event.unit
-            )
+    def __init__(self, model: Model, relation_name: str) -> None:
+        super().__init__(model, relation_name)
 
     def set_index(self, relation_id: int, index: str) -> None:
         """Set the index in the application relation databag.
@@ -2594,45 +3206,87 @@ def set_version(self, relation_id: int, version: str) -> None:
         self.update_relation_data(relation_id, {"version": version})
 
 
-class OpenSearchRequires(DataRequires):
-    """Requires-side of the OpenSearch relation."""
+class OpenSearchProvidesEventHandlers(EventHandlers):
+    """Provider-side of the OpenSearch relation."""
 
-    on = OpenSearchRequiresEvents()  # pyright: ignore[reportAssignmentType]
+    on = OpenSearchProvidesEvents()  # pyright: ignore[reportAssignmentType]
+
+    def __init__(self, charm: CharmBase, relation_data: OpenSearchProvidesData) -> None:
+        super().__init__(charm, relation_data)
+        # Just to keep lint quiet, can't resolve inheritance. The same happened in super().__init__() above
+        self.relation_data = relation_data
+
+    def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
+        """Event emitted when the relation has changed."""
+        # Leader only
+        if not self.relation_data.local_unit.is_leader():
+            return
+        # Check which data has changed to emit customs events.
+        diff = self._diff(event)
+
+        # Emit an index requested event if the setup key (index name and optional extra user roles)
+        # have been added to the relation databag by the application.
+        if "index" in diff.added:
+            getattr(self.on, "index_requested").emit(
+                event.relation, app=event.app, unit=event.unit
+            )
+
+
+class OpenSearchProvides(OpenSearchProvidesData, OpenSearchProvidesEventHandlers):
+    """Provider-side of the OpenSearch relation."""
+
+    def __init__(self, charm: CharmBase, relation_name: str) -> None:
+        OpenSearchProvidesData.__init__(self, charm.model, relation_name)
+        OpenSearchProvidesEventHandlers.__init__(self, charm, self)
+
+
+class OpenSearchRequiresData(RequirerData):
+    """Requires data side of the OpenSearch relation."""
 
     def __init__(
         self,
-        charm,
+        model: Model,
         relation_name: str,
         index: str,
         extra_user_roles: Optional[str] = None,
         additional_secret_fields: Optional[List[str]] = [],
     ):
         """Manager of OpenSearch client relations."""
-        super().__init__(charm, relation_name, extra_user_roles, additional_secret_fields)
-        self.charm = charm
+        super().__init__(model, relation_name, extra_user_roles, additional_secret_fields)
         self.index = index
 
+
+class OpenSearchRequiresEventHandlers(RequirerEventHandlers):
+    """Requires events side of the OpenSearch relation."""
+
+    on = OpenSearchRequiresEvents()  # pyright: ignore[reportAssignmentType]
+
+    def __init__(self, charm: CharmBase, relation_data: OpenSearchRequiresData) -> None:
+        super().__init__(charm, relation_data)
+        # Just to keep lint quiet, can't resolve inheritance. The same happened in super().__init__() above
+        self.relation_data = relation_data
+
     def _on_relation_created_event(self, event: RelationCreatedEvent) -> None:
         """Event emitted when the OpenSearch relation is created."""
         super()._on_relation_created_event(event)
 
-        if not self.local_unit.is_leader():
+        if not self.relation_data.local_unit.is_leader():
             return
 
         # Sets both index and extra user roles in the relation if the roles are provided.
         # Otherwise, sets only the index.
-        data = {"index": self.index}
-        if self.extra_user_roles:
-            data["extra-user-roles"] = self.extra_user_roles
+        data = {"index": self.relation_data.index}
+        if self.relation_data.extra_user_roles:
+            data["extra-user-roles"] = self.relation_data.extra_user_roles
 
-        self.update_relation_data(event.relation.id, data)
+        self.relation_data.update_relation_data(event.relation.id, data)
 
     def _on_secret_changed_event(self, event: SecretChangedEvent):
         """Event notifying about a new value of a secret."""
         if not event.secret.label:
             return
 
-        relation = self._relation_from_secret_label(event.secret.label)
+        relation = self.relation_data._relation_from_secret_label(event.secret.label)
         if not relation:
             logging.info(
                 f"Received secret {event.secret.label} but couldn't parse, seems irrelevant"
@@ -2661,11 +3315,11 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
         diff = self._diff(event)
 
         # Register all new secrets with their labels
-        if any(newval for newval in diff.added if self._is_secret_field(newval)):
-            self._register_secrets_to_relation(event.relation, diff.added)
+        if any(newval for newval in diff.added if self.relation_data._is_secret_field(newval)):
+            self.relation_data._register_secrets_to_relation(event.relation, diff.added)
 
-        secret_field_user = self._generate_secret_field_name(SecretGroup.USER)
-        secret_field_tls = self._generate_secret_field_name(SecretGroup.TLS)
+        secret_field_user = self.relation_data._generate_secret_field_name(SECRET_GROUPS.USER)
+        secret_field_tls = self.relation_data._generate_secret_field_name(SECRET_GROUPS.TLS)
         updates = {"username", "password", "tls", "tls-ca", secret_field_user, secret_field_tls}
         if len(set(diff._asdict().keys()) - updates) < len(diff):
             logger.info("authentication updated at: %s", datetime.now())
@@ -2695,3 +3349,25 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
                 event.relation, app=event.app, unit=event.unit
             )  # here check if this is the right design
             return
+
+
+class OpenSearchRequires(OpenSearchRequiresData, OpenSearchRequiresEventHandlers):
+    """Requires-side of the OpenSearch relation."""
+
+    def __init__(
+        self,
+        charm: CharmBase,
+        relation_name: str,
+        index: str,
+        extra_user_roles: Optional[str] = None,
+        additional_secret_fields: Optional[List[str]] = [],
+    ) -> None:
+        OpenSearchRequiresData.__init__(
+            self,
+            charm.model,
+            relation_name,
+            index,
+            extra_user_roles,
+            additional_secret_fields,
+        )
+        OpenSearchRequiresEventHandlers.__init__(self, charm, self)
diff --git a/metadata.yaml b/metadata.yaml
index eda42f1..0257cc1 100644
--- a/metadata.yaml
+++ b/metadata.yaml
@@ -11,7 +11,7 @@ resources:
   oci-image:
     type: oci-image
     description: Backing OCI image
-    upstream-source: docker.io/zenmldocker/zenml-server:0.55.3
+    upstream-source: docker.io/zenmldocker/zenml-server:0.56.3
 requires:
   relational-db:
     interface: mysql_client
diff --git a/releases/0.56/edge/zenml/bundle.yaml b/releases/0.56/edge/zenml/bundle.yaml
new file mode 100644
index 0000000..9e20b20
--- /dev/null
+++ b/releases/0.56/edge/zenml/bundle.yaml
@@ -0,0 +1,18 @@
+bundle: kubernetes
+name: zenml
+docs: https://discourse.charmhub.io/t/zenml-docs-index/12832
+applications:
+  zenml-mysql:
+    charm: mysql-k8s
+    channel: 8.0/stable
+    scale: 1
+    trust: true
+    _github_repo_name: mysql-k8s-operator
+  zenml-server:
+    charm: zenml-server
+    channel: 0.56/edge
+    scale: 1
+    trust: true
+    _github_repo_name: RafalSiwek/zenml-server-operator
+relations:
+  - [zenml-server, zenml-mysql]
diff --git a/releases/0.56/edge/zenml/charmcraft.yaml b/releases/0.56/edge/zenml/charmcraft.yaml
new file mode 100644
index 0000000..d7a57a8
--- /dev/null
+++ b/releases/0.56/edge/zenml/charmcraft.yaml
@@ -0,0 +1 @@
+type: bundle
diff --git a/releases/0.56/stable/zenml/bundle.yaml b/releases/0.56/stable/zenml/bundle.yaml
new file mode 100644
index 0000000..ee32f12
--- /dev/null
+++ b/releases/0.56/stable/zenml/bundle.yaml
@@ -0,0 +1,18 @@
+bundle: kubernetes
+name: zenml
+docs: https://discourse.charmhub.io/t/zenml-docs-index/12832
+applications:
+  zenml-mysql:
+    charm: mysql-k8s
+    channel: 8.0/stable
+    scale: 1
+    trust: true
+    _github_repo_name: mysql-k8s-operator
+  zenml-server:
+    charm: zenml-server
+    channel: 0.56/stable
+    scale: 1
+    trust: true
+    _github_repo_name: RafalSiwek/zenml-server-operator
+relations:
+  - [zenml-server, zenml-mysql]
diff --git a/releases/0.56/stable/zenml/charmcraft.yaml b/releases/0.56/stable/zenml/charmcraft.yaml
new file mode 100644
index 0000000..d7a57a8
--- /dev/null
+++ b/releases/0.56/stable/zenml/charmcraft.yaml
@@ -0,0 +1 @@
+type: bundle
diff --git a/requirements-integration.txt b/requirements-integration.txt
index de52f16..dec5532 100644
--- a/requirements-integration.txt
+++ b/requirements-integration.txt
@@ -3,5 +3,5 @@ jinja2
 juju<4.0
 pytest-operator
 requests
-zenml==0.55.3
+zenml==0.56.3
 -r requirements.txt
\ No newline at end of file