Use python 3.10 syntax (#53)

Author: carlcsaposs-canonical

common/common/abstract_charm.py

@@ -4,9 +4,9 @@
 """MySQL Router charm"""
 
 import abc
+import collections.abc
 import dataclasses
 import logging
-import typing
 
 import charm_refresh
 import ops
@@ -99,7 +99,7 @@ def __init__(self, *args) -> None:
 
     @property
     @abc.abstractmethod
-    def _subordinate_relation_endpoint_names(self) -> typing.Optional[typing.Iterable[str]]:
+    def _subordinate_relation_endpoint_names(self) -> collections.abc.Collection[str] | None:
         """Subordinate relation endpoint names
 
         Does NOT include relations where charm is principal
@@ -117,7 +117,7 @@ def _logrotate(self) -> logrotate.LogRotate:
 
     @property
     @abc.abstractmethod
-    def _cos_relation_type(self) -> typing.Type[cos.COSRelation]:
+    def _cos_relation_type(self) -> type[cos.COSRelation]:
         """COSRelation type"""
 
     @abc.abstractmethod
@@ -129,22 +129,22 @@ def _read_only_endpoints(self, *, event) -> str:
         """MySQL Router read-only endpoint"""
 
     @abc.abstractmethod
-    def is_externally_accessible(self, *, event) -> typing.Optional[bool]:
+    def is_externally_accessible(self, *, event) -> bool | None:
         """Whether endpoints should be externally accessible.
 
         Only defined in vm charm to return True/False. In k8s charm, returns None.
         """
 
     @abc.abstractmethod
-    def tls_sans_ip(self, *, event) -> typing.Optional[typing.List[str]]:
+    def tls_sans_ip(self, *, event) -> list[str] | None:
         """TLS IP subject alternative names"""
 
     @abc.abstractmethod
-    def tls_sans_dns(self, *, event) -> typing.Optional[typing.List[str]]:
+    def tls_sans_dns(self, *, event) -> list[str] | None:
         """TLS DNS subject alternative names"""
 
     @abc.abstractmethod
-    def _status(self, *, event) -> typing.Optional[ops.StatusBase]:
+    def _status(self, *, event) -> ops.StatusBase | None:
         """Status of the charm."""
 
     @property
@@ -153,26 +153,26 @@ def _tls_certificate_saved(self) -> bool:
         return self.tls.certificate_saved
 
     @property
-    def _tls_key(self) -> typing.Optional[str]:
+    def _tls_key(self) -> str | None:
         """Custom TLS key"""
         return self.tls.key
 
     @property
-    def _tls_certificate_authority(self) -> typing.Optional[str]:
+    def _tls_certificate_authority(self) -> str | None:
         """Custom TLS certificate authority"""
         return self.tls.certificate_authority
 
     @property
-    def _tls_certificate(self) -> typing.Optional[str]:
+    def _tls_certificate(self) -> str | None:
         """Custom TLS certificate"""
         return self.tls.certificate
 
     @property
-    def tracing_endpoint(self) -> typing.Optional[str]:
+    def tracing_endpoint(self) -> str | None:
         """Otlp http endpoint for charm instrumentation."""
         return self._cos_relation.tracing_endpoint
 
-    def _cos_exporter_config(self, event) -> typing.Optional[cos.ExporterConfig]:
+    def _cos_exporter_config(self, event) -> cos.ExporterConfig | None:
         """Returns the exporter config for MySQLRouter exporter if cos relation exists"""
         cos_relation_exists = (
             self._cos_relation.relation_exists
@@ -203,8 +203,7 @@ def get_workload(self, *, event, refresh: charm_refresh.Common = None):
         )
 
     @staticmethod
-    # TODO python3.10 min version: Use `list` instead of `typing.List`
-    def _prioritize_statuses(statuses: typing.List[ops.StatusBase]) -> ops.StatusBase:
+    def _prioritize_statuses(statuses: list[ops.StatusBase]) -> ops.StatusBase:
         """Report the highest priority status.
 
         (Statuses of the same type are reported in the order they were added to `statuses`)

common/common/container.py

@@ -55,10 +55,7 @@ def rmtree(self):
 class CalledProcessError(subprocess.CalledProcessError):
     """Command returned non-zero exit code"""
 
-    # TODO python3.10 min version: Use `list` instead of `typing.List`
-    def __init__(
-        self, *, returncode: int, cmd: typing.List[str], output: str, stderr: str
-    ) -> None:
+    def __init__(self, *, returncode: int, cmd: list[str], output: str, stderr: str) -> None:
         super().__init__(returncode=returncode, cmd=cmd, output=output, stderr=stderr)
 
 
@@ -201,12 +198,11 @@ def refresh(
         """
 
     @abc.abstractmethod
-    # TODO python3.10 min version: Use `list` instead of `typing.List`
     def _run_command(
         self,
-        command: typing.List[str],
+        command: list[str],
         *,
-        timeout: typing.Optional[int],
+        timeout: int | None,
         input: str = None,  # noqa: A002 Match subprocess.run()
     ) -> str:
         """Run command in container.
@@ -215,8 +211,7 @@ def _run_command(
             CalledProcessError: Command returns non-zero exit code
         """
 
-    # TODO python3.10 min version: Use `list` instead of `typing.List`
-    def run_mysql_router(self, args: typing.List[str], *, timeout: int = None) -> str:
+    def run_mysql_router(self, args: list[str], *, timeout: int = None) -> str:
         """Run MySQL Router command.
 
         Raises:
@@ -225,10 +220,9 @@ def run_mysql_router(self, args: typing.List[str], *, timeout: int = None) -> st
         args.insert(0, self._mysql_router_command)
         return self._run_command(args, timeout=timeout)
 
-    # TODO python3.10 min version: Use `list` instead of `typing.List`
     def run_mysql_shell(
         self,
-        args: typing.List[str],
+        args: list[str],
         *,
         timeout: int = None,
         input: str = None,  # noqa: A002 Match subprocess.run()
@@ -251,9 +245,7 @@ def create_router_rest_api_credentials_file(self) -> None:
             # create empty credentials file
             self.rest_api_credentials_file.write_text("")
 
-    def set_mysql_router_rest_api_password(
-        self, *, user: str, password: typing.Optional[str]
-    ) -> None:
+    def set_mysql_router_rest_api_password(self, *, user: str, password: str | None) -> None:
         """Set REST API credentials using the mysqlrouter_password command."""
         self.create_router_rest_api_credentials_file()
 

common/common/lifecycle.py

@@ -6,9 +6,9 @@
 https://juju.is/docs/sdk/a-charms-life
 """
 
+import collections.abc
 import enum
 import logging
-import typing
 
 import ops
 
@@ -37,7 +37,7 @@ class Unit(ops.Object):
     def __init__(
         self,
         charm: ops.CharmBase,
-        subordinated_relation_endpoint_names: typing.Optional[typing.Iterable[str]],
+        subordinated_relation_endpoint_names: collections.abc.Collection[str] | None,
     ):
         """Unit lifecycle
 

common/common/mysql_shell/__init__.py

@@ -26,8 +26,7 @@
 logger = logging.getLogger(__name__)
 
 
-# TODO python3.10 min version: Add `(kw_only=True)`
-@dataclasses.dataclass
+@dataclasses.dataclass(kw_only=True)
 class RouterUserInformation:
     """MySQL Router user information"""
 
@@ -49,8 +48,7 @@ def __init__(self, *, message: str, code: int, traceback_message: str):
         self.traceback_message = traceback_message
 
 
-# TODO python3.10 min version: Add `(kw_only=True)`
-@dataclasses.dataclass
+@dataclasses.dataclass(kw_only=True)
 class Shell:
     """MySQL Shell connected to MySQL cluster"""
 
@@ -109,8 +107,7 @@ def _run_code(self, code: str) -> None:
                 )
                 raise
 
-    # TODO python3.10 min version: Use `list` instead of `typing.List`
-    def _run_sql(self, sql_statements: typing.List[str]) -> None:
+    def _run_sql(self, sql_statements: list[str]) -> None:
         """Connect to MySQL cluster and execute SQL."""
         self._run_code(
             _jinja_env.get_template("run_sql.py.jinja").render(statements=sql_statements)
@@ -127,10 +124,9 @@ def _get_attributes(self, additional_attributes: dict = None) -> str:
             attributes.update(additional_attributes)
         return json.dumps(attributes)
 
-    # TODO python3.10 min version: Use `set` instead of `typing.Set`
-    def _get_mysql_databases(self) -> typing.Set[str]:
+    def _get_mysql_databases(self) -> set[str]:
         """Returns a set with the MySQL databases."""
-        logger.debug(f"Getting MySQL databases")
+        logger.debug("Getting MySQL databases")
         output_file = self._container.path("/tmp/mysqlsh_output.json")
         self._run_code(
             _jinja_env.get_template("get_mysql_databases.py.jinja").render(
@@ -143,8 +139,7 @@ def _get_mysql_databases(self) -> typing.Set[str]:
         logger.debug(f"MySQL databases found: {len(rows)}")
         return {row[0] for row in rows}
 
-    # TODO python3.10 min version: Use `set` instead of `typing.Set`
-    def _get_mysql_roles(self, name_pattern: str) -> typing.Set[str]:
+    def _get_mysql_roles(self, name_pattern: str) -> set[str]:
         """Returns a set with the MySQL roles."""
         logger.debug(f"Getting MySQL roles with {name_pattern=}")
         output_file = self._container.path("/tmp/mysqlsh_output.json")
@@ -231,9 +226,7 @@ def add_attributes_to_mysql_router_user(
         self._run_sql([f"ALTER USER `{username}` ATTRIBUTE '{attributes}'"])
         logger.debug(f"Added {attributes=} to {username=}")
 
-    def get_mysql_router_user_for_unit(
-        self, unit_name: str
-    ) -> typing.Optional[RouterUserInformation]:
+    def get_mysql_router_user_for_unit(self, unit_name: str) -> RouterUserInformation | None:
         """Get MySQL Router user created by a previous instance of the unit.
 
         Get username & router ID attribute.

common/common/relations/cos.py

@@ -71,7 +71,7 @@ def relation_exists(self) -> bool:
 
     @property
     @abc.abstractmethod
-    def tracing_endpoint(self) -> typing.Optional[str]:
+    def tracing_endpoint(self) -> str | None:
         """The tracing endpoint."""
 
     def get_monitoring_password(self) -> str:

common/common/relations/database_provides.py

@@ -191,8 +191,7 @@ def __init__(self, charm_: "abstract_charm.MySQLRouterCharm") -> None:
         charm_.framework.observe(self._interface.on.database_requested, charm_.reconcile)
 
     @property
-    # TODO python3.10 min version: Use `list` instead of `typing.List`
-    def _shared_users(self) -> typing.List[_RelationWithSharedUser]:
+    def _shared_users(self) -> list[_RelationWithSharedUser]:
         shared_users = []
         for relation in self._interface.relations:
             try:
@@ -297,15 +296,14 @@ def delete_all_databags(self) -> None:
             relation.delete_databag()
         logger.debug("Deleted all application databags")
 
-    def get_status(self, event) -> typing.Optional[ops.StatusBase]:
+    def get_status(self, event) -> ops.StatusBase | None:
         """Report non-active status."""
         requested_users = []
         exception_reporting_priority = (
             _UnsupportedExtraUserRole,
             remote_databag.IncompleteDatabag,
         )
-        # TODO python3.10 min version: Use `list` instead of `typing.List`
-        exceptions: typing.List[status_exception.StatusException] = []
+        exceptions: list[status_exception.StatusException] = []
         for relation in self._interface.relations:
             try:
                 requested_users.append(

common/common/relations/database_requires.py

@@ -113,7 +113,7 @@ def __init__(self, charm_: "abstract_charm.MySQLRouterCharm") -> None:
         charm_.framework.observe(self._interface.on.database_created, charm_.reconcile)
         charm_.framework.observe(self._interface.on.endpoints_changed, charm_.reconcile)
 
-    def get_connection_info(self, *, event) -> typing.Optional[CompleteConnectionInformation]:
+    def get_connection_info(self, *, event) -> CompleteConnectionInformation | None:
         """Information for connection to MySQL cluster"""
         try:
             return CompleteConnectionInformation(interface=self._interface, event=event)
@@ -130,7 +130,7 @@ def is_relation_breaking(self, event) -> bool:
             pass
         return False
 
-    def get_status(self, event) -> typing.Optional[ops.StatusBase]:
+    def get_status(self, event) -> ops.StatusBase | None:
         """Report non-active status."""
         try:
             CompleteConnectionInformation(interface=self._interface, event=event)

common/common/relations/remote_databag.py

@@ -4,7 +4,6 @@
 """Relation databag for remote application"""
 
 import logging
-import typing
 
 import ops
 
@@ -28,10 +27,7 @@ class RemoteDatabag(dict):
 
     def __init__(
         self,
-        # TODO python3.10 min version: Use `|` instead of `typing.Union`
-        interface: typing.Union[
-            data_interfaces.DatabaseRequires, data_interfaces.DatabaseProvides
-        ],
+        interface: data_interfaces.DatabaseRequires | data_interfaces.DatabaseProvides,
         relation: ops.Relation,
     ) -> None:
         super().__init__(interface.fetch_relation_data()[relation.id])

common/common/relations/secrets.py

@@ -28,8 +28,8 @@ def __init__(
         self,
         charm: "abstract_charm.MySQLRouterCharm",
         relation_name: str,
-        app_secret_fields: typing.List[str] = [],
-        unit_secret_fields: typing.List[str] = [],
+        app_secret_fields: list[str] = [],
+        unit_secret_fields: list[str] = [],
     ) -> None:
         self._charm = charm
         self._relation_name = relation_name
@@ -54,17 +54,15 @@ def _peer_relation_data(self, scope: Scopes) -> data_interfaces.DataPeer:
         elif scope == UNIT_SCOPE:
             return self._peer_relation_unit
 
-    def get_value(self, scope: Scopes, key: str) -> typing.Optional[str]:
+    def get_value(self, scope: Scopes, key: str) -> str | None:
         """Get secret from the secret storage."""
         if scope not in typing.get_args(Scopes):
             raise ValueError("Unknown secret scope")
 
         peers = self._charm.model.get_relation(self._relation_name)
         return self._peer_relation_data(scope).fetch_my_relation_field(peers.id, key)
 
-    def set_value(
-        self, scope: Scopes, key: str, value: typing.Optional[str]
-    ) -> typing.Optional[str]:
+    def set_value(self, scope: Scopes, key: str, value: str | None) -> str | None:
         """Set secret from the secret storage."""
         if scope not in typing.get_args(Scopes):
             raise ValueError("Unknown secret scope")

common/common/relations/tls.py

@@ -177,7 +177,7 @@ def __init__(self, charm_: "abstract_charm.MySQLRouterCharm") -> None:
         )
 
     @property
-    def _relation(self) -> typing.Optional[_Relation]:
+    def _relation(self) -> _Relation | None:
         if not self._charm.model.get_relation(self.NAME):
             return
         return _Relation(
@@ -194,21 +194,21 @@ def certificate_saved(self) -> bool:
         return self._relation.certificate_saved
 
     @property
-    def key(self) -> typing.Optional[str]:
+    def key(self) -> str | None:
         """The TLS private key"""
         if self._relation is None:
             return None
         return self._relation.key
 
     @property
-    def certificate(self) -> typing.Optional[str]:
+    def certificate(self) -> str | None:
         """The TLS certificate"""
         if self._relation is None:
             return None
         return self._relation.certificate
 
     @property
-    def certificate_authority(self) -> typing.Optional[str]:
+    def certificate_authority(self) -> str | None:
         """The TLS certificate authority"""
         if self._relation is None:
             return None