Restructure backup config to allow future expansion

Author: RealOrangeOne

db-auto-backup.py

@@ -5,7 +5,7 @@
 from datetime import datetime
 from io import StringIO
 from pathlib import Path
-from typing import Callable, Dict, Optional, Sequence
+from typing import Callable, Dict, NamedTuple, Optional, Sequence
 
 import docker
 import pycron
@@ -17,6 +17,12 @@
 BackupCandidate = Callable[[Container], str]
 
 
+class BackupProvider(NamedTuple):
+    patterns: list[str]
+    backup_method: BackupCandidate
+    file_extension: str
+
+
 def get_container_env(container: Container) -> Dict[str, Optional[str]]:
     """
     Get all environment variables from a container.
@@ -45,23 +51,29 @@ def backup_mysql(container: Container) -> str:
     return f"bash -c 'mysqldump {auth} --all-databases'"
 
 
-BACKUP_MAPPING: Dict[str, BackupCandidate] = {
-    "postgres": backup_psql,
-    "mysql": backup_mysql,
-    "mariadb": backup_mysql,  # Basically the same thing
-    "*/linuxserver/mariadb": backup_mysql,  # LSIO have their own MariaDB container
-}
+BACKUP_PROVIDERS: list[BackupProvider] = [
+    BackupProvider(
+        patterns=["postgres"], backup_method=backup_psql, file_extension="sql"
+    ),
+    BackupProvider(
+        patterns=["mysql", "mariadb", "*/linuxserver/mariadb"],
+        backup_method=backup_mysql,
+        file_extension="sql",
+    ),
+]
+
 
 BACKUP_DIR = Path(os.environ.get("BACKUP_DIR", "/var/backups"))
 SCHEDULE = os.environ.get("SCHEDULE", "@daily")
 SHOW_PROGRESS = sys.stdout.isatty()
 
 
-def get_backup_method(container_names: Sequence[str]) -> Optional[BackupCandidate]:
+def get_backup_provider(container_names: Sequence[str]) -> Optional[BackupProvider]:
     for name in container_names:
-        for container_pattern, backup_candidate in BACKUP_MAPPING.items():
-            if fnmatch.fnmatch(name, container_pattern):
-                return backup_candidate
+        for provider in BACKUP_PROVIDERS:
+            for pattern in provider.patterns:
+                if fnmatch.fnmatch(name, pattern):
+                    return provider
 
     return None
 
@@ -74,12 +86,12 @@ def backup(now: datetime) -> None:
 
     for container in docker_client.containers.list():
         container_names = [tag.rsplit(":", 1)[0] for tag in container.image.tags]
-        backup_method = get_backup_method(container_names)
-        if backup_method is None:
+        backup_provider = get_backup_provider(container_names)
+        if backup_provider is None:
             continue
 
-        backup_command = backup_method(container)
-        backup_file = BACKUP_DIR / f"{container.name}.sql"
+        backup_command = backup_provider.backup_method(container)
+        backup_file = BACKUP_DIR / f"{container.name}.{backup_provider.file_extension}"
         _, output = container.exec_run(backup_command, stream=True, demux=True)
 
         with tqdm.wrapattr(