Recursively fetch dependent rows with mysqldump

mysql-recursive-dump.py

#!/usr/bin/env python3
#
# Copyright (c) 2021 Philip Iezzi, Onlime GmbH - https://www.onlime.ch
#

import argparse
import subprocess

header = """SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT;
SET @OLD_CHARACTER_SET_RESULTS=@@CHARACTER_SET_RESULTS;
SET @OLD_COLLATION_CONNECTION=@@COLLATION_CONNECTION;
SET NAMES utf8mb4;
SET @OLD_TIME_ZONE=@@TIME_ZONE;
SET TIME_ZONE='+00:00';
SET @OLD_UNIQUE_CHECKS=@@UNIQUE_CHECKS, UNIQUE_CHECKS=0;
SET @OLD_FOREIGN_KEY_CHECKS=@@FOREIGN_KEY_CHECKS, FOREIGN_KEY_CHECKS=0;
SET @OLD_SQL_MODE=@@SQL_MODE, SQL_MODE='NO_AUTO_VALUE_ON_ZERO';
SET @OLD_SQL_NOTES=@@SQL_NOTES, SQL_NOTES=0;
"""
footer = """SET TIME_ZONE=@OLD_TIME_ZONE
SET SQL_MODE=@OLD_SQL_MODE;
SET FOREIGN_KEY_CHECKS=@OLD_FOREIGN_KEY_CHECKS;
SET UNIQUE_CHECKS=@OLD_UNIQUE_CHECKS;
SET CHARACTER_SET_CLIENT=@OLD_CHARACTER_SET_CLIENT;
SET CHARACTER_SET_RESULTS=@OLD_CHARACTER_SET_RESULTS;
SET COLLATION_CONNECTION=@OLD_COLLATION_CONNECTION;
SET SQL_NOTES=@OLD_SQL_NOTES;
"""

def prepend_line(filename: str, line: str):
    with open(filename, 'r') as f:
        data = f.read()
    with open(filename, 'w') as f:
        f.write(line + data)

def append_line(filename: str, line: str):
    with open(filename, 'a') as fd:
        fd.write(line)

class RecursiveDumper:

    def __init__(self, database: str, model: str, id: int, dryrun: bool = False):
        self.dumpfile = f'dump-{model}-{id}.sql'
        self.database = database
        self.model = model
        self.id = id
        self.dryrun = dryrun

    def dump(self):
        self.dump_customer() if self.model == 'customer' else self.dump_webabo()

    def dump_customer(self):
        custid = self.id
        customers_query = f'customer_id = {custid}'
        webabos_query = f'webabo_id IN (SELECT id FROM webabos WHERE {customers_query})'
        # ... (stripped down)
        queries = {
            'customers': f'id = {custid}',
            'webabos': f'customer_id = {custid}',
            'subdomains': webabos_query,
            'mailaccounts': webabos_query,
            # ...
        }
        self.run_dumps(queries)

    def dump_webabo(self):
        webaboid = self.id
        webabos_query = f'webabo_id = {webaboid}'
        subdomains_query = f'subdomain_id IN (SELECT id FROM subdomains WHERE {webabos_query})'
        addondomains_query = f'addondomain_id IN (SELECT id FROM addondomains WHERE {webabos_query} OR {subdomains_query})'
        dnszones_query = f'{webabos_query} OR {addondomains_query}'
        # ... (stripped down)
        queries = {
            'webabos': f'id = {webaboid}',
            'logins': webabos_query,
            'subdomains': webabos_query,
            'mailaccounts': webabos_query,
            # ...
        }
        self.run_dumps(queries)
    
    def run_dumps(self, queries: object):
        with open(self.dumpfile, 'w') as f:
            for table, where in queries.items():
                cmd = f"mysqldump --skip-extended-insert --skip-triggers --replace --compact --no-create-info --lock-all-tables --where '{where}' {self.database} {table}"
                print(cmd) if self.dryrun else subprocess.run(cmd, stdout=f, shell=True)
        prepend_line(self.dumpfile, header)
        append_line(self.dumpfile, footer)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument('model', choices=['customer', 'webabo'], help='Object to dump')
    parser.add_argument('id', type=int, help='Customer or webabo ID')
    parser.add_argument('--database', type=str, default='mydb', help='Airpane database')
    parser.add_argument('--dryrun', '--dry-run', action='store_true', help='Dry-run without dumping any data')
    args = parser.parse_args()

    dumper = RecursiveDumper(args.database, args.model, args.id, args.dryrun)
    dumper.dump()