[MISC] Add ty typechecking (#1744)

* Enable ty

* fix(types): resolve ty type checking errors

Fix all 194 type errors reported by ty across 11 source files. Main
patterns fixed: annotate self.charm as PostgresqlOperatorCharm via
TYPE_CHECKING guards, add str | None narrowing for optional secrets and
endpoints, fix implicit None returns, convert dict_keys to list before
subscripting, and suppress dynamic Literal constructs with ty: ignore.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Tweaks

* Apply suggestions from code review

Co-authored-by: Marcelo Henrique Neppel <marcelo.neppel@gmail.com>
Signed-off-by: Dragomir Penev <6687393+dragomirp@users.noreply.github.com>

* Empty string setting

---------

Signed-off-by: Dragomir Penev <6687393+dragomirp@users.noreply.github.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: Marcelo Henrique Neppel <marcelo.neppel@gmail.com>

Author: 3 people

poetry.lock

@@ -47,6 +47,7 @@ optional = true
 
 [tool.poetry.group.lint.dependencies]
 codespell = "^2.4.2"
+ty = "^0.0.50"
 
 [tool.poetry.group.unit]
 optional = true
@@ -139,3 +140,11 @@ max-complexity = 10
 
 [tool.ruff.lint.pydocstyle]
 convention = "google"
+
+[tool.ty.environment]
+python = ".tox/lint/"
+extra-paths = ["./lib"]
+
+[tool.ty.src]
+include = ["src", "scripts"]
+exclude = ["tests"]

pyproject.toml

@@ -133,7 +133,7 @@ def _get_s3_session_resource(self, s3_parameters: dict):
             ),
         )
 
-    def _are_backup_settings_ok(self) -> tuple[bool, str | None]:
+    def _are_backup_settings_ok(self) -> tuple[bool, str]:
         """Validates whether backup settings are OK."""
         if self.model.get_relation(self.relation_name) is None:
             return (
@@ -145,7 +145,7 @@ def _are_backup_settings_ok(self) -> tuple[bool, str | None]:
         if missing_parameters:
             return False, f"Missing S3 parameters: {missing_parameters}"
 
-        return True, None
+        return True, ""
 
     @property
     def _can_initialise_stanza(self) -> bool:
@@ -209,7 +209,7 @@ def _is_standby_cluster(self) -> bool:
 
         return not self.charm.async_replication.is_primary_cluster()
 
-    def can_use_s3_repository(self) -> tuple[bool, str | None]:
+    def can_use_s3_repository(self) -> tuple[bool, str]:
         """Returns whether the charm was configured to use another cluster repository."""
         # Check model uuid
         s3_parameters, _ = self._retrieve_s3_parameters()
@@ -272,7 +272,7 @@ def can_use_s3_repository(self) -> tuple[bool, str | None]:
                 )
                 return False, ANOTHER_CLUSTER_REPOSITORY_ERROR_MESSAGE
 
-        return True, None
+        return True, ""
 
     def _change_connectivity_to_database(self, connectivity: bool) -> None:
         """Enable or disable the connectivity to the database."""
@@ -1059,8 +1059,8 @@ def _run_backup(
             try:
                 backup_id = list(self._list_backups(show_failed=True).keys())[-1]
             except ListBackupsError:
-                logger.exception(error_message)
                 error_message = "Failed to retrieve backup id"
+                logger.exception(error_message)
                 logger.error(f"Backup failed: {error_message}")
                 event.fail(error_message)
                 return
@@ -1238,7 +1238,7 @@ def _generate_fake_backup_id(self, backup_type: str) -> str:
         if backup_type == "full":
             return datetime.strftime(datetime.now(), "%Y%m%d-%H%M%SF")
         if backup_type == "differential":
-            backups = self._list_backups(show_failed=False, parse=False).keys()
+            backups = list(self._list_backups(show_failed=False, parse=False).keys())
             last_full_backup = None
             for label in backups[::-1]:
                 if label.endswith("F"):
@@ -1249,12 +1249,13 @@ def _generate_fake_backup_id(self, backup_type: str) -> str:
                 raise TypeError("Differential backup requested but no previous full backup")
             return f"{last_full_backup}_{datetime.strftime(datetime.now(), '%Y%m%d-%H%M%SD')}"
         if backup_type == "incremental":
-            backups = self._list_backups(show_failed=False, parse=False).keys()
+            backups = list(self._list_backups(show_failed=False, parse=False).keys())
             if not backups:
                 raise TypeError("Incremental backup requested but no previous successful backup")
             return f"{backups[-1]}_{datetime.strftime(datetime.now(), '%Y%m%d-%H%M%SI')}"
+        raise TypeError(f"Invalid backup type: {backup_type}")
 
-    def _fetch_backup_from_id(self, backup_id: str) -> str:
+    def _fetch_backup_from_id(self, backup_id: str) -> str | None:
         """Fetches backup's pgbackrest label from backup id."""
         timestamp = f"{datetime.strftime(datetime.strptime(backup_id, '%Y-%m-%dT%H:%M:%SZ'), '%Y%m%d-%H%M%S')}"
         backups = self._list_backups(show_failed=False, parse=False).keys()
@@ -1361,7 +1362,7 @@ def _render_pgbackrest_conf_file(self) -> bool:
             storage_path=self.charm._storage_path,
             user=BACKUP_USER,
             retention_full=s3_parameters["delete-older-than-days"],
-            process_max=max(os.cpu_count() - 2, 1),
+            process_max=max(self.charm.cpu_count - 2, 1),
         )
         # Render pgBackRest config file.
         self.charm._patroni.render_file(f"{PGBACKREST_CONF_PATH}/pgbackrest.conf", rendered, 0o640)