Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/backups_v2/backup_file.py: 29%
57 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:32 +0000
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:32 +0000
1import json 1a
2import shutil 1a
3import tempfile 1a
4from pathlib import Path 1a
7class BackupContents: 1a
8 _tables: dict | None = None 1a
10 def __init__(self, file: Path) -> None: 1a
11 self.base = self._find_base(file)
12 self.data_directory = self._find_data_dir_from_base(self.base)
13 self.tables = self._find_database_from_base(self.base)
15 @classmethod 1a
16 def _find_base(cls, file: Path) -> Path: 1a
17 # Safari mangles our ZIP structure and adds a "__MACOSX" directory at the root along with
18 # an arbitrarily-named directory containing the actual contents. So, if we find a dunder directory
19 # at the root (i.e. __MACOSX) we traverse down the first non-dunder directory and assume this is the base.
20 # This works because our backups never contain a directory that starts with "__".
21 dirs = [d for d in file.iterdir() if d.is_dir()]
22 dunder_dirs = [d for d in dirs if d.name.startswith("__")]
23 normal_dirs = [d for d in dirs if not d.name.startswith("__")]
25 if not dunder_dirs:
26 return file
28 # If the backup somehow adds a __MACOSX directory alongside the data directory, rather than in the
29 # parent directory, we don't want to traverse down. We check for our database.json file, and if it exists,
30 # we're already at the correct base.
31 if cls._find_database_from_base(file).exists():
32 return file
34 # This ZIP file was mangled, so we return the first non-dunder directory (if it exists).
35 return normal_dirs[0] if normal_dirs else file
37 @classmethod 1a
38 def _find_data_dir_from_base(cls, base: Path) -> Path: 1a
39 return base / "data"
41 @classmethod 1a
42 def _find_database_from_base(cls, base: Path) -> Path: 1a
43 return base / "database.json"
45 def validate(self) -> bool: 1a
46 if not self.base.is_dir():
47 return False
49 if not self.data_directory.is_dir():
50 return False
52 if not self.tables.is_file():
53 return False
55 return True
57 def schema_version(self) -> str: 1a
58 tables = self.read_tables()
60 alembic_version = tables.get("alembic_version", [])
62 if not alembic_version:
63 return ""
65 return alembic_version[0].get("version_num", "")
67 def read_tables(self) -> dict: 1a
68 if self._tables is None:
69 with open(self.tables) as f:
70 self._tables = json.load(f)
72 return self._tables
75class BackupFile: 1a
76 temp_dir: Path | None = None 1a
78 def __init__(self, file: Path) -> None: 1a
79 self.zip = file
81 def __enter__(self) -> BackupContents: 1a
82 self.temp_dir = Path(tempfile.mkdtemp())
83 shutil.unpack_archive(str(self.zip), str(self.temp_dir))
84 return BackupContents(self.temp_dir)
86 def __exit__(self, exc_type, exc_val, exc_tb): 1a
87 if self.temp_dir and self.temp_dir.is_dir():
88 shutil.rmtree(self.temp_dir)
90 self.temp_dir = None