conqp · October 10, 2023 07:44 · Jan 26, 2021 · Jan 26, 2021 · Jan 26, 2021 · Jan 26, 2021
diff --git a/db_backup.py b/db_backup.py
@@ -25,7 +25,7 @@
 from pathlib import Path
 from subprocess import PIPE, CalledProcessError, run
 from sys import exit    # pylint: disable=W0622
-from tarfile import TarFile, TarInfo, open
+from tarfile import TarFile, TarInfo, open      # pylint: disable=W0622
 from typing import Iterable, Iterator, NamedTuple
 
 
@@ -108,8 +108,7 @@ def dump_databases_to_tarfile(tarfile: TarFile, databases: Iterable[str], *,
     """Dumps the given databases to the given tar file."""
 
     for mysql_dump in dump_databases(databases, user=user, passwd=passwd):
-        filename = f'{mysql_dump.database}-{mysql_dump.table}.sql'
-        tarinfo = TarInfo(filename)
+        tarinfo = TarInfo(f'{mysql_dump.database}.{mysql_dump.table}.sql')
         tarinfo.size = len(mysql_dump.dump)
         tarinfo.mode = 0o644
         bytesio = BytesIO(mysql_dump.dump)
@@ -143,7 +142,7 @@ def main():
     file = Path(filename)
 
     if file.exists():
-        LOGGER.error('File exists.')
+        LOGGER.error('File exists: %s', file)
         exit(1)
 
     with open(file, mode='w:gz') as tarfile:

diff --git a/db_backup.py b/db_backup.py
@@ -1,6 +1,6 @@
 #! /usr/bin/env python3
 #
-#  db_backup - Update local MySQL databases.
+#  db_backup - Backup local MySQL databases.
 #  Copyright (C) 2021  HOMEINFO - Digitale Informationssysteme GmbH
 #
 #  This program is free software: you can redistribute it and/or modify

diff --git a/db_backup.py b/db_backup.py
@@ -80,6 +80,7 @@ def dump_table(database: str, table: str, *,
                user: str, passwd: str) -> MySQLDump:
     """Dumps a table of a database."""
 
+    LOGGER.info('Dumping %s.%s', database, table)
     command = ['mysqldump', f'--user={user}', f'--password={passwd}',
                database, table]
     result = run(command, check=True, stdout=PIPE)
@@ -108,7 +109,6 @@ def dump_databases_to_tarfile(tarfile: TarFile, databases: Iterable[str], *,
 
     for mysql_dump in dump_databases(databases, user=user, passwd=passwd):
         filename = f'{mysql_dump.database}-{mysql_dump.table}.sql'
-        LOGGER.info('Adding MySQL dump: %s', filename)
         tarinfo = TarInfo(filename)
         tarinfo.size = len(mysql_dump.dump)
         tarinfo.mode = 0o644

diff --git a/db_backup.py b/db_backup.py
@@ -0,0 +1,158 @@
+#! /usr/bin/env python3
+#
+#  db_backup - Update local MySQL databases.
+#  Copyright (C) 2021  HOMEINFO - Digitale Informationssysteme GmbH
+#
+#  This program is free software: you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation, either version 3 of the License, or
+#  (at your option) any later version.
+#
+#  This program is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program.  If not, see <https://www.gnu.org/licenses/>.
+"""Backups local databases."""
+
+from argparse import ArgumentParser, Namespace
+from datetime import datetime
+from io import BytesIO
+from logging import INFO, basicConfig, getLogger
+from os import linesep
+from pathlib import Path
+from subprocess import PIPE, CalledProcessError, run
+from sys import exit    # pylint: disable=W0622
+from tarfile import TarFile, TarInfo, open
+from typing import Iterable, Iterator, NamedTuple
+
+
+LOG_FORMAT = '[%(levelname)s] %(name)s: %(message)s'
+LOGGER = getLogger(Path(__file__).stem)
+EXCLUDE = {
+    'information_schema',
+    'mysql',
+    'performance_schema'
+}
+
+
+class MySQLDump(NamedTuple):
+    """Represents a MySQL table dump."""
+
+    database: str
+    table: str
+    dump: bytes
+
+
+def get_args() -> Namespace:
+    """Parses the command line arguments."""
+
+    parser = ArgumentParser(description='Backups MariaDB databases.')
+    parser.add_argument('-u', '--user', metavar='user', default='backup')
+    parser.add_argument('-p', '--passwd', metavar='password')
+    parser.add_argument('-e', '--exclude', nargs='*', metavar='database')
+    parser.add_argument('-t', '--timestamp', action='store_true')
+    parser.add_argument('file')
+    return parser.parse_args()
+
+
+def get_databases(user: str, passwd: str) -> Iterator[str]:
+    """Returns a list of database names."""
+
+    command = ['mysql', f'--user={user}', f'--password={passwd}',
+               '--skip-column-names', '--execute', 'show databases;']
+    result = run(command, check=True, stdout=PIPE, text=True)
+    return filter(None, result.stdout.split(linesep))
+
+
+def get_tables(database: str, *, user: str, passwd: str) -> Iterator[str]:
+    """Returns a list of tables in the given database."""
+
+    command = ['mysql', f'--user={user}', f'--password={passwd}',
+               '--skip-column-names', '--execute', 'show tables;', database]
+    result = run(command, check=True, stdout=PIPE, text=True)
+    return filter(None, result.stdout.split(linesep))
+
+
+def dump_table(database: str, table: str, *,
+               user: str, passwd: str) -> MySQLDump:
+    """Dumps a table of a database."""
+
+    command = ['mysqldump', f'--user={user}', f'--password={passwd}',
+               database, table]
+    result = run(command, check=True, stdout=PIPE)
+    return MySQLDump(database, table, result.stdout)
+
+
+def dump_database(database: str, *, user: str, passwd: str) \
+        -> Iterator[MySQLDump]:
+    """Dumps a database."""
+
+    for table in get_tables(database, user=user, passwd=passwd):
+        yield dump_table(database, table, user=user, passwd=passwd)
+
+
+def dump_databases(databases: Iterable[str], *, user: str, passwd: str) \
+        -> Iterator[MySQLDump]:
+    """Dumps the given databases."""
+
+    for database in databases:
+        yield from dump_database(database, user=user, passwd=passwd)
+
+
+def dump_databases_to_tarfile(tarfile: TarFile, databases: Iterable[str], *,
+                              user: str, passwd: str) -> None:
+    """Dumps the given databases to the given tar file."""
+
+    for mysql_dump in dump_databases(databases, user=user, passwd=passwd):
+        filename = f'{mysql_dump.database}-{mysql_dump.table}.sql'
+        LOGGER.info('Adding MySQL dump: %s', filename)
+        tarinfo = TarInfo(filename)
+        tarinfo.size = len(mysql_dump.dump)
+        tarinfo.mode = 0o644
+        bytesio = BytesIO(mysql_dump.dump)
+        tarfile.addfile(tarinfo, fileobj=bytesio)
+
+
+def main():
+    """Runs the script."""
+
+    args = get_args()
+    basicConfig(format=LOG_FORMAT, level=INFO)
+
+    try:
+        databases = get_databases(args.user, args.passwd)
+    except CalledProcessError as error:
+        exit(error.returncode)
+
+    exclude = {*EXCLUDE, *args.exclude}
+    databases = filter(lambda db: db not in exclude, databases)
+
+    if args.timestamp:
+        timestamp = datetime.now().isoformat()
+
+        try:
+            filename = args.file % timestamp
+        except TypeError:
+            filename = args.file.format(timestamp)
+    else:
+        filename = args.file
+
+    file = Path(filename)
+
+    if file.exists():
+        LOGGER.error('File exists.')
+        exit(1)
+
+    with open(file, mode='w:gz') as tarfile:
+        try:
+            dump_databases_to_tarfile(
+                tarfile, databases, user=args.user, passwd=args.passwd)
+        except CalledProcessError as error:
+            exit(error.returncode)
+
+
+if __name__ == '__main__':
+    main()