[MISC] Port MySQL v0 libs to Python 3.10+ (#660)

Author: sinclert-canonical

lib/charms/mysql/v0/async_replication.py

@@ -30,7 +30,6 @@
 )
 from ops.framework import Object
 from tenacity import RetryError, Retrying, stop_after_attempt, wait_fixed
-from typing_extensions import Optional
 
 from constants import (
     BACKUPS_PASSWORD_KEY,
@@ -54,7 +53,7 @@
 # The unique Charmhub library identifier, never change it
 LIBID = "4de21f1a022c4e2c87ac8e672ec16f6a"
 LIBAPI = 0
-LIBPATCH = 8
+LIBPATCH = 9
 
 RELATION_OFFER = "replication-offer"
 RELATION_CONSUMER = "replication"
@@ -126,21 +125,21 @@ def cluster_set_name(self) -> str:
         return self._charm.app_peer_data["cluster-set-domain-name"]
 
     @property
-    def relation(self) -> Optional[Relation]:
+    def relation(self) -> Relation | None:
         """Relation."""
         return self.model.get_relation(RELATION_OFFER) or self.model.get_relation(
             RELATION_CONSUMER
         )
 
     @property
-    def relation_data(self) -> Optional[RelationDataContent]:
+    def relation_data(self) -> RelationDataContent | None:
         """Relation data."""
         if not self.relation:
             return
         return self.relation.data[self.model.app]
 
     @property
-    def remote_relation_data(self) -> Optional[RelationDataContent]:
+    def remote_relation_data(self) -> RelationDataContent | None:
         """Remote relation data."""
         if not self.relation or not self.relation.app:
             return
@@ -362,7 +361,7 @@ def __init__(self, charm: "MySQLOperatorCharm"):
         self.framework.observe(self._charm.on.secret_changed, self._on_secret_change)
 
     @property
-    def state(self) -> Optional[States]:
+    def state(self) -> States | None:
         """State of the relation, on primary side."""
         if not self.relation:
             return States.UNINITIALIZED
@@ -413,7 +412,7 @@ def idle(self) -> bool:
         return True
 
     @property
-    def secret(self) -> Optional[Secret]:
+    def secret(self) -> Secret | None:
         """Return the async replication secret."""
         if not self.relation:
             return
@@ -642,7 +641,7 @@ def __init__(self, charm: "MySQLOperatorCharm"):
         self.framework.observe(self._charm.on.secret_changed, self._on_secret_change)
 
     @property
-    def state(self) -> Optional[States]:
+    def state(self) -> States | None:
         """State of the relation, on consumer side."""
         if not self.relation:
             return None
@@ -719,7 +718,7 @@ def _async_replication_credentials(self) -> dict[str, str]:
         secret = self._obtain_secret()
         return secret.peek_content()
 
-    def _get_endpoint(self) -> Optional[str]:
+    def _get_endpoint(self) -> str | None:
         """Get endpoint to be used by the primary cluster.
 
         This is the address in which the unit must be reachable from the primary cluster.

lib/charms/mysql/v0/backups.py

@@ -50,7 +50,6 @@ def is_unit_blocked(self) -> bool:
 import pathlib
 import re
 import typing
-from typing import Dict, List, Optional, Tuple
 
 from charms.data_platform_libs.v0.s3 import (
     CredentialsChangedEvent,
@@ -113,7 +112,7 @@ def is_unit_blocked(self) -> bool:
 
 # Increment this PATCH version before using `charmcraft publish-lib` or reset
 # to 0 if you are raising the major API version
-LIBPATCH = 14
+LIBPATCH = 15
 
 ANOTHER_S3_CLUSTER_REPOSITORY_ERROR_MESSAGE = "S3 repository claimed by another cluster"
 MOVE_RESTORED_CLUSTER_TO_ANOTHER_S3_REPOSITORY_ERROR = (
@@ -150,7 +149,7 @@ def _s3_integrator_relation_exists(self) -> bool:
         """Returns whether a relation with the s3-integrator exists."""
         return bool(self.model.get_relation(S3_INTEGRATOR_RELATION_NAME))
 
-    def _retrieve_s3_parameters(self) -> Tuple[Dict[str, str], List[str]]:
+    def _retrieve_s3_parameters(self) -> tuple[dict[str, str], list[str]]:
         """Retrieve S3 parameters from the S3 integrator relation.
 
         Returns: tuple of (s3_parameters, missing_required_parameters)
@@ -196,7 +195,7 @@ def _upload_logs_to_s3(
         stdout: str,
         stderr: str,
         log_filename: str,
-        s3_parameters: Dict[str, str],
+        s3_parameters: dict[str, str],
     ) -> bool:
         """Upload logs to S3 at the specified location.
 
@@ -219,7 +218,7 @@ def _upload_logs_to_s3(
     # ------------------ List Backups ------------------
 
     @staticmethod
-    def _format_backups_list(backup_list: List[Tuple[str, str]]) -> str:
+    def _format_backups_list(backup_list: list[tuple[str, str]]) -> str:
         """Formats the provided list of backups as a table."""
         backups = [f"{'backup-id':<21} | {'backup-type':<12} | backup-status"]
 
@@ -359,7 +358,7 @@ def _on_create_backup(self, event: ActionEvent) -> None:
         })
         self.charm._on_update_status(None)
 
-    def _can_unit_perform_backup(self) -> Tuple[bool, Optional[str]]:
+    def _can_unit_perform_backup(self) -> tuple[bool, str | None]:
         """Validates whether this unit can perform a backup.
 
         Returns: tuple of (success, error_message)
@@ -390,7 +389,7 @@ def _can_unit_perform_backup(self) -> Tuple[bool, Optional[str]]:
 
         return True, None
 
-    def _pre_backup(self) -> Tuple[bool, Optional[str]]:
+    def _pre_backup(self) -> tuple[bool, str | None]:
         """Runs operations required before performing a backup.
 
         Returns: tuple of (success, error_message)
@@ -415,7 +414,7 @@ def _pre_backup(self) -> Tuple[bool, Optional[str]]:
 
         return True, None
 
-    def _backup(self, backup_path: str, s3_parameters: Dict) -> Tuple[bool, Optional[str]]:
+    def _backup(self, backup_path: str, s3_parameters: dict) -> tuple[bool, str | None]:
         """Runs the backup operations.
 
         Args:
@@ -450,7 +449,7 @@ def _backup(self, backup_path: str, s3_parameters: Dict) -> Tuple[bool, Optional
 
         return True, None
 
-    def _post_backup(self) -> Tuple[bool, Optional[str]]:
+    def _post_backup(self) -> tuple[bool, str | None]:
         """Runs operations required after performing a backup.
 
         Returns: tuple of (success, error_message)
@@ -613,7 +612,7 @@ def _on_restore(self, event: ActionEvent) -> None:  # noqa: C901
         # update status as soon as possible
         self.charm._on_update_status(None)
 
-    def _pre_restore(self) -> Tuple[bool, str]:
+    def _pre_restore(self) -> tuple[bool, str]:
         """Perform operations that need to be done before performing a restore.
 
         Returns: tuple of (success, error_message)
@@ -635,7 +634,7 @@ def _pre_restore(self) -> Tuple[bool, str]:
 
         return True, ""
 
-    def _restore(self, backup_id: str, s3_parameters: Dict[str, str]) -> Tuple[bool, bool, str]:
+    def _restore(self, backup_id: str, s3_parameters: dict[str, str]) -> tuple[bool, bool, str]:
         """Run the restore operations.
 
         Args:
@@ -687,7 +686,7 @@ def _restore(self, backup_id: str, s3_parameters: Dict[str, str]) -> Tuple[bool,
 
         return True, True, ""
 
-    def _clean_data_dir_and_start_mysqld(self) -> Tuple[bool, str]:
+    def _clean_data_dir_and_start_mysqld(self) -> tuple[bool, str]:
         """Run idempotent operations run after restoring a backup.
 
         Returns tuple of (success, error_message)
@@ -711,8 +710,8 @@ def _clean_data_dir_and_start_mysqld(self) -> Tuple[bool, str]:
         return True, ""
 
     def _pitr_restore(
-        self, restore_to_time: str, s3_parameters: Dict[str, str]
-    ) -> Tuple[bool, str]:
+        self, restore_to_time: str, s3_parameters: dict[str, str]
+    ) -> tuple[bool, str]:
         try:
             logger.info("Restoring point-in-time-recovery")
             stdout, stderr = self.charm._mysql.restore_pitr(
@@ -728,7 +727,7 @@ def _pitr_restore(
             return False, f"Failed to restore point-in-time-recovery to the {restore_to_time}"
         return True, ""
 
-    def _post_restore(self) -> Tuple[bool, str]:
+    def _post_restore(self) -> tuple[bool, str]:
         """Run operations required after restoring a backup.
 
         Returns: tuple of (success, error_message)
@@ -836,7 +835,7 @@ def _on_s3_credentials_gone(self, event: CredentialsGoneEvent) -> None:
                 "Exception is occurred when trying to stop binlogs collecting after S3 relation depart. It may be a leader departure"
             )
 
-    def get_binlogs_collector_config(self) -> Dict[str, str]:
+    def get_binlogs_collector_config(self) -> dict[str, str]:
         """Return binlogs collector service config file.
 
         Returns: dict of binlogs collector service config