diff --git a/lib/charms/data_platform_libs/v0/data_interfaces.py b/lib/charms/data_platform_libs/v0/data_interfaces.py
index d24aa6ffc7..3ce69e155e 100644
--- a/lib/charms/data_platform_libs/v0/data_interfaces.py
+++ b/lib/charms/data_platform_libs/v0/data_interfaces.py
@@ -295,10 +295,21 @@ 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, Model, Secret, SecretInfo, SecretNotFoundError
 from ops.charm import (
@@ -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 = 29
+LIBPATCH = 34
 
 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]]:
@@ -453,6 +493,7 @@ def wrapper(self, *args, **kwargs):
             return
         return f(self, *args, **kwargs)
 
+    wrapper.leader_only = True
     return wrapper
 
 
@@ -467,6 +508,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."""
 
@@ -474,6 +543,11 @@ class Scope(Enum):
     UNIT = "unit"
 
 
+################################################################################
+# Secrets internal caching
+################################################################################
+
+
 class CachedSecret:
     """Locally cache a secret.
 
@@ -486,6 +560,7 @@ def __init__(
         component: Union[Application, Unit],
         label: str,
         secret_uri: Optional[str] = None,
+        legacy_labels: List[str] = [],
     ):
         self._secret_meta = None
         self._secret_content = {}
@@ -493,16 +568,25 @@ def __init__(
         self.label = label
         self._model = model
         self.component = component
+        self.legacy_labels = legacy_labels
+        self.current_label = None
 
-    def add_secret(self, content: Dict[str, str], relation: Relation) -> Secret:
+    def add_secret(
+        self,
+        content: Dict[str, str],
+        relation: Optional[Relation] = None,
+        label: Optional[str] = None,
+    ) -> Secret:
         """Create a new secret."""
         if self._secret_uri:
             raise SecretAlreadyExistsError(
                 "Secret is already defined with uri %s", self._secret_uri
             )
 
-        secret = self.component.add_secret(content, label=self.label)
-        if relation.app != self._model.app:
+        label = self.label if not label else label
+
+        secret = self.component.add_secret(content, label=label)
+        if relation and 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
@@ -515,13 +599,20 @@ def meta(self) -> Optional[Secret]:
         if not self._secret_meta:
             if not (self._secret_uri or self.label):
                 return
-            try:
-                self._secret_meta = self._model.get_secret(label=self.label)
-            except SecretNotFoundError:
-                if self._secret_uri:
-                    self._secret_meta = self._model.get_secret(
-                        id=self._secret_uri, label=self.label
-                    )
+
+            for label in [self.label] + self.legacy_labels:
+                try:
+                    self._secret_meta = self._model.get_secret(label=label)
+                except SecretNotFoundError:
+                    pass
+                else:
+                    if label != self.label:
+                        self.current_label = label
+                    break
+
+            # If still not found, to be checked by URI, to be labelled with the proposed label
+            if not self._secret_meta and self._secret_uri:
+                self._secret_meta = self._model.get_secret(id=self._secret_uri, label=self.label)
         return self._secret_meta
 
     def get_content(self) -> Dict[str, str]:
@@ -545,12 +636,30 @@ def get_content(self) -> Dict[str, str]:
                     self._secret_content = self.meta.get_content()
         return self._secret_content
 
+    def _move_to_new_label_if_needed(self):
+        """Helper function to re-create the secret with a different label."""
+        if not self.current_label or not (self.meta and self._secret_meta):
+            return
+
+        # Create a new secret with the new label
+        old_meta = self._secret_meta
+        content = self._secret_meta.get_content()
+
+        # I wish we could just check if we are the owners of the secret...
+        try:
+            self._secret_meta = self.add_secret(content, label=self.label)
+        except ModelError as err:
+            if "this unit is not the leader" not in str(err):
+                raise
+        old_meta.remove_all_revisions()
+
     def set_content(self, content: Dict[str, str]) -> None:
         """Setting cached secret content."""
         if not self.meta:
             return
 
         if content:
+            self._move_to_new_label_if_needed()
             self.meta.set_content(content)
             self._secret_content = content
         else:
@@ -582,10 +691,14 @@ def __init__(self, model: Model, component: Union[Application, Unit]):
         self.component = component
         self._secrets: Dict[str, CachedSecret] = {}
 
-    def get(self, label: str, uri: Optional[str] = None) -> Optional[CachedSecret]:
+    def get(
+        self, label: str, uri: Optional[str] = None, legacy_labels: List[str] = []
+    ) -> Optional[CachedSecret]:
         """Getting a secret from Juju Secret store or cache."""
         if not self._secrets.get(label):
-            secret = CachedSecret(self._model, self.component, label, uri)
+            secret = CachedSecret(
+                self._model, self.component, label, uri, legacy_labels=legacy_labels
+            )
             if secret.meta:
                 self._secrets[label] = secret
         return self._secrets.get(label)
@@ -603,15 +716,130 @@ def add(self, label: str, content: Dict[str, str], relation: Relation) -> Cached
     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)
+            try:
+                secret.remove()
+                self._secrets.pop(label)
+            except (SecretsUnavailableError, KeyError):
+                pass
+            else:
+                return
+        logging.debug("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
+
+        # Avoiding "leader_only" error when cross-charm non-leader unit, not to report useless error
+        if (
+            not hasattr(self.relation_data.fetch_my_relation_field, "leader_only")
+            or self.relation_data.component != self.relation_data.local_app
+            or self.relation_data.local_unit.is_leader()
+        ):
+            result = self.relation_data.fetch_my_relation_field(self.relation_id, key)
+
+        if not result:
+            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()
+
+    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 Data(ABC):
     """Base relation data mainpulation (abstract) class."""
 
@@ -619,11 +847,11 @@ class Data(ABC):
 
     # 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__(
@@ -656,6 +884,11 @@ def secrets_enabled(self):
             self._jujuversion = JujuVersion.from_environ()
         return self._jujuversion.has_secrets
 
+    @property
+    def secret_label_map(self):
+        """Exposing secret-label map via a property -- could be overridden in descendants!"""
+        return self.SECRET_LABEL_MAP
+
     # Mandatory overrides for internal/helper methods
 
     @abstractmethod
@@ -710,11 +943,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."""
@@ -739,8 +972,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.
@@ -748,44 +980,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
@@ -919,7 +1149,7 @@ def _delete_relation_data_without_secrets(
             try:
                 relation.data[component].pop(field)
             except KeyError:
-                logger.error(
+                logger.debug(
                     "Non-existing field '%s' was attempted to be removed from the databag (relation ID: %s)",
                     str(field),
                     str(relation.id),
@@ -929,6 +1159,10 @@ 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._model.get_relation(relation_name, relation_id)
@@ -1171,7 +1405,7 @@ def _delete_relation_secret(
             try:
                 new_content.pop(field)
             except KeyError:
-                logging.error(
+                logging.debug(
                     "Non-existing secret was attempted to be removed %s, %s",
                     str(relation.id),
                     str(field),
@@ -1363,7 +1597,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):
@@ -1497,7 +1731,7 @@ def _on_relation_created_event(self, event: RelationCreatedEvent) -> None:
         if self.relation_data.secret_fields:  # pyright: ignore [reportAttributeAccessIssue]
             set_encoded_field(
                 event.relation,
-                self.charm.app,
+                self.relation_data.component,
                 REQ_SECRET_FIELDS,
                 self.relation_data.secret_fields,  # pyright: ignore [reportAttributeAccessIssue]
             )
@@ -1508,13 +1742,15 @@ def _on_secret_changed_event(self, event: RelationChangedEvent) -> None:
         raise NotImplementedError
 
 
-# Base DataPeer
+################################################################################
+# Peer Relation Data
+################################################################################
 
 
 class DataPeerData(RequirerData, ProviderData):
     """Represents peer relations data."""
 
-    SECRET_FIELDS = ["operator-password"]
+    SECRET_FIELDS = []
     SECRET_FIELD_NAME = "internal_secret"
     SECRET_LABEL_MAP = {}
 
@@ -1524,6 +1760,7 @@ def __init__(
         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,
     ):
@@ -1537,6 +1774,19 @@ def __init__(
         )
         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 = []
+        self._additional_secret_group_mapping = additional_secret_group_mapping
+
+        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]:
@@ -1546,15 +1796,232 @@ 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 = []
+
+        ignores = [SECRET_GROUPS.get_group("user"), SECRET_GROUPS.get_group("tls")]
+        for group in SECRET_GROUPS.groups():
+            if group in ignores:
+                continue
+            if content := self._get_group_secret_contents(relation, group):
+                fields += list(content.keys())
+        return list(set(fields) | set(self._new_secrets))
+
+    @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 self.secrets_enabled and full_field not in self.current_secret_fields:
+            self._new_secrets.append(full_field)
+        if self._no_group_with_databag(field, 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
+    @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 (
+            self.secrets_enabled
+            and full_field not in self.current_secret_fields
+            and field not in self.current_secret_fields
+        ):
+            return
+        if self._no_group_with_databag(field, full_field):
+            return self.fetch_my_relation_field(relation_id, full_field)
+
+    @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 self.secrets_enabled and full_field not in self.current_secret_fields:
+            logger.warning(f"Secret {field} from group {group_mapping} was not found")
+            return
+        if self._no_group_with_databag(field, full_field):
+            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
+        }
+
+    # Backwards compatibility
+
+    def _check_deleted_label(self, relation, fields) -> None:
+        """Helper function for legacy behavior."""
+        current_data = self.fetch_my_relation_data([relation.id], fields)
+        if current_data is not None:
+            # Check if the secret we wanna delete actually exists
+            # Given the "deleted label", here we can't rely on the default mechanism (i.e. 'key not found')
+            if non_existent := (set(fields) & set(self.secret_fields)) - set(
+                current_data.get(relation.id, [])
+            ):
+                logger.debug(
+                    "Non-existing secret %s was attempted to be removed.",
+                    ", ".join(non_existent),
+                )
+
+    def _remove_secret_from_databag(self, relation, fields: List[str]) -> None:
+        """For Rolling Upgrades -- when moving from databag to secrets usage.
+
+        Practically what happens here is to remove stuff from the databag that is
+        to be stored in secrets.
+        """
+        if not self.secret_fields:
+            return
+
+        secret_fields_passed = set(self.secret_fields) & set(fields)
+        for field in secret_fields_passed:
+            if self._fetch_relation_data_without_secrets(self.component, relation, [field]):
+                self._delete_relation_data_without_secrets(self.component, relation, [field])
+
+    def _remove_secret_field_name_from_databag(self, relation) -> None:
+        """Making sure that the old databag URI is gone.
+
+        This action should not be executed more than once.
+        """
+        # Nothing to do if 'internal-secret' is not in the databag
+        if not (relation.data[self.component].get(self._generate_secret_field_name())):
+            return
+
+        # Making sure that the secret receives its label
+        # (This should have happened by the time we get here, rather an extra security measure.)
+        secret = self._get_relation_secret(relation.id)
+
+        # Either app scope secret with leader executing, or unit scope secret
+        leader_or_unit_scope = self.component != self.local_app or self.local_unit.is_leader()
+        if secret and leader_or_unit_scope:
+            # Databag reference to the secret URI can be removed, now that it's labelled
+            relation.data[self.component].pop(self._generate_secret_field_name(), None)
+
+    def _previous_labels(self) -> List[str]:
+        """Generator for legacy secret label names, for backwards compatibility."""
+        result = []
+        members = [self._model.app.name]
+        if self.scope:
+            members.append(self.scope.value)
+        result.append(f"{'.'.join(members)}")
+        return result
+
+    def _no_group_with_databag(self, field: str, full_field: str) -> bool:
+        """Check that no secret group is attempted to be used together with databag."""
+        if not self.secrets_enabled and full_field != field:
+            logger.error(
+                f"Can't access {full_field}: no secrets available (i.e. no secret groups either)."
+            )
+            return False
+        return True
+
+    # Event handlers
+
+    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
+
+    # Overrides of Relation Data handling functions
+
     def _generate_secret_label(
         self, relation_name: str, relation_id: int, group_mapping: SecretGroup
     ) -> str:
-        members = [self._model.app.name]
+        members = [relation_name, 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}"
 
@@ -1562,7 +2029,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.
@@ -1581,51 +2048,29 @@ def _get_relation_secret(
         label = self._generate_secret_label(relation_name, relation_id, group_mapping)
         secret_uri = relation.data[self.component].get(self._generate_secret_field_name(), None)
 
-        # Fetching the secret with fallback to URI (in case label is not yet known)
-        # Label would we "stuck" on the secret in case it is found
-        secret = self.secrets.get(label, secret_uri)
-
-        # Either app scope secret with leader executing, or unit scope secret
-        leader_or_unit_scope = self.component != self.local_app or self.local_unit.is_leader()
-        if secret_uri and secret and leader_or_unit_scope:
-            # Databag reference to the secret URI can be removed, now that it's labelled
-            relation.data[self.component].pop(self._generate_secret_field_name(), None)
-        return secret
+        # URI or legacy label is only to applied when moving single legacy secret to a (new) label
+        if group_mapping == SECRET_GROUPS.EXTRA:
+            # Fetching the secret with fallback to URI (in case label is not yet known)
+            # Label would we "stuck" on the secret in case it is found
+            return self.secrets.get(label, secret_uri, legacy_labels=self._previous_labels())
+        return self.secrets.get(label)
 
     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}
-
-    def _remove_secret_from_databag(self, relation, fields: List[str]) -> None:
-        """For Rolling Upgrades -- when moving from databag to secrets usage.
-
-        Practically what happens here is to remove stuff from the databag that is
-        to be stored in secrets.
-        """
-        if not self.secret_fields:
-            return
-
-        secret_fields_passed = set(self.secret_fields) & set(fields)
-        for field in secret_fields_passed:
-            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
-        )
+        if self.deleted_label:
+            result = {key: result[key] for key in result if result[key] != self.deleted_label}
+        if self._additional_secret_group_mapping:
+            return {self._field_to_internal_name(key, group): result[key] for key in result}
+        return result
 
+    @either_static_or_dynamic_secrets
     def _fetch_my_specific_relation_data(
         self, relation: Relation, fields: Optional[List[str]]
     ) -> Dict[str, str]:
@@ -1634,6 +2079,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()))
@@ -1645,24 +2091,17 @@ def _update_relation_data(self, relation: Relation, data: Dict[str, str]) -> Non
             data=data,
             uri_to_databag=False,
         )
+        self._remove_secret_field_name_from_databag(relation)
 
         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:
-            current_data = self.fetch_my_relation_data([relation.id], fields)
-            if current_data is not None:
-                # Check if the secret we wanna delete actually exists
-                # Given the "deleted label", here we can't rely on the default mechanism (i.e. 'key not found')
-                if non_existent := (set(fields) & set(self.secret_fields)) - set(
-                    current_data.get(relation.id, [])
-                ):
-                    logger.error(
-                        "Non-existing secret %s was attempted to be removed.",
-                        ", ".join(non_existent),
-                    )
+            # Legacy, backwards compatibility
+            self._check_deleted_label(relation, fields)
 
             _, normal_fields = self._process_secret_fields(
                 relation,
@@ -1704,7 +2143,7 @@ def fetch_relation_field(
     fetch_my_relation_field = Data.fetch_my_relation_field
 
 
-class DataPeerEventHandlers(EventHandlers):
+class DataPeerEventHandlers(RequirerEventHandlers):
     """Requires-side of the relation."""
 
     def __init__(self, charm: CharmBase, relation_data: RequirerData, unique_key: str = ""):
@@ -1729,6 +2168,7 @@ def __init__(
         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 = "",
@@ -1739,6 +2179,7 @@ def __init__(
             relation_name,
             extra_user_roles,
             additional_secret_fields,
+            additional_secret_group_mapping,
             secret_field_name,
             deleted_label,
         )
@@ -1763,6 +2204,7 @@ def __init__(
         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 = "",
@@ -1773,6 +2215,7 @@ def __init__(
             relation_name,
             extra_user_roles,
             additional_secret_fields,
+            additional_secret_group_mapping,
             secret_field_name,
             deleted_label,
         )
@@ -1787,6 +2230,14 @@ def __init__(self, unit: Unit, *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."""
@@ -1807,23 +2258,29 @@ def __init__(
         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__(
+        DataPeerOtherUnitData.__init__(
             self,
+            unit,
             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)
+        DataPeerOtherUnitEventHandlers.__init__(self, charm, self)
+
 
+################################################################################
+# Cross-charm Relatoins Data Handling and Evenets
+################################################################################
 
-# General events
+# Generic events
 
 
 class ExtraRoleEvent(RelationEvent):
@@ -2390,7 +2847,7 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
 
         # Check if the database is created
         # (the database charm shared the credentials).
-        secret_field_user = self.relation_data._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:
@@ -2462,7 +2919,11 @@ def __init__(
         DatabaseRequirerEventHandlers.__init__(self, charm, self)
 
 
-# Kafka related events
+################################################################################
+# Charm-specific Relations Data and Events
+################################################################################
+
+# Kafka Events
 
 
 class KafkaProvidesEvent(RelationEvent):
@@ -2704,7 +3165,7 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
         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.relation_data._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:
@@ -2949,8 +3410,8 @@ def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
         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.relation_data._generate_secret_field_name(SecretGroup.USER)
-        secret_field_tls = self.relation_data._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())
diff --git a/src/charm.py b/src/charm.py
index 837917b96e..dd9a3b8fca 100755
--- a/src/charm.py
+++ b/src/charm.py
@@ -14,7 +14,7 @@
 from typing import Dict, List, Literal, Optional, Set, get_args
 
 import psycopg2
-from charms.data_platform_libs.v0.data_interfaces import DataPeer, DataPeerUnit
+from charms.data_platform_libs.v0.data_interfaces import DataPeerData, DataPeerUnitData
 from charms.data_platform_libs.v0.data_models import TypedCharmBase
 from charms.grafana_agent.v0.cos_agent import COSAgentProvider
 from charms.operator_libs_linux.v2 import snap
@@ -118,28 +118,15 @@ def __init__(self, *args):
             self.unit.status = BlockedStatus("Disabled")
             sys.exit(0)
 
-        self.peer_relation_app = DataPeer(
-            self,
+        self.peer_relation_app = DataPeerData(
+            self.model,
             relation_name=PEER,
-            additional_secret_fields=[
-                "monitoring-password",
-                "operator-password",
-                "replication-password",
-                "rewind-password",
-            ],
             secret_field_name=SECRET_INTERNAL_LABEL,
             deleted_label=SECRET_DELETED_LABEL,
         )
-        self.peer_relation_unit = DataPeerUnit(
-            self,
+        self.peer_relation_unit = DataPeerUnitData(
+            self.model,
             relation_name=PEER,
-            additional_secret_fields=[
-                "key",
-                "csr",
-                "cauth",
-                "cert",
-                "chain",
-            ],
             secret_field_name=SECRET_INTERNAL_LABEL,
             deleted_label=SECRET_DELETED_LABEL,
         )
@@ -246,6 +233,13 @@ def _scope_obj(self, scope: Scopes):
         if scope == UNIT_SCOPE:
             return self.unit
 
+    def peer_relation_data(self, scope: Scopes) -> DataPeerData:
+        """Returns the peer relation data per scope."""
+        if scope == APP_SCOPE:
+            return self.peer_relation_app
+        elif scope == UNIT_SCOPE:
+            return self.peer_relation_unit
+
     def _translate_field_to_secret_key(self, key: str) -> str:
         """Change 'key' to secrets-compatible key field."""
         if not JujuVersion.from_environ().has_secrets:
@@ -260,12 +254,16 @@ def get_secret(self, scope: Scopes, key: str) -> Optional[str]:
             raise RuntimeError("Unknown secret scope.")
 
         peers = self.model.get_relation(PEER)
+        if not peers:
+            return None
         secret_key = self._translate_field_to_secret_key(key)
-        if scope == APP_SCOPE:
-            value = self.peer_relation_app.fetch_my_relation_field(peers.id, secret_key)
-        else:
-            value = self.peer_relation_unit.fetch_my_relation_field(peers.id, secret_key)
-        return value
+        # Old translation in databag is to be taken
+        if key != secret_key and (
+            result := self.peer_relation_data(scope).fetch_my_relation_field(peers.id, key)
+        ):
+            return result
+
+        return self.peer_relation_data(scope).get_secret(peers.id, secret_key)
 
     def set_secret(self, scope: Scopes, key: str, value: Optional[str]) -> Optional[str]:
         """Set secret from the secret storage."""
@@ -277,10 +275,12 @@ def set_secret(self, scope: Scopes, key: str, value: Optional[str]) -> Optional[
 
         peers = self.model.get_relation(PEER)
         secret_key = self._translate_field_to_secret_key(key)
-        if scope == APP_SCOPE:
-            self.peer_relation_app.update_relation_data(peers.id, {secret_key: value})
-        else:
-            self.peer_relation_unit.update_relation_data(peers.id, {secret_key: value})
+        # Old translation in databag is to be deleted
+        if key != secret_key and self.peer_relation_data(scope).fetch_my_relation_field(
+            peers.id, key
+        ):
+            self.peer_relation_data(scope).delete_relation_data(peers.id, [key])
+        self.peer_relation_data(scope).set_secret(peers.id, secret_key, value)
 
     def remove_secret(self, scope: Scopes, key: str) -> None:
         """Removing a secret."""
diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py
index 057868fc71..3a4bb9d718 100644
--- a/tests/unit/conftest.py
+++ b/tests/unit/conftest.py
@@ -12,3 +12,20 @@
 def _has_secrets(request, monkeypatch):
     monkeypatch.setattr("charm.JujuVersion.has_secrets", PropertyMock(return_value=request.param))
     return request.param
+
+
+@pytest.fixture
+def only_with_juju_secrets(_has_secrets):
+    """Pretty way to skip Juju 3 tests."""
+    if not _has_secrets:
+        pytest.skip("Secrets test only applies on Juju 3.x")
+
+
+@pytest.fixture
+def only_without_juju_secrets(_has_secrets):
+    """Pretty way to skip Juju 2-specific tests.
+
+    Typically: to save CI time, when the same check were executed in a Juju 3-specific way already
+    """
+    if _has_secrets:
+        pytest.skip("Skipping legacy secrets tests")
diff --git a/tests/unit/test_charm.py b/tests/unit/test_charm.py
index a4b5bb649b..456784f0fb 100644
--- a/tests/unit/test_charm.py
+++ b/tests/unit/test_charm.py
@@ -1719,24 +1719,30 @@ def test_scope_obj(harness):
 
 
 @patch_network_get(private_address="1.1.1.1")
-def test_get_secret(harness):
+def test_get_secret_from_databag(harness):
+    """Asserts that get_secret method can read secrets from databag.
+
+    This must be backwards-compatible so it runs on both juju2 and juju3.
+    """
     with patch("charm.PostgresqlOperatorCharm._on_leader_elected"):
         rel_id = harness.model.get_relation(PEER).id
         # App level changes require leader privileges
         harness.set_leader()
         # Test application scope.
-        assert harness.charm.get_secret("app", "password") is None
-        harness.update_relation_data(rel_id, harness.charm.app.name, {"password": "test-password"})
-        assert harness.charm.get_secret("app", "password") == "test-password"
+        assert harness.charm.get_secret("app", "operator_password") is None
+        harness.update_relation_data(
+            rel_id, harness.charm.app.name, {"operator_password": "test-password"}
+        )
+        assert harness.charm.get_secret("app", "operator_password") == "test-password"
 
         # Unit level changes don't require leader privileges
         harness.set_leader(False)
         # Test unit scope.
-        assert harness.charm.get_secret("unit", "password") is None
+        assert harness.charm.get_secret("unit", "operator_password") is None
         harness.update_relation_data(
-            rel_id, harness.charm.unit.name, {"password": "test-password"}
+            rel_id, harness.charm.unit.name, {"operator_password": "test-password"}
         )
-        assert harness.charm.get_secret("unit", "password") == "test-password"
+        assert harness.charm.get_secret("unit", "operator_password") == "test-password"
 
 
 @patch_network_get(private_address="1.1.1.1")
@@ -1769,21 +1775,25 @@ def test_on_get_password_secrets(harness):
         mock_event.set_results.assert_called_once_with({"password": "replication-test-password"})
 
 
-@pytest.mark.parametrize("scope", [("app"), ("unit")])
+@pytest.mark.parametrize("scope,field", [("app", "operator-password"), ("unit", "csr")])
 @patch_network_get(private_address="1.1.1.1")
-def test_get_secret_secrets(harness, scope):
+def test_get_secret_secrets(harness, scope, field):
     with (
         patch("charm.PostgresqlOperatorCharm._on_leader_elected"),
     ):
         harness.set_leader()
 
-        assert harness.charm.get_secret(scope, "operator-password") is None
-        harness.charm.set_secret(scope, "operator-password", "test-password")
-        assert harness.charm.get_secret(scope, "operator-password") == "test-password"
+        assert harness.charm.get_secret(scope, field) is None
+        harness.charm.set_secret(scope, field, "test")
+        assert harness.charm.get_secret(scope, field) == "test"
 
 
 @patch_network_get(private_address="1.1.1.1")
-def test_set_secret(harness):
+def test_set_secret_in_databag(harness, only_without_juju_secrets):
+    """Asserts that set_secret method writes to relation databag.
+
+    This is juju2 specific. In juju3, set_secret writes to juju secrets.
+    """
     with patch("charm.PostgresqlOperatorCharm._on_leader_elected"):
         rel_id = harness.model.get_relation(PEER).id
         harness.set_leader()
@@ -1843,7 +1853,7 @@ def test_invalid_secret(harness, scope, is_leader):
         # App has to be leader, unit can be either
         harness.set_leader(is_leader)
 
-        with pytest.raises(RelationDataTypeError):
+        with pytest.raises((RelationDataTypeError, TypeError)):
             harness.charm.set_secret(scope, "somekey", 1)
 
         harness.charm.set_secret(scope, "somekey", "")
@@ -1867,7 +1877,7 @@ def test_delete_password(harness, _has_secrets, caplog):
         assert harness.charm.get_secret("unit", "operator-password") is None
 
         harness.set_leader(True)
-        with caplog.at_level(logging.ERROR):
+        with caplog.at_level(logging.DEBUG):
             if _has_secrets:
                 error_message = (
                     "Non-existing secret operator-password was attempted to be removed."
@@ -1898,28 +1908,26 @@ def test_delete_password(harness, _has_secrets, caplog):
 
 @pytest.mark.parametrize("scope,is_leader", [("app", True), ("unit", True), ("unit", False)])
 @patch_network_get(private_address="1.1.1.1")
-def test_migration_from_databag(harness, _has_secrets, scope, is_leader):
-    """Check if we're moving on to use secrets when live upgrade from databag to Secrets usage."""
+def test_migration_from_databag(harness, only_with_juju_secrets, scope, is_leader):
+    """Check if we're moving on to use secrets when live upgrade from databag to Secrets usage.
+
+    Since it checks for a migration from databag to juju secrets, it's specific to juju3.
+    """
     with (
         patch("charm.PostgresqlOperatorCharm._on_leader_elected"),
     ):
-        # as this test checks for a migration from databag to secrets,
-        # there's no need for this test when secrets are not enabled.
-        if not _has_secrets:
-            return
-
         rel_id = harness.model.get_relation(PEER).id
         # App has to be leader, unit can be either
         harness.set_leader(is_leader)
 
         # Getting current password
         entity = getattr(harness.charm, scope)
-        harness.update_relation_data(rel_id, entity.name, {"operator-password": "bla"})
-        assert harness.charm.get_secret(scope, "operator-password") == "bla"
+        harness.update_relation_data(rel_id, entity.name, {"operator_password": "bla"})
+        assert harness.charm.get_secret(scope, "operator_password") == "bla"
 
         # Reset new secret
         harness.charm.set_secret(scope, "operator-password", "blablabla")
-        assert harness.charm.model.get_secret(label=f"postgresql.{scope}")
+        assert harness.charm.model.get_secret(label=f"{PEER}.postgresql.{scope}")
         assert harness.charm.get_secret(scope, "operator-password") == "blablabla"
         assert "operator-password" not in harness.get_relation_data(
             rel_id, getattr(harness.charm, scope).name
@@ -1928,16 +1936,14 @@ def test_migration_from_databag(harness, _has_secrets, scope, is_leader):
 
 @pytest.mark.parametrize("scope,is_leader", [("app", True), ("unit", True), ("unit", False)])
 @patch_network_get(private_address="1.1.1.1")
-def test_migration_from_single_secret(harness, _has_secrets, scope, is_leader):
-    """Check if we're moving on to use secrets when live upgrade from databag to Secrets usage."""
+def test_migration_from_single_secret(harness, only_with_juju_secrets, scope, is_leader):
+    """Check if we're moving on to use secrets when live upgrade from databag to Secrets usage.
+
+    Since it checks for a migration from databag to juju secrets, it's specific to juju3.
+    """
     with (
         patch("charm.PostgresqlOperatorCharm._on_leader_elected"),
     ):
-        # as this test checks for a migration from databag to secrets,
-        # there's no need for this test when secrets are not enabled.
-        if not _has_secrets:
-            return
-
         rel_id = harness.model.get_relation(PEER).id
 
         # App has to be leader, unit can be either
@@ -1957,7 +1963,7 @@ def test_migration_from_single_secret(harness, _has_secrets, scope, is_leader):
         harness.charm.set_secret(scope, "operator-password", "blablabla")
         with harness.hooks_disabled():
             harness.set_leader(is_leader)
-        assert harness.charm.model.get_secret(label=f"postgresql.{scope}")
+        assert harness.charm.model.get_secret(label=f"{PEER}.postgresql.{scope}")
         assert harness.charm.get_secret(scope, "operator-password") == "blablabla"
         assert SECRET_INTERNAL_LABEL not in harness.get_relation_data(
             rel_id, getattr(harness.charm, scope).name
diff --git a/tests/unit/test_db.py b/tests/unit/test_db.py
index b914db6841..e241adab99 100644
--- a/tests/unit/test_db.py
+++ b/tests/unit/test_db.py
@@ -443,7 +443,6 @@ def test_update_endpoints_with_relation(harness):
             side_effect="postgresql/0",
         ) as _get_state,
     ):
-        peer_rel_id = harness.model.get_relation(PEER).id
         # Set some side effects to test multiple situations.
         postgresql_mock.get_postgresql_version = PropertyMock(
             side_effect=[
@@ -477,23 +476,8 @@ def test_update_endpoints_with_relation(harness):
         # Set some required data before update_endpoints is called.
         for rel in [rel_id, another_rel_id]:
             user = f"relation-{rel}"
-            harness.update_relation_data(
-                rel,
-                harness.charm.app.name,
-                {
-                    "user": user,
-                    "password": password,
-                    "database": DATABASE,
-                },
-            )
-            harness.update_relation_data(
-                peer_rel_id,
-                harness.charm.app.name,
-                {
-                    user: password,
-                    f"{user}-database": DATABASE,
-                },
-            )
+            harness.charm.set_secret("app", user, password)
+            harness.charm.set_secret("app", f"{user}-database", DATABASE)
 
         # BlockedStatus due to a PostgreSQLGetPostgreSQLVersionError.
         harness.charm.legacy_db_relation.update_endpoints(relation)
@@ -560,7 +544,6 @@ def test_update_endpoints_without_relation(harness):
         ) as _get_state,
     ):
         # Set some side effects to test multiple situations.
-        peer_rel_id = harness.model.get_relation(PEER).id
         postgresql_mock.get_postgresql_version = PropertyMock(
             side_effect=[
                 PostgreSQLGetPostgreSQLVersionError,
@@ -590,23 +573,8 @@ def test_update_endpoints_without_relation(harness):
         # Set some required data before update_endpoints is called.
         for rel in [rel_id, another_rel_id]:
             user = f"relation-{rel}"
-            harness.update_relation_data(
-                rel,
-                harness.charm.app.name,
-                {
-                    "user": user,
-                    "password": password,
-                    "database": DATABASE,
-                },
-            )
-            harness.update_relation_data(
-                peer_rel_id,
-                harness.charm.app.name,
-                {
-                    user: password,
-                    f"{user}-database": DATABASE,
-                },
-            )
+            harness.charm.set_secret("app", user, password)
+            harness.charm.set_secret("app", f"{user}-database", DATABASE)
 
         # BlockedStatus due to a PostgreSQLGetPostgreSQLVersionError.
         harness.charm.legacy_db_relation.update_endpoints()