Something like this

lib/charms/data_platform_libs/v0/data_interfaces.py

@@ -1481,13 +1481,19 @@ def __init__(
         additional_secret_fields: Optional[List[str]] = [],
         secret_field_name: Optional[str] = None,
         deleted_label: Optional[str] = None,
+        # This parameter is specifically set up for charms that had `ca`-like,
+        # Juju Secrets incompatible fields. Don't use this parameter except if you
+        # EXPLICITLY target this case.
+        field_translations: Dict[str, str] = {},
     ):
         """Manager of base client relations."""
         DataRequires.__init__(
             self, charm, 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.field_translations = field_translations
+        self._secret_fields = self._translate(self._secret_fields)
 
     @property
     def scope(self) -> Optional[Scope]:
@@ -1497,6 +1503,8 @@ def scope(self) -> Optional[Scope]:
         if isinstance(self.component, Unit):
             return Scope.UNIT
 
+    # No event handlers needed
+
     def _on_relation_changed_event(self, event: RelationChangedEvent) -> None:
         """Event emitted when the relation has changed."""
         pass
@@ -1505,6 +1513,90 @@ def _on_secret_changed_event(self, event: SecretChangedEvent) -> None:
         """Event emitted when the secret has changed."""
         pass
 
+    # Backwards compatibility functions
+
+    def _translate(
+        self, fields: Union[List[str], Dict[str, str]], relation: Optional[Relation] = None
+    ) -> List[str]:
+        """Backwards compatibility function to deal with legacy secret fields naming.
+
+        Translate (typically incompatible, old) fields to Juju Secrets compatible fields.
+        We only perform a translation, if the old field is not in use in the databag anymore.
+        Replacement of old field name to new field name may happen either within a list or on
+        dictionary keys.
+        """
+        # No action taken if no translations specified or if no "old" fields were requested
+        if not self.field_translations or not (
+            impacted := set(self.field_translations) & set(fields)
+        ):
+            return fields
+
+        translated = fields.copy()
+        for old in impacted:
+            new = self.field_translations[old]
+
+            # We translate old field to new secret field, if the old field is not in use in the databag
+            if relation and self._fetch_relation_data_without_secrets(
+                self.component, relation, [old]
+            ):
+                continue
+
+            if isinstance(fields, list):
+                translated.append(new)
+                translated.remove(old)
+            elif isinstance(fields, dict):
+                translated[new] = translated.pop(old)
+        return translated
+
+    def _reverse_translate(self, fields: Set[str], content: Dict[str, str]):
+        """Switch new field name keys to their old correspondant in a dictionary."""
+        if not self.field_translations or not (
+            impacted := set(content) & set(self.field_translations.values()) & fields
+        ):
+            return content
+
+        inverted = {value: key for key, value in self.field_translations.items()}
+        for new_key in impacted:
+            content[inverted[new_key]] = content.pop(new_key)
+        return content
+
+    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.
+        Reason: Typically when performing rolling upgrades from databag to secrets
+        """
+        if not self.secret_fields:
+            return
+
+        translated_fields = self._translate(fields, relation)
+        for translated_field in translated_fields:
+            if self._fetch_relation_data_without_secrets(
+                self.component, relation, [translated_field]
+            ):
+                self._delete_relation_data_without_secrets(
+                    self.component, relation, [translated_field]
+                )
+
+    def _verify_if_field_exists_when_deleted_label(self, relation: Relation, fields: List[str]) -> None:
+        """
+
+        """
+        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),
+                )
+
+    # Internal overrides of parents, allowing to take advantage of a unified process
+
     def _generate_secret_label(
         self, relation_name: str, relation_id: int, group_mapping: SecretGroup
     ) -> str:
@@ -1563,77 +1655,54 @@ def _get_group_secret_contents(
             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
-        )
-
     def _fetch_my_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 for owner/this_app."""
-        return self._fetch_relation_data_with_secrets(
-            self.component, self.secret_fields, relation, fields
+        translated_fields = self._translate(fields, relation)
+        result = self._fetch_relation_data_with_secrets(
+            self.component, self.secret_fields, relation, translated_fields
         )
 
+        if translated_fields == fields:
+            return result
+        return self._reverse_translate(set(translated_fields) - set(fields), result)
+
     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()))
+        translated_data = self._translate(data, relation)
         _, normal_fields = self._process_secret_fields(
             relation,
             self.secret_fields,
-            list(data),
+            list(translated_data),
             self._add_or_update_relation_secrets,
-            data=data,
+            data=translated_data,
             uri_to_databag=False,
         )
 
-        normal_content = {k: v for k, v in data.items() if k in normal_fields}
+        normal_content = {k: v for k, v in translated_data.items() if k in normal_fields}
         self._update_relation_data_without_secrets(self.component, relation, normal_content)
 
     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."""
+        translated_fields = self._translate(fields, relation)
         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),
-                    )
+            self._verify_if_field_exists_when_deleted_label(relation, translated_fields)
+            arguments = {
+                "operation": self._update_relation_secret,
+                "data": {field: self.deleted_label for field in translated_fields},
+            }
 
-            _, normal_fields = self._process_secret_fields(
-                relation,
-                self.secret_fields,
-                fields,
-                self._update_relation_secret,
-                data={field: self.deleted_label for field in fields},
-            )
         else:
-            _, normal_fields = self._process_secret_fields(
-                relation, self.secret_fields, fields, self._delete_relation_secret, fields=fields
-            )
+            arguments = {
+                "operation": self._delete_relation_secret,
+                "fields": translated_fields,
+            }
+
+        _, normal_fields = self._process_secret_fields(
+            relation, self.secret_fields, translated_fields, **arguments
+        )
         self._delete_relation_data_without_secrets(self.component, relation, list(normal_fields))
 
     def fetch_relation_data(

src/charm.py

@@ -78,22 +78,21 @@ def __init__(self, *args):
                 self._translate_field_to_secret_key(AUTH_FILE_DATABAG_KEY),
                 self._translate_field_to_secret_key(CFG_FILE_DATABAG_KEY),
                 self._translate_field_to_secret_key(MONITORING_PASSWORD_KEY),
+                # AUTH_FILE_DATABAG_KEY,
+                # CFG_FILE_DATABAG_KEY,
+                # MONITORING_PASSWORD_KEY,
             ],
             secret_field_name=SECRET_INTERNAL_LABEL,
             deleted_label=SECRET_DELETED_LABEL,
+            field_translations=SECRET_KEY_OVERRIDES,
         )
         self.peer_relation_unit = DataPeerUnit(
             self,
             relation_name=PEER_RELATION_NAME,
-            additional_secret_fields=[
-                "key",
-                "csr",
-                "cauth",
-                "cert",
-                "chain",
-            ],
+            additional_secret_fields=list(SECRET_KEY_OVERRIDES.keys()),
             secret_field_name=SECRET_INTERNAL_LABEL,
             deleted_label=SECRET_DELETED_LABEL,
+            field_translations=SECRET_KEY_OVERRIDES,
         )
 
         self.framework.observe(self.on.install, self._on_install)
@@ -284,8 +283,9 @@ def get_secret(self, scope: Scopes, key: str) -> Optional[str]:
             raise RuntimeError("Unknown secret scope.")
 
         peers = self.model.get_relation(PEER_RELATION_NAME)
-        secret_key = self._translate_field_to_secret_key(key)
-        return self.peer_relation_data(scope).fetch_my_relation_field(peers.id, secret_key)
+        # secret_key = self._translate_field_to_secret_key(key)
+        # return self.peer_relation_data(scope).fetch_my_relation_field(peers.id, secret_key)
+        return self.peer_relation_data(scope).fetch_my_relation_field(peers.id, key)
 
     def set_secret(self, scope: Scopes, key: str, value: Optional[str]) -> Optional[str]:
         """Set secret from the secret storage."""
@@ -296,17 +296,19 @@ def set_secret(self, scope: Scopes, key: str, value: Optional[str]) -> Optional[
             return self.remove_secret(scope, key)
 
         peers = self.model.get_relation(PEER_RELATION_NAME)
-        secret_key = self._translate_field_to_secret_key(key)
-        self.peer_relation_data(scope).update_relation_data(peers.id, {secret_key: value})
+        # secret_key = self._translate_field_to_secret_key(key)
+        # self.peer_relation_data(scope).update_relation_data(peers.id, {secret_key: value})
+        self.peer_relation_data(scope).update_relation_data(peers.id, {key: value})
 
     def remove_secret(self, scope: Scopes, key: str) -> None:
         """Removing a secret."""
         if scope not in get_args(Scopes):
             raise RuntimeError("Unknown secret scope.")
 
         peers = self.model.get_relation(PEER_RELATION_NAME)
-        secret_key = self._translate_field_to_secret_key(key)
-        self.peer_relation_data(scope).delete_relation_data(peers.id, [secret_key])
+        # secret_key = self._translate_field_to_secret_key(key)
+        # self.peer_relation_data(scope).delete_relation_data(peers.id, [secret_key])
+        self.peer_relation_data(scope).delete_relation_data(peers.id, [key])
 
     def _on_start(self, _) -> None:
         """On Start hook.

tests/unit/test_charm.py

@@ -539,6 +539,33 @@ def test_migration_from_databag(self, scope, is_leader, _):
             self.rel_id, getattr(self.charm, scope).name
         )
 
+    # @parameterized.expand([("unit", True), ("unit", False)])
+    @parameterized.expand([("app", True)])
+    @patch_network_get(private_address="1.1.1.1")
+    @patch("charm.JujuVersion.has_secrets", new_callable=PropertyMock, return_value=True)
+    def test_migration_from_databag_translated_field(self, scope, is_leader, _):
+        """Check if we're moving on to use secrets when live upgrade from databag to Secrets usage."""
+        # App has to be leader, unit can be either
+        with self.harness.hooks_disabled():
+            self.harness.set_leader(is_leader)
+
+        oldname = "cfg_file"
+        newname = "cfg-file"
+
+        # Getting current password
+        entity = getattr(self.charm, scope)
+        self.harness.update_relation_data(self.rel_id, entity.name, {oldname: "bla"})
+        assert self.harness.charm.get_secret(scope, oldname) == "bla"
+
+        # Reset new secret
+        self.harness.charm.set_secret(scope, oldname, "blablabla")
+        assert self.harness.charm.model.get_secret(label=f"pgbouncer.{scope}")
+        assert self.harness.charm.get_secret(scope, oldname) == "blablabla"
+        assert self.harness.charm.get_secret(scope, newname) == "blablabla"
+        assert oldname not in self.harness.get_relation_data(
+            self.rel_id, getattr(self.charm, scope).name
+        )
+
     @parameterized.expand([("app", True), ("unit", True), ("unit", False)])
     @patch_network_get(private_address="1.1.1.1")
     @patch("charm.JujuVersion.has_secrets", new_callable=PropertyMock, return_value=True)