Source code for anuket.scripts.upgradedb

# -*- coding: utf-8 -*-
""" Script for upgrading the Anuket database with Alembic."""
import argparse
import os
import sys
from datetime import date

from alembic.command import upgrade
from pyramid.paster import get_appsettings

from anuket.lib.alembic_utils import get_alembic_settings

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

[docs]class UpgradeDBCommand(object): """ Upgrade the database using the configuration from the .ini file passed as a positional argument. """ description = 'Upgrade 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('-f', '--force', action='store_true', help='force the upgrade even if there is no available database backup') def __init__(self, argv): """ Get arguments from the ``argparse`` parser.""" self.args = self.parser.parse_args(argv[1:])
[docs] def run(self): """ Run the ``upgrade_db`` method or display the parser help message if the `config_uri` argument is missing. :return: ``upgrade_db`` method or 2 (missing argument error) """ if not self.args.config_uri: self.parser.print_help() return 2 else: return self.upgrade_db()
[docs] def upgrade_db(self): """ Upgrade the database to the head revision with Alembic. :return: 0 (OK) or 1 (abnormal termination error) """ config_uri = self.args.config_uri force = self.args.force 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 there is a database backup isfile = os.path.isfile(path) if not isfile and not force: print("There is no up to date backup for the database. " "Please use the backup script before upgrading!") return 1 # upgrade the database alembic_cfg = get_alembic_settings(config_uri) upgrade(alembic_cfg, 'head') print("Database upgrade done.") return 0

Project Versions