docker service stopper decorator

src/restic/backup.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import subprocess
+from functools import partial
 from pathlib import Path
 
 import click
@@ -8,8 +9,9 @@ from rich.console import Console
 from rich.logging import RichHandler
 from rich.progress import Progress
 
-from restic import loki, snapshots
+from restic import snapshots
 from restic.console import console, logger
+from restic.docker import manage_containers
 from restic.loki import send_to_loki
 
 
@@ -55,16 +57,35 @@ def run(backup_dir: Path, loki_url: str = None, tag: str = 'python-script'):
     envvar='BACKUP_DIR',
 )
 @click.option('--loki-url', type=str, help='Loki URL for logging', envvar='LOKI_URL')
-@click.option('--tag', type=str, help='Tag to use in restic')
+@click.option('--tag', type=str, help='Tag to use in restic', default='python-script')
-def main(backup_dir: Path, loki_url: str = None, tag: str = 'python-script'):
+@click.option(
+    '--project', type=str, help='Name of the Docker compose project to use when stopping containers'
+)
+@click.option(
+    '--services',
+    type=str,
+    help='Comma-delimited names of the Docker compose services to stop before backing up. ',
+)
+def main(
+    backup_dir: Path,
+    loki_url: str = None,
+    tag: str = 'python-script',
+    project: str = None,
+    services: str = None,
+):
     console = Console()
 
     logging.basicConfig(
         level='DEBUG', format='%(message)s', handlers=[RichHandler(markup=True, console=console)]
     )
 
-    run(backup_dir, loki_url, tag)
+    if project is not None and services is not None:
-    snapshots.run(loki_url)
+        decorator = manage_containers(project=project, services=services.split(','))
+        func = decorator(run)
+    else:
+        func = run
+
+    func(backup_dir, loki_url, tag)
 
 
 if __name__ == '__main__':

src/restic/docker.py

@@ -0,0 +1,44 @@
+from functools import wraps
+
+import docker
+from docker.models.containers import Container
+
+
+from restic.console import logger
+
+
+def manage_containers(project: str, services: list[str]):
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            client = docker.from_env()
+
+            containers: list[Container] = [
+                c
+                for c in client.containers.list()
+                if c.labels['com.docker.compose.project'] == project
+                and c.labels['com.docker.compose.service'] in services
+            ]
+            try:
+                # Start the containers
+                for container in containers:
+                    container.stop()
+                    logger.info(f'Stopped container [blue]{container.name}[/]')
+
+                # Execute the wrapped function
+                result = func(*args, **kwargs)
+
+            except Exception as e:
+                # Handle exceptions and ensure containers are stopped
+                raise e
+            else:
+                return result
+            finally:
+                # Stop the containers
+                for container in containers:
+                    container.start()
+                    logger.info(f'Started container [blue]{container.name}[/]')
+
+        return wrapper
+
+    return decorator