restic backup script (incl stopping and restarting other user services)

backup.py

#!/usr/bin/env python3

import argparse
import json
import logging
import os
import subprocess
import sys

__script_location__ = os.path.realpath(os.path.join(os.getcwd(), os.path.dirname(__file__)))

class App():
    def __init__(self, args):
        self.args = args
        self.logger = logging.getLogger()
        self.config = self.load_config()

    def load_config(self):
        config_json_path = os.path.join(__script_location__, self.args.config)
        with open(config_json_path) as config_json:
            return json.load(config_json)

    def execute_command(self, command, allow_in_dryrun=False, dryrun_returncode=0, check=True, print_stdout_info=False):
        if self.args.dry_run and not allow_in_dryrun:
            self.logger.debug(f'Would execute command: {' '.join(command)}')
            ret = DryrunDummyCompletedCommand()
            ret.returncode = dryrun_returncode
            return ret
        else:
            try:
                self.logger.debug(f'Executing command: {' '.join(command)}')
                ret = subprocess.run(command, text=True, capture_output=True, check=check)
                if ret.stdout:
                    if print_stdout_info:
                        self.logger.info(ret.stdout)
                    else:
                        self.logger.debug(ret.stdout)
                if ret.returncode != 0:
                    self.logger.error(f'  Returned: {ret.returncode}\n{ret.stderr}')
                return ret
            except subprocess.CalledProcessError as e:
                if e.stdout:
                    self.logger.error(e.stdout)
                self.logger.error(f'  Returned: {e.returncode}\n{e.stderr}')
                raise

    def execute_systemctl_command(self, user, command, **kwargs):
        full_command = ['systemctl', '--user', '-M', f'"{user}"@'] + command
        return self.execute_command(['sh', '-c', ' '.join(full_command)], **kwargs)

    def execute_restic_command(self, command, **kwargs):
        real_command = self.remove_dry_run(command)
        if self.args.dry_run and command != real_command:
            full_command = self.config['restic']['command_base'] + command
            return self.execute_command(full_command, **dict(kwargs, allow_in_dryrun=True, print_stdout_info=True))

        full_command = self.config['restic']['command_base'] + real_command
        return self.execute_command(full_command, **dict(kwargs, print_stdout_info=True))

    def remove_dry_run(self, command):
        dry_run_arg = self.config['restic']['dry_run_arg']
        return [s for s in command if s != dry_run_arg]

    def stop_services(self):
        try:
            self.logger.info('Stopping services...')
            self.stopped_units = []
            for service in self.config['disable_services']:
                user = service['user']

                for unit in service['units']:
                    command = self.execute_systemctl_command(user, ['is-active', '--quiet', f'"{unit}"'], allow_in_dryrun=True, check=False)
                    if command.returncode != 0:
                        self.logger.warning(f'Unit "{unit}"@"{user}" not running, will ignore')
                        continue
                    if self.stop_unit(user, unit):
                        self.stopped_units.append([user, unit])
        except:
            self.logger.exception('  Failed to stop services')

    def stop_unit(self, user, unit):
        try:
            self.logger.info(f'  Stopping "{unit}"@"{user}"...')
            cmd = self.execute_systemctl_command(user, ['stop', '--quiet', f'"{unit}"'])
            if cmd.returncode == 0:
                if not self.args.dry_run:
                    self.logger.info('    Stopped')
                return True
            else:
                self.logger.error(f'    Failed to stop "{unit}"@"{user}" with exit code {cmd.returncode}:\n{cmd.stderr}')
                return False
        except:
            self.logger.error(f'    Failed to stop "{unit}"@"{user}"', exc_info=True)
            return False

    def perform_backup(self):
        try:
            self.logger.info('Performing backup...')
            for backup_pass in self.config['restic']['passes']:
                self.execute_restic_command(backup_pass)
        except:
            self.logger.exception('  Failed to perform backups')

    def restart_services(self):
        try:
            self.logger.info('Restarting previously stopped services...')
            for user, unit in reversed(self.stopped_units):
                self.restart_unit(user, unit)
        except:
            self.logger.exception('  Failed to restart services')

    def restart_unit(self, user, unit):
        try:
            self.logger.info(f'  Starting "{unit}"@"{user}"...')
            cmd = self.execute_systemctl_command(user, ['start', '--quiet', f'"{unit}"'])
            if cmd.returncode == 0:
                if not self.args.dry_run:
                    self.logger.info('    Started')
                return True
            else:
                self.logger.error(f'    Failed to start "{unit}"@"{user}" with exit code {cmd.returncode}:\n{cmd.stderr}')
                return False
        except:
            self.logger.error(f'    Failed to start "{unit}"@"{user}"', exec_info=True)
            return False


class DryrunDummyCompletedCommand():
    def __init__(self):
        self.returncode = 0

def parse_args():
    log_levels = {
        'critical': logging.CRITICAL,
        'error': logging.ERROR,
        'warning': logging.WARNING,
        'info': logging.INFO,
        'debug': logging.DEBUG
    }

    parser = argparse.ArgumentParser()
    parser.add_argument('--log', default='info', help=f'Provide logging level, one of {"/".join(log_levels.keys())}')
    parser.add_argument('--dry-run', action='store_true')
    parser.add_argument('config', default='config.json', nargs='?', help='config file name, default: config.json')
    args = parser.parse_args()

    args.log = log_levels.get(args.log.lower())

    return args

def main():
    args = parse_args()
    logging.basicConfig(level=args.log, stream=sys.stdout, style='{', format='[{levelname:.1}]: {message}')
    app = App(args)

    if args.dry_run:
        logging.getLogger().warning('Dry run...')

    app.stop_services()
    app.perform_backup()
    app.restart_services()

if __name__ == '__main__':
    main()

config.example.json

{
  "disable_services": [
    {
      "user": "git",
      "units": [ "forgejo.service" ]
    },
    {
      "user": "postgres",
      "units": [ "postgres.service" ]
    }
  ],

  "restic": {
    "command_base": [ "restic", "--repository-file=/etc/restic/repository.txt", "--password-file=/etc/restic/password.txt" ],
    "dry_run_arg": "--dry-run",
    "passes": [
      [ "backup", "--dry-run", "--skip-if-unchanged", "--tag=files", "--files-from=/etc/restic/includes.txt", "--exclude-file=/etc/restic/excludes.txt" ],
      [ "forget", "--dry-run", "--keep-within-daily=7d", "--keep-within-weekly=1m", "--keep-within-monthly=1y", "--keep-within-yearly=100y", "--tag=files", "--group-by=" ],
      [ "prune", "--dry-run" ],
      [ "check" ]
    ]
  }
}