Correctly extract name from image tag

Fixes #71

Author: RealOrangeOne

db-auto-backup.py

@@ -9,7 +9,7 @@
 from datetime import datetime
 from io import StringIO
 from pathlib import Path
-from typing import IO, Callable, Dict, NamedTuple, Optional, Sequence
+from typing import IO, Callable, Dict, Iterable, NamedTuple, Optional
 
 import docker
 import pycron
@@ -150,7 +150,7 @@ def backup_redis(container: Container) -> str:
 INCLUDE_LOGS = bool(os.environ.get("INCLUDE_LOGS"))
 
 
-def get_backup_provider(container_names: Sequence[str]) -> Optional[BackupProvider]:
+def get_backup_provider(container_names: Iterable[str]) -> Optional[BackupProvider]:
     for name in container_names:
         for provider in BACKUP_PROVIDERS:
             if any(fnmatch.fnmatch(name, pattern) for pattern in provider.patterns):
@@ -159,14 +159,23 @@ def get_backup_provider(container_names: Sequence[str]) -> Optional[BackupProvid
     return None
 
 
+def get_container_names(container: Container) -> Iterable[str]:
+    names = set()
+    for tag in container.image.tags:
+        registry, image = docker.auth.resolve_repository_name(tag)
+        image, tag_name = image.split(":", 1)
+        names.add(image)
+    return names
+
+
 @pycron.cron(SCHEDULE)
 def backup(now: datetime) -> None:
     docker_client = docker.from_env()
 
     backed_up_containers = []
 
     for container in docker_client.containers.list():
-        container_names = [tag.rsplit(":", 1)[0] for tag in container.image.tags]
+        container_names = get_container_names(container)
         backup_provider = get_backup_provider(container_names)
         if backup_provider is None:
             continue

requirements.txt

@@ -1,4 +1,4 @@
-docker==6.1.3
+docker==7.1.0
 python-dotenv==1.0.0
 tqdm==4.66.2
 requests

tests/tests.py

@@ -2,6 +2,7 @@
 from importlib.util import module_from_spec, spec_from_loader
 from pathlib import Path
 from typing import Any, Callable
+from unittest.mock import MagicMock
 
 import pytest
 
@@ -90,3 +91,18 @@ def test_healthchecks_success_hook_url_custom_host(monkeypatch: Any) -> None:
 def test_uptime_kuma_success_hook_url(monkeypatch: Any) -> None:
     monkeypatch.setenv("UPTIME_KUMA_URL", "https://uptime-kuma.com")
     assert db_auto_backup.get_success_hook_url() == "https://uptime-kuma.com"
+
+
+@pytest.mark.parametrize(
+    "tag,name",
+    [
+        ("postgres:14-alpine", "postgres"),
+        ("docker.io/postgres:14-alpine", "postgres"),
+        ("ghcr.io/realorangeone/db-auto-backup:latest", "realorangeone/db-auto-backup"),
+        ("theorangeone/db-auto-backup:latest:latest", "theorangeone/db-auto-backup"),
+    ],
+)
+def test_get_container_names(tag: str, name: str) -> None:
+    container = MagicMock()
+    container.image.tags = [tag]
+    assert db_auto_backup.get_container_names(container) == {name}