typing: src/vorta/store

src/vorta/store/connection.py

@@ -1,8 +1,9 @@
 import os
 import shutil
 from datetime import datetime, timedelta
+from typing import Any, Optional
 
-from peewee import Tuple, fn
+from peewee import SqliteDatabase, Tuple, fn
 from playhouse import signals
 
 from vorta import config
@@ -28,18 +29,18 @@
 
 
 @signals.post_save(sender=SettingsModel)
-def setup_autostart(model_class, instance, created):
+def setup_autostart(model_class, instance, created) -> None:
     if instance.key == 'autostart':
         open_app_at_startup(instance.value)
 
 
-def cleanup_db():
+def cleanup_db() -> None:
     # Clean up database
     DB.execute_sql("VACUUM")
     DB.close()
 
 
-def init_db(con=None):
+def init_db(con: Optional[SqliteDatabase] = None) -> None:
     if con is not None:
         os.umask(0o0077)
         DB.initialize(con)
@@ -62,12 +63,12 @@ def init_db(con=None):
     # Delete old log entries after 6 months.
     # The last `create` command of each profile must not be deleted
     # since the scheduler uses it to determine the last backup time.
-    last_backups_per_profile = (
+    last_backups_per_profile = Tuple(
         EventLogModel.select(EventLogModel.profile, fn.MAX(EventLogModel.start_time))
         .where(EventLogModel.subcommand == 'create')
         .group_by(EventLogModel.profile)
     )
-    last_scheduled_backups_per_profile = (
+    last_scheduled_backups_per_profile = Tuple(
         EventLogModel.select(EventLogModel.profile, fn.MAX(EventLogModel.start_time))
         .where(EventLogModel.subcommand == 'create', EventLogModel.category == 'scheduled')
         .group_by(EventLogModel.profile)
@@ -105,7 +106,7 @@ def init_db(con=None):
         s.save()
 
 
-def backup_current_db(schema_version):
+def backup_current_db(schema_version: Any) -> None:
     """
     Creates a backup copy of settings.db
     """

src/vorta/store/migrations.py

@@ -1,6 +1,8 @@
 from datetime import datetime
+from typing import Any
 
 import peewee as pw
+from peewee import SqliteDatabase
 from playhouse.migrate import SqliteMigrator, migrate
 
 from .models import (
@@ -9,13 +11,14 @@
     BackupProfileModel,
     EventLogModel,
     RepoModel,
+    SchemaVersion,
     SettingsModel,
     SourceFileModel,
     WifiSettingModel,
 )
 
 
-def run_migrations(current_schema, db_connection):
+def run_migrations(current_schema: SchemaVersion, db_connection: SqliteDatabase):
     """
     Apply new schema versions to database.
 
@@ -251,7 +254,7 @@ def run_migrations(current_schema, db_connection):
         )
 
 
-def _apply_schema_update(current_schema, version_after, *operations):
+def _apply_schema_update(current_schema: Any, version_after: int, *operations) -> None:
     with DB.atomic():
         migrate(*operations)
         current_schema.version = version_after

src/vorta/store/models.py

@@ -8,6 +8,7 @@
 import logging
 from datetime import datetime
 from enum import Enum
+from typing import Any, Dict, Optional
 
 import peewee as pw
 from playhouse import signals
@@ -26,11 +27,11 @@ class JSONField(pw.TextField):
     From: https://gist.github.com/rosscdh/f4f26758b0228f475b132c688f15af2b
     """
 
-    def db_value(self, value):
+    def db_value(self, value) -> Optional[str]:
         """Convert the python value for storage in the database."""
         return value if value is None else json.dumps(value)
 
-    def python_value(self, value):
+    def python_value(self, value) -> Optional[str]:
         """Convert the database value to a pythonic value."""
         return value if value is None else json.loads(value)
 
@@ -53,7 +54,7 @@ class RepoModel(BaseModel):
     create_backup_cmd = pw.CharField(default='')
     extra_borg_arguments = pw.CharField(default='')
 
-    def is_remote_repo(self):
+    def is_remote_repo(self) -> bool:
         return not self.url.startswith('/')
 
     class Meta:
@@ -103,14 +104,14 @@ class BackupProfileModel(BaseModel):
     post_backup_cmd = pw.CharField(default='')
     dont_run_on_metered_networks = pw.BooleanField(default=True)
 
-    def refresh(self):
+    def refresh(self) -> None:
         return type(self).get(self._pk_expr())
 
-    def slug(self):
+    def slug(self) -> str:
         return slugify(self.name)
 
-    def get_combined_exclusion_string(self):
-        allPresets = get_exclusion_presets()
+    def get_combined_exclusion_string(self) -> str:
+        allPresets: Dict[str, Dict[str, Any]] = get_exclusion_presets()
         excludes = ""
 
         if (
@@ -202,7 +203,7 @@ class ArchiveModel(BaseModel):
     size = pw.IntegerField(null=True)
     trigger = pw.CharField(null=True)
 
-    def formatted_time(self):
+    def formatted_time(self) -> None:
         return
 
     class Meta:
@@ -266,5 +267,5 @@ class Meta:
 class BackupProfileMixin:
     """Extend to support multiple profiles later."""
 
-    def profile(self):
+    def profile(self) -> BackupProfileModel:
         return BackupProfileModel.get(id=self.window().current_profile.id)