Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Recursively fetch dependent rows with mysqldump
#!/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()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment