Add type annotations to borg layer modules (Phase 3 of #2362)

pyproject.toml

@@ -141,6 +141,10 @@ ignore_missing_imports = true
 module = ["vorta.store.models", "vorta.store.connection", "vorta.store.settings", "vorta.network_status.*"]
 disallow_untyped_defs = true
 
+[[tool.mypy.overrides]]
+module = ["vorta.borg.*"]
+disallow_untyped_defs = true
+
 [[tool.mypy.overrides]]
 module = ["vorta.store.migrations"]
 ignore_errors = true

src/vorta/borg/_compatibility.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from packaging.version import Version
 
 MIN_BORG_FOR_FEATURE = {
@@ -23,13 +25,13 @@ class BorgCompatibility:
     version = "1.1.4"
     path = ""
 
-    def set_version(self, version, path):
+    def set_version(self, version: str, path: str) -> None:
         self.version = version
         self.path = path
 
-    def check(self, feature_name):
+    def check(self, feature_name: str) -> bool:
         return Version(self.version) >= MIN_BORG_FOR_FEATURE[feature_name]
 
-    def get_version(self):
+    def get_version(self) -> tuple[str, str]:
         """Returns the version and path of the Borg binary."""
         return self.version, self.path

src/vorta/borg/borg_job.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 import logging
 import os
@@ -9,6 +11,7 @@
 import time
 from collections import namedtuple
 from datetime import datetime as dt
+from typing import Any
 from subprocess import PIPE, Popen, TimeoutExpired
 from threading import Lock
 
@@ -20,7 +23,7 @@
 from vorta.i18n import trans_late, translate
 from vorta.keyring.abc import VortaKeyring
 from vorta.keyring.db import VortaDBKeyring
-from vorta.store.models import EventLogModel
+from vorta.store.models import BackupProfileModel, EventLogModel
 from vorta.utils import borg_compat, pretty_bytes
 
 keyring_lock = Lock()
@@ -49,7 +52,7 @@ class BorgJob(JobInterface):
     result = QtCore.pyqtSignal(dict)
     keyring = None  # Store keyring to minimize imports
 
-    def __init__(self, cmd, params, site="default"):
+    def __init__(self, cmd: list[str], params: dict[str, Any], site: str = "default"):
         """
         Thread to run Borg operations in.
 
@@ -109,7 +112,7 @@ def __init__(self, cmd, params, site="default"):
         self.process = None
         self.cleanup_files = params.get('cleanup_files', [])
 
-    def repo_id(self):
+    def repo_id(self) -> str:
         return self.site_id
 
     def cancel(self):
@@ -125,7 +128,7 @@ def cancel(self):
                     pass
 
     @classmethod
-    def prepare(cls, profile):
+    def prepare(cls, profile: BackupProfileModel | FakeProfile) -> dict[str, Any]:
         """
         Prepare for running Borg. This function in the base class should be called from all
         subclasses and calls that define their own `cmd`.
@@ -201,7 +204,7 @@ def prepare(cls, profile):
         return ret
 
     @classmethod
-    def prepare_bin(cls):
+    def prepare_bin(cls) -> str | None:
         """Find packaged borg binary. Prefer globally installed."""
         # On MacOS, the PATH environment variable does not seem to be set when run as a pyinstaller binary.
         # More info at https://github.com/borgbase/vorta/issues/2100
@@ -347,11 +350,11 @@ def read_async(fd):
         for tmpfile in self.cleanup_files:
             tmpfile.close()
 
-    def process_result(self, result):
+    def process_result(self, result: dict[str, Any]) -> None:
         pass
 
     def started_event(self):
         self.updated.emit(self.tr('Task started'))
 
-    def finished_event(self, result):
+    def finished_event(self, result: dict[str, Any]):
         self.result.emit(result)

src/vorta/borg/break_lock.py

@@ -1,3 +1,9 @@
+from __future__ import annotations
+
+from typing import Any
+
+from vorta.store.models import BackupProfileModel
+
 from .borg_job import BorgJob
 
 
@@ -6,15 +12,15 @@ def started_event(self):
         self.app.backup_started_event.emit()
         self.app.backup_progress_event.emit(f"[{self.params['profile_name']}] {self.tr('Breaking repository lock…')}")
 
-    def finished_event(self, result):
+    def finished_event(self, result: dict[str, Any]):
         self.app.backup_finished_event.emit(result)
         self.app.backup_progress_event.emit(
             f"[{self.params['profile_name']}] {self.tr('Repository lock broken. Please redo your last action.')}"
         )
         self.result.emit(result)
 
     @classmethod
-    def prepare(cls, profile):
+    def prepare(cls, profile: BackupProfileModel) -> dict[str, Any]:
         ret = super().prepare(profile)
         if not ret['ok']:
             return ret

src/vorta/borg/change_passphrase.py

@@ -1,10 +1,12 @@
-from typing import Any, Dict
+from __future__ import annotations
+
+from typing import Any
 
 from vorta import config
 from vorta.borg._compatibility import MIN_BORG_FOR_FEATURE
 from vorta.i18n import trans_late, translate
 from vorta.i18n.richtext import escape, format_richtext, link
-from vorta.store.models import RepoModel
+from vorta.store.models import BackupProfileModel, RepoModel
 from vorta.utils import borg_compat
 
 from .borg_job import BorgJob
@@ -15,7 +17,7 @@ def started_event(self):
         self.app.backup_started_event.emit()
         self.app.backup_progress_event.emit(f"[{self.params['profile_name']}] {self.tr('Changing Borg passphrase…')}")
 
-    def finished_event(self, result: Dict[str, Any]):
+    def finished_event(self, result: dict[str, Any]):
         """
         Process that the job terminated with the given results.
         Parameters
@@ -37,7 +39,7 @@ def finished_event(self, result: Dict[str, Any]):
             self.app.backup_progress_event.emit(f"[{self.params['profile_name']}] {self.tr('Passphrase changed.')}")
 
     @classmethod
-    def prepare(cls, profile, new_passphrase):
+    def prepare(cls, profile: BackupProfileModel, new_passphrase: str) -> dict[str, Any]:
         ret = super().prepare(profile)
         if not ret['ok']:
             return ret
@@ -65,7 +67,7 @@ def prepare(cls, profile, new_passphrase):
 
         return ret
 
-    def process_result(self, result):
+    def process_result(self, result: dict[str, Any]):
         if result['returncode'] == 0:
             # Change passphrase in keyring
             repo = RepoModel.get(url=result['params']['repo_url'])

src/vorta/borg/check.py

@@ -1,10 +1,14 @@
-from typing import Any, Dict
+from __future__ import annotations
+
+from typing import Any
 
 from vorta import config
 from vorta.i18n import translate
 from vorta.i18n.richtext import escape, format_richtext, link
 from vorta.utils import borg_compat
 
+from vorta.store.models import BackupProfileModel
+
 from .borg_job import BorgJob
 
 
@@ -13,7 +17,7 @@ def started_event(self):
         self.app.backup_started_event.emit()
         self.app.backup_progress_event.emit(f"[{self.params['profile_name']}] {self.tr('Starting consistency check…')}")
 
-    def finished_event(self, result: Dict[str, Any]):
+    def finished_event(self, result: dict[str, Any]):
         """
         Process that the job terminated with the given results.
 
@@ -36,7 +40,7 @@ def finished_event(self, result: Dict[str, Any]):
             self.app.backup_progress_event.emit(f"[{self.params['profile_name']}] {self.tr('Check completed.')}")
 
     @classmethod
-    def prepare(cls, profile):
+    def prepare(cls, profile: BackupProfileModel) -> dict[str, Any]:
         ret = super().prepare(profile)
         if not ret['ok']:
             return ret

src/vorta/borg/compact.py

@@ -1,10 +1,14 @@
-from typing import Any, Dict
+from __future__ import annotations
+
+from typing import Any
 
 from vorta import config
 from vorta.i18n import translate
 from vorta.i18n.richtext import escape, format_richtext, link
 from vorta.utils import borg_compat
 
+from vorta.store.models import BackupProfileModel
+
 from .borg_job import BorgJob
 
 
@@ -15,7 +19,7 @@ def started_event(self):
             f"[{self.params['profile_name']} {self.tr('Starting repository compaction...')}]"
         )
 
-    def finished_event(self, result: Dict[str, Any]):
+    def finished_event(self, result: dict[str, Any]):
         """
         Process that the job terminated with the given results.
 
@@ -37,7 +41,7 @@ def finished_event(self, result: Dict[str, Any]):
             self.app.backup_progress_event.emit(f"[{self.params['profile_name']}] {self.tr('Compaction completed.')}")
 
     @classmethod
-    def prepare(cls, profile):
+    def prepare(cls, profile: BackupProfileModel) -> dict[str, Any]:
         ret = super().prepare(profile)
         if not ret['ok']:
             return ret

src/vorta/borg/create.py

@@ -1,8 +1,11 @@
+from __future__ import annotations
+
 import os
 import shlex
 import subprocess
 import tempfile
 from datetime import datetime as dt
+from typing import Any
 
 from PyQt6.QtCore import QCoreApplication
 
@@ -12,11 +15,13 @@
 from vorta.store.models import ArchiveModel, RepoModel, SourceFileModel, WifiSettingModel
 from vorta.utils import borg_compat, format_archive_name, get_network_status_monitor
 
+from vorta.store.models import BackupProfileModel
+
 from .borg_job import BorgJob
 
 
 class BorgCreateJob(BorgJob):
-    def process_result(self, result):
+    def process_result(self, result: dict[str, Any]) -> None:
         if result['returncode'] in [0, 1] and 'archive' in result['data']:
             new_archive, created = ArchiveModel.get_or_create(
                 snapshot_id=result['data']['archive']['id'],
@@ -51,20 +56,20 @@ def process_result(self, result):
                 self.app.backup_log_event.emit('', {})
                 self.app.backup_progress_event.emit(f"[{self.params['profile_name']}] {self.tr('Backup finished.')}")
 
-    def progress_event(self, fmt):
+    def progress_event(self, fmt: str) -> None:
         self.app.backup_progress_event.emit(f"[{self.params['profile_name']}] {fmt}")
 
     def started_event(self):
         self.app.backup_started_event.emit()
         self.app.backup_progress_event.emit(f"[{self.params['profile_name']}] {self.tr('Backup started.')}")
 
-    def finished_event(self, result):
+    def finished_event(self, result: dict[str, Any]):
         self.pre_post_backup_cmd(self.params, cmd='post_backup_cmd', returncode=result['returncode'])
         self.app.backup_finished_event.emit(result)
         self.result.emit(result)
 
     @classmethod
-    def pre_post_backup_cmd(cls, params, cmd='pre_backup_cmd', returncode=0):
+    def pre_post_backup_cmd(cls, params: dict[str, Any], cmd: str = 'pre_backup_cmd', returncode: int = 0) -> int:
         cmd = getattr(params['profile'], cmd)
         if cmd:
             env = {
@@ -85,7 +90,7 @@ def pre_post_backup_cmd(cls, params, cmd='pre_backup_cmd', returncode=0):
             return 0  # 0 if no command was run.
 
     @classmethod
-    def prepare(cls, profile):
+    def prepare(cls, profile: BackupProfileModel) -> dict[str, Any]:
         """
         `borg create` is called from different places and needs some preparation.
         Centralize it here and return the required arguments to the caller.

src/vorta/borg/delete.py

@@ -1,8 +1,12 @@
-from typing import List
+from __future__ import annotations
+
+from typing import Any
 
 from vorta.store.models import RepoModel
 from vorta.utils import borg_compat
 
+from vorta.store.models import BackupProfileModel
+
 from .borg_job import BorgJob
 
 
@@ -11,7 +15,7 @@ def started_event(self):
         self.app.backup_started_event.emit()
         self.app.backup_progress_event.emit(f"[{self.params['profile_name']}] {self.tr('Deleting archive…')}")
 
-    def finished_event(self, result):
+    def finished_event(self, result: dict[str, Any]):
         # set repo stats to N/A
         repo = RepoModel.get(id=result['params']['repo_id'])
         repo.total_size = None
@@ -25,7 +29,7 @@ def finished_event(self, result):
         self.app.backup_progress_event.emit(f"[{self.params['profile_name']}] {self.tr('Archive deleted.')}")
 
     @classmethod
-    def prepare(cls, profile, archives: List[str]):
+    def prepare(cls, profile: BackupProfileModel, archives: list[str]) -> dict[str, Any]:
         ret = super().prepare(profile)
         if not ret['ok']:
             return ret

src/vorta/borg/diff.py

@@ -1,5 +1,11 @@
+from __future__ import annotations
+
+from typing import Any
+
 from vorta.utils import borg_compat
 
+from vorta.store.models import BackupProfileModel
+
 from .borg_job import BorgJob
 
 
@@ -10,15 +16,15 @@ def started_event(self):
             f"[{self.params['profile_name']}] {self.tr('Requesting differences between archives…')}"
         )
 
-    def finished_event(self, result):
+    def finished_event(self, result: dict[str, Any]):
         self.app.backup_finished_event.emit(result)
         self.app.backup_progress_event.emit(
             f"[{self.params['profile_name']}] {self.tr('Obtained differences between archives.')}"
         )
         self.result.emit(result)
 
     @classmethod
-    def prepare(cls, profile, archive_name_1, archive_name_2):
+    def prepare(cls, profile: BackupProfileModel, archive_name_1: str, archive_name_2: str) -> dict[str, Any]:
         ret = super().prepare(profile)
         if not ret['ok']:
             return ret