mirror of https://github.com/borgbase/vorta
272 lines
8.4 KiB
Python
272 lines
8.4 KiB
Python
"""
|
|
This module provides the app's data store using Peewee with SQLite.
|
|
|
|
At the bottom there is a simple schema migration system.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from datetime import datetime
|
|
from enum import Enum
|
|
from typing import Any, Dict, Optional
|
|
|
|
import peewee as pw
|
|
from playhouse import signals
|
|
|
|
from vorta.utils import slugify
|
|
from vorta.views.utils import get_exclusion_presets
|
|
|
|
DB = pw.Proxy()
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class JSONField(pw.TextField):
|
|
"""
|
|
Class to "fake" a JSON field with a text field. Not efficient but works nicely.
|
|
|
|
From: https://gist.github.com/rosscdh/f4f26758b0228f475b132c688f15af2b
|
|
"""
|
|
|
|
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) -> Optional[str]:
|
|
"""Convert the database value to a pythonic value."""
|
|
return value if value is None else json.loads(value)
|
|
|
|
|
|
class BaseModel(signals.Model):
|
|
"""Common model superclass."""
|
|
|
|
|
|
class RepoModel(BaseModel):
|
|
"""A single remote repo with unique URL."""
|
|
|
|
url = pw.CharField(unique=True)
|
|
name = pw.CharField(default='')
|
|
added_at = pw.DateTimeField(default=datetime.now)
|
|
encryption = pw.CharField(null=True)
|
|
unique_size = pw.IntegerField(null=True)
|
|
unique_csize = pw.IntegerField(null=True)
|
|
total_size = pw.IntegerField(null=True)
|
|
total_unique_chunks = pw.IntegerField(null=True)
|
|
create_backup_cmd = pw.CharField(default='')
|
|
extra_borg_arguments = pw.CharField(default='')
|
|
|
|
def is_remote_repo(self) -> bool:
|
|
return not self.url.startswith('/')
|
|
|
|
class Meta:
|
|
database = DB
|
|
|
|
|
|
class RepoPassword(BaseModel):
|
|
"""Fallback to save repo passwords. Only used if no Keyring available."""
|
|
|
|
url = pw.CharField(unique=True)
|
|
password = pw.CharField()
|
|
|
|
class Meta:
|
|
database = DB
|
|
|
|
|
|
class BackupProfileModel(BaseModel):
|
|
"""Allows the user to switch between different configurations."""
|
|
|
|
name = pw.CharField()
|
|
added_at = pw.DateTimeField(default=datetime.now)
|
|
repo = pw.ForeignKeyField(RepoModel, default=None, null=True)
|
|
ssh_key = pw.CharField(default=None, null=True)
|
|
compression = pw.CharField(default='lz4')
|
|
exclude_patterns = pw.TextField(null=True)
|
|
exclude_if_present = pw.TextField(null=True)
|
|
schedule_mode = pw.CharField(default='off')
|
|
schedule_interval_count = pw.IntegerField(default=3)
|
|
schedule_interval_unit = pw.CharField(default='hours')
|
|
schedule_fixed_hour = pw.IntegerField(default=3)
|
|
schedule_fixed_minute = pw.IntegerField(default=42)
|
|
schedule_interval_hours = pw.IntegerField(default=3) # no longer used
|
|
schedule_interval_minutes = pw.IntegerField(default=42) # no longer used
|
|
schedule_make_up_missed = pw.BooleanField(default=True)
|
|
validation_on = pw.BooleanField(default=True)
|
|
validation_weeks = pw.IntegerField(default=3)
|
|
prune_on = pw.BooleanField(default=False)
|
|
prune_hour = pw.IntegerField(default=2)
|
|
prune_day = pw.IntegerField(default=7)
|
|
prune_week = pw.IntegerField(default=4)
|
|
prune_month = pw.IntegerField(default=6)
|
|
prune_year = pw.IntegerField(default=2)
|
|
prune_keep_within = pw.CharField(default='10H', null=True)
|
|
new_archive_name = pw.CharField(default="{hostname}-{now:%Y-%m-%d-%H%M%S}")
|
|
prune_prefix = pw.CharField(default="{hostname}-")
|
|
pre_backup_cmd = pw.CharField(default='')
|
|
post_backup_cmd = pw.CharField(default='')
|
|
dont_run_on_metered_networks = pw.BooleanField(default=True)
|
|
|
|
def refresh(self) -> None:
|
|
return type(self).get(self._pk_expr())
|
|
|
|
def slug(self) -> str:
|
|
return slugify(self.name)
|
|
|
|
def get_combined_exclusion_string(self) -> str:
|
|
allPresets: Dict[str, Dict[str, Any]] = get_exclusion_presets()
|
|
excludes = ""
|
|
|
|
if (
|
|
ExclusionModel.select()
|
|
.where(
|
|
ExclusionModel.profile == self,
|
|
ExclusionModel.enabled,
|
|
ExclusionModel.source == ExclusionModel.SourceFieldOptions.CUSTOM.value,
|
|
)
|
|
.count()
|
|
> 0
|
|
):
|
|
excludes = "# custom added rules\n"
|
|
|
|
for exclude in ExclusionModel.select().where(
|
|
ExclusionModel.profile == self,
|
|
ExclusionModel.enabled,
|
|
ExclusionModel.source == ExclusionModel.SourceFieldOptions.CUSTOM.value,
|
|
):
|
|
excludes += f"{exclude.name}\n"
|
|
|
|
raw_excludes = self.exclude_patterns
|
|
if raw_excludes:
|
|
excludes += "\n# raw exclusions\n"
|
|
excludes += raw_excludes
|
|
excludes += "\n"
|
|
|
|
# go through all source=='preset' exclusions, find the name in the allPresets dict, and add the patterns
|
|
for exclude in ExclusionModel.select().where(
|
|
ExclusionModel.profile == self,
|
|
ExclusionModel.enabled,
|
|
ExclusionModel.source == ExclusionModel.SourceFieldOptions.PRESET.value,
|
|
):
|
|
if exclude.name not in allPresets:
|
|
logger.warning("Exclusion preset %s not found in built-in presets.", exclude.name)
|
|
continue
|
|
excludes += f"\n# {exclude.name}\n"
|
|
for pattern in allPresets[exclude.name]['patterns']:
|
|
excludes += f"{pattern}\n"
|
|
|
|
return excludes
|
|
|
|
class Meta:
|
|
database = DB
|
|
|
|
|
|
class ExclusionModel(BaseModel):
|
|
"""
|
|
If this is a user created exclusion, the name will be the same as the pattern added. For exclusions added from
|
|
presets, the name will be the same as the preset name. Duplicate patterns are already handled by Borg.
|
|
"""
|
|
|
|
class SourceFieldOptions(Enum):
|
|
CUSTOM = 'custom'
|
|
PRESET = 'preset'
|
|
|
|
profile = pw.ForeignKeyField(BackupProfileModel, backref='exclusions')
|
|
name = pw.CharField()
|
|
enabled = pw.BooleanField(default=True)
|
|
source = pw.CharField(default=SourceFieldOptions.CUSTOM.value)
|
|
|
|
class Meta:
|
|
database = DB
|
|
|
|
|
|
class SourceFileModel(BaseModel):
|
|
"""A folder to be backed up, related to a Backup Configuration."""
|
|
|
|
dir = pw.CharField()
|
|
dir_size = pw.BigIntegerField(default=-1)
|
|
dir_files_count = pw.BigIntegerField(default=-1)
|
|
path_isdir = pw.BooleanField(default=False)
|
|
profile = pw.ForeignKeyField(BackupProfileModel, default=1)
|
|
added_at = pw.DateTimeField(default=datetime.utcnow)
|
|
|
|
class Meta:
|
|
database = DB
|
|
table_name = 'sourcedirmodel'
|
|
|
|
|
|
class ArchiveModel(BaseModel):
|
|
"""An archive in a remote repository."""
|
|
|
|
snapshot_id = pw.CharField()
|
|
name = pw.CharField()
|
|
repo = pw.ForeignKeyField(RepoModel, backref='archives')
|
|
time = pw.DateTimeField()
|
|
duration = pw.FloatField(null=True)
|
|
size = pw.IntegerField(null=True)
|
|
trigger = pw.CharField(null=True)
|
|
|
|
def formatted_time(self) -> None:
|
|
return
|
|
|
|
class Meta:
|
|
database = DB
|
|
|
|
|
|
class WifiSettingModel(BaseModel):
|
|
"""Save Wifi Settings"""
|
|
|
|
ssid = pw.CharField()
|
|
last_connected = pw.DateTimeField(null=True)
|
|
allowed = pw.BooleanField(default=True)
|
|
profile = pw.ForeignKeyField(BackupProfileModel, default=1)
|
|
|
|
class Meta:
|
|
database = DB
|
|
|
|
|
|
class EventLogModel(BaseModel):
|
|
"""Keep a log of background jobs."""
|
|
|
|
start_time = pw.DateTimeField(default=datetime.now)
|
|
end_time = pw.DateTimeField(default=datetime.now)
|
|
category = pw.CharField()
|
|
subcommand = pw.CharField(null=True)
|
|
message = pw.CharField(null=True)
|
|
returncode = pw.IntegerField(default=-1)
|
|
params = JSONField(null=True)
|
|
profile = pw.CharField(null=True)
|
|
repo_url = pw.CharField(null=True)
|
|
|
|
class Meta:
|
|
database = DB
|
|
|
|
|
|
class SchemaVersion(BaseModel):
|
|
"""Keep DB version to apply the correct migrations."""
|
|
|
|
version = pw.IntegerField()
|
|
changed_at = pw.DateTimeField(default=datetime.now)
|
|
|
|
class Meta:
|
|
database = DB
|
|
|
|
|
|
class SettingsModel(BaseModel):
|
|
"""App settings unrelated to a single profile or repo"""
|
|
|
|
key = pw.CharField(unique=True)
|
|
value = pw.BooleanField(default=False)
|
|
str_value = pw.CharField(default='')
|
|
label = pw.CharField()
|
|
group = pw.CharField(default='') # Settings group name and label
|
|
tooltip = pw.CharField(default='') # optional tooltip for `checkbox` type
|
|
type = pw.CharField()
|
|
|
|
class Meta:
|
|
database = DB
|
|
|
|
|
|
class BackupProfileMixin:
|
|
"""Extend to support multiple profiles later."""
|
|
|
|
def profile(self) -> BackupProfileModel:
|
|
return BackupProfileModel.get(id=self.window().current_profile.id)
|