diff --git a/src/restic_compose_backup/commands.py b/src/restic_compose_backup/commands.py deleted file mode 100644 index cf209cb..0000000 --- a/src/restic_compose_backup/commands.py +++ /dev/null @@ -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) diff --git a/src/restic_compose_backup/utils.py b/src/restic_compose_backup/utils.py index 64bcad6..b58ca08 100644 --- a/src/restic_compose_backup/utils.py +++ b/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)