Use docker exec api for database backups

src/restic_compose_backup/restic.py

@@ -2,9 +2,10 @@
 Restic commands
 """
 import logging
-from typing import List, Tuple
+from typing import List, Tuple, Union
 from subprocess import Popen, PIPE
 from restic_compose_backup import commands
+from restic_compose_backup import utils
 
 logger = logging.getLogger(__name__)
 
@@ -27,9 +28,10 @@ def backup_files(repository: str, source='/volumes'):
     ]))
 
 
-def backup_from_stdin(repository: str, filename: str, source_command: List[str]):
+def backup_from_stdin(repository: str, filename: str, container_id: str,
+                      source_command: List[str], environment: Union[dict, list] = None):
     """
-    Backs up from stdin running the source_command passed in.
+    Backs up from stdin running the source_command passed in within the given container.
     It will appear in restic with the filename (including path) passed in.
     """
     dest_command = restic(repository, [
@@ -39,20 +41,43 @@ def backup_from_stdin(repository: str, filename: str, source_command: List[str])
         filename,
     ])
 
-    # pipe source command into dest command
-    source_process = Popen(source_command, stdout=PIPE, bufsize=65536)
-    dest_process = Popen(dest_command, stdin=source_process.stdout, stdout=PIPE, stderr=PIPE, bufsize=65536)
+    client = utils.docker_client()
+
+    logger.debug('docker exec inside %s: %s', container_id, ' '.join(source_command))
+
+    # Create and start source command inside the given container
+    handle = client.api.exec_create(container_id, source_command, environment=environment)
+    exec_id = handle["Id"]
+    stream = client.api.exec_start(exec_id, stream=True, demux=True)
+    source_stderr = ""
+
+    # Create the restic process to receive the output of the source command
+    dest_process = Popen(dest_command, stdin=PIPE, stdout=PIPE, stderr=PIPE, bufsize=65536)
+
+    # Send the ouptut of the source command over to restic in the chunks received
+    for stdout_chunk, stderr_chunk in stream:
+        if stdout_chunk:
+            dest_process.stdin.write(stdout_chunk)
+
+        if stderr_chunk:
+            source_stderr += stderr_chunk.decode()
+
+    # Wait for restic to finish
     stdout, stderr = dest_process.communicate()
 
     # Ensure both processes exited with code 0
-    source_exit, dest_exit = source_process.poll(), dest_process.poll()
-    exit_code = 0 if (source_exit == 0 and dest_exit == 0) else 1
+    source_exit = client.api.exec_inspect(exec_id).get("ExitCode")
+    dest_exit = dest_process.poll()
+    exit_code = source_exit or dest_exit
 
     if stdout:
         commands.log_std('stdout', stdout, logging.DEBUG if exit_code == 0 else logging.ERROR)
 
+    if source_stderr:
+        commands.log_std(f'stderr ({source_command[0]})', source_stderr, logging.ERROR)
+
     if stderr:
-        commands.log_std('stderr', stderr, logging.ERROR)
+        commands.log_std('stderr (restic)', stderr, logging.ERROR)
 
     return exit_code