Source code for anuket.scripts.backupdb

# -*- coding: utf-8 -*-
""" Script to backup the Anuket database."""
import argparse
import bz2
import os
import sqlite3
import sys
from datetime import date

from sqlalchemy import engine_from_config
from pyramid.paster import get_appsettings

def main(argv=None):
    """ Main entry point for the `backupdb` script."""
    if argv is None:  # pragma: no cover
        argv = sys.argv
    command = BackupDBCommand(argv)

[docs]class BackupDBCommand(object): """Dump the database for backup purpose. Supported database: SQLite. """ description = 'Dump the database' usage = '%(prog)s config_uri' epilog = 'example: %(prog)s development.ini' parser = argparse.ArgumentParser( description=description, usage=usage, epilog=epilog) parser.add_argument('config_uri', nargs='?', help='the application config file') parser.add_argument('-o', '--overwrite', action='store_true', help='overwrite existing backups files') def __init__(self, argv): """ Get arguments from the ``argparse`` parser.""" self.args = self.parser.parse_args(argv[1:])
[docs] def run(self): """ Run the ``backup_db`` method or display the parser help message if the `config_uri` argument is missing. :return: ``backup_db`` method or 2 (missing argument error) """ if not self.args.config_uri: self.parser.print_help() return 2 else: return self.backup_db()
[docs] def backup_db(self): """ Dump the database and then compress and save the file. :return: 0 (OK) or 1 (abnormal termination error) """ config_uri = self.args.config_uri overwrite = self.args.overwrite settings = get_appsettings(config_uri) name = settings['anuket.brand_name'] directory = settings['anuket.backup_directory'] today = filename = '{0}-{1}.sql.bz2'.format(name, today) path = os.path.join(directory, filename) # check if the backup file already exist isfile = os.path.isfile(path) if isfile and not overwrite: print("There is already a database backup with the same name!") return 1 # get db engine from settings and create a dump engine = engine_from_config(settings, 'sqlalchemy.') connect_args = engine.url.translate_connect_args() if == 'sqlite': sql_dump = self.dump_sqlite(connect_args) # if == 'mysql': # pass # if == 'postgresql': # pass else: # pragma: no cover print("Unsuported database engine!") return 1 if sql_dump: # verify and/or create the backup directory if necessary if not os.path.exists(directory): try: os.makedirs(directory, 0775) except OSError: # pragma: no cover print("Could not create the backup directory.") return 1 # bzip and save the file bz = bz2.BZ2File(path, 'w') bz.write(sql_dump) bz.close() print("Database backup done.") return 0 # def dump_mysql(self, connect_args=None): # """ Dump a MySQL database.""" # def dump_postgresql(self, connect_args=None): # """ Dump a PostgreSQL database."""
def dump_sqlite(self, connect_args=None): """ Dump a SQLite database.""" con = sqlite3.connect(connect_args['database']) sql_dump = '\n'.join(con.iterdump()) con.close() return sql_dump

Project Versions