Merge commands and utils

src/restic_compose_backup/commands.py

@@ -1,91 +0,0 @@
-import logging
-from typing import List, Tuple
-from subprocess import Popen, PIPE
-
-logger = logging.getLogger(__name__)
-
-
-def test():
-    return run(['ls', '/volumes'])
-
-
-def ping_mysql(host, port, username) -> int:
-    """Check if the mysql is up and can be reached"""
-    return run([
-        'mysqladmin',
-        'ping',
-        '--host',
-        host,
-        '--port',
-        port,
-        '--user',
-        username,
-    ])
-
-
-def ping_mariadb(host, port, username) -> int:
-    """Check if the mariadb is up and can be reached"""
-    return run([
-        'mysqladmin',
-        'ping',
-        '--host',
-        host,
-        '--port',
-        port,
-        '--user',
-        username,
-    ])
-
-
-def ping_postgres(host, port, username, password) -> int:
-    """Check if postgres can be reached"""
-    return run([
-        "pg_isready",
-        f"--host={host}",
-        f"--port={port}",
-        f"--username={username}",
-    ])
-
-
-def run(cmd: List[str]) -> int:
-    """Run a command with parameters"""
-    logger.debug('cmd: %s', ' '.join(cmd))
-    child = Popen(cmd, stdout=PIPE, stderr=PIPE)
-    stdoutdata, stderrdata = child.communicate()
-
-    if stdoutdata.strip():
-        log_std('stdout', stdoutdata.decode(),
-                logging.DEBUG if child.returncode == 0 else logging.ERROR)
-
-    if stderrdata.strip():
-        log_std('stderr', stderrdata.decode(), logging.ERROR)
-
-    logger.debug("returncode %s", child.returncode)
-    return child.returncode
-
-
-def run_capture_std(cmd: List[str]) -> Tuple[str, str]:
-    """Run a command with parameters and return stdout, stderr"""
-    logger.debug('cmd: %s', ' '.join(cmd))
-    child = Popen(cmd, stdout=PIPE, stderr=PIPE)
-    return child.communicate()
-
-
-def log_std(source: str, data: str, level: int):
-    if isinstance(data, bytes):
-        data = data.decode()
-
-    if not data.strip():
-        return
-
-    log_func = logger.debug if level == logging.DEBUG else logger.error
-    log_func('%s %s %s', '-' * 10, source, '-' * 10)
-
-    lines = data.split('\n')
-    if lines[-1] == '':
-        lines.pop()
-
-    for line in lines:
-        log_func(line)
-
-    log_func('-' * 28)

src/restic_compose_backup/utils.py

@@ -1,6 +1,7 @@
 import os
 import logging
-from typing import List, TYPE_CHECKING
+from typing import List, Tuple, TYPE_CHECKING
+from subprocess import Popen, PIPE
 from contextlib import contextmanager
 import docker
 
@@ -8,7 +9,6 @@ if TYPE_CHECKING:
     from restic_compose_backup.containers import Container
 
 logger = logging.getLogger(__name__)
-
 TRUE_VALUES = ['1', 'true', 'True', True, 1]
 
 
@@ -95,3 +95,89 @@ def environment(name, value):
             del os.environ[name]
         else:
             os.environ[name] = old_val
+
+
+def test():
+    return run(['ls', '/volumes'])
+
+
+def ping_mysql(host, port, username) -> int:
+    """Check if the mysql is up and can be reached"""
+    return run([
+        'mysqladmin',
+        'ping',
+        '--host',
+        host,
+        '--port',
+        port,
+        '--user',
+        username,
+    ])
+
+
+def ping_mariadb(host, port, username) -> int:
+    """Check if the mariadb is up and can be reached"""
+    return run([
+        'mysqladmin',
+        'ping',
+        '--host',
+        host,
+        '--port',
+        port,
+        '--user',
+        username,
+    ])
+
+
+def ping_postgres(host, port, username, password) -> int:
+    """Check if postgres can be reached"""
+    return run([
+        "pg_isready",
+        f"--host={host}",
+        f"--port={port}",
+        f"--username={username}",
+    ])
+
+
+def run(cmd: List[str]) -> int:
+    """Run a command with parameters"""
+    logger.debug('cmd: %s', ' '.join(cmd))
+    child = Popen(cmd, stdout=PIPE, stderr=PIPE)
+    stdoutdata, stderrdata = child.communicate()
+
+    if stdoutdata.strip():
+        log_std('stdout', stdoutdata.decode(),
+                logging.DEBUG if child.returncode == 0 else logging.ERROR)
+
+    if stderrdata.strip():
+        log_std('stderr', stderrdata.decode(), logging.ERROR)
+
+    logger.debug("returncode %s", child.returncode)
+    return child.returncode
+
+
+def run_capture_std(cmd: List[str]) -> Tuple[str, str]:
+    """Run a command with parameters and return stdout, stderr"""
+    logger.debug('cmd: %s', ' '.join(cmd))
+    child = Popen(cmd, stdout=PIPE, stderr=PIPE)
+    return child.communicate()
+
+
+def log_std(source: str, data: str, level: int):
+    if isinstance(data, bytes):
+        data = data.decode()
+
+    if not data.strip():
+        return
+
+    log_func = logger.debug if level == logging.DEBUG else logger.error
+    log_func('%s %s %s', '-' * 10, source, '-' * 10)
+
+    lines = data.split('\n')
+    if lines[-1] == '':
+        lines.pop()
+
+    for line in lines:
+        log_func(line)
+
+    log_func('-' * 28)