Back an SQLITE3 database

gistfile1.txt

#!/usr/bin/env python3
#

"""Backup a sqlite3 database into a zstandard compressed file"""

import logging
import os
import sqlite3
import sys

from functools import partial

import zstd

LOG_FILE = '/tmp/dxbackup.log'
DATABASE = '/home/fred/.local/dxcluster.db'
BACKUP_FILE = '/tmp/dxcluster.db'
BACKUP_FILE_ZST = BACKUP_FILE + '.zst'
BUF_SIZE = 2**20 # = 1MB

def sizeof_fmt(num, suffix="B"):
  for unit in ["", "Ki", "Mi", "Gi", "Ti", "Pi", "Ei", "Zi"]:
    if abs(num) < 1024.0:
      return f"{num:3.1f}{unit}{suffix}"
    num /= 1024.0
  return f"{num:.1f}Yi{suffix}"

def dump_db():
  logging.info('Backing up %s', DATABASE)

  if os.path.exists(BACKUP_FILE):
    os.unlink(BACKUP_FILE)

  try:
    with sqlite3.connect(DATABASE) as conn:
      curs = conn.cursor()
      curs.execute(f'VACUUM INTO "{BACKUP_FILE}"')
  except sqlite3.Error as err:
    logging.error(err)
    sys.exit(os.EX_IOERR)

def compress_backup():
  logging.info('Compressing %s into %s', BACKUP_FILE, BACKUP_FILE_ZST)
  try:
    with open(BACKUP_FILE, 'rb') as fdin:
      with open(BACKUP_FILE_ZST, 'wb') as fdout:
        for buf in iter(partial(fdin.read, BUF_SIZE), b''):
          fdout.write(zstd.compress(buf, 6))
  except IOError as err:
    logging.error(err)
    sys.exit(os.EX_IOERR)

  st_size = os.stat(BACKUP_FILE_ZST).st_size
  os.unlink(BACKUP_FILE)
  logging.info('Backup: %s, Size: %d (%s) done', BACKUP_FILE_ZST, st_size, sizeof_fmt(st_size))

def main():
  if sys.stdin.isatty():
    logfile = None
  else:
    logfile = LOG_FILE

  logging.basicConfig(
    filename=logfile,
    format='%(asctime)s %(name)s:%(lineno)d %(levelname)s - %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S',
    level=logging.getLevelName(os.getenv('LOG_LEVEL', 'INFO'))
  )

  dump_db()
  compress_backup()

if __name__ == "__main__":
  main()