Source code for pacifica.metadata.admin_cmd

# -*- coding: utf-8 -*-
"""Admin Module for Admin Commands."""
from __future__ import print_function
from sys import argv as sys_argv
from datetime import timedelta
from json import loads, dumps
from argparse import ArgumentParser
from peewee import PeeweeException
from .orm.all_objects import ORM_OBJECTS
from .orm.sync import OrmSync, MetadataSystem

[docs]def render_obj(args): """Render an object based on args.""" OrmSync.connect_and_check() test_obj = args.object() test_obj = args.object.get( test_obj.where_clause(args.where_clause) ) print( dumps( test_obj.to_hash(recursion_depth=args.recursion), indent=4 ) )
[docs]def delete_obj(args): """Delete an object based on args.""" OrmSync.connect_and_check() test_obj = args.object() args.where_clause['force'] = args.force args.where_clause['recursive'] = args.recursive # pylint: disable=protected-access test_obj._delete(**args.where_clause)
# pylint: enable=protected-access
[docs]def create_table(args): """Create a specific object.""" OrmSync.dbconn_blocking() if not args.object.table_exists(): args.object.create_table()
[docs]def bool2cmdint(command_bool): """Convert a boolean to either 0 for true or -1 for false.""" if command_bool: return 0 return -1
[docs]def dbchk(args): """Check the database for the version running.""" OrmSync.dbconn_blocking() if args.check_equal: res = MetadataSystem.is_equal() OrmSync.close() return bool2cmdint(res) res = MetadataSystem.is_safe() OrmSync.close() return bool2cmdint(res)
[docs]def dbsync(_args=None): """Create or update the database.""" OrmSync.dbconn_blocking() try: MetadataSystem.get_version() except PeeweeException: OrmSync.dbconn_blocking() OrmSync.create_tables() OrmSync.close() return OrmSync.update_tables() OrmSync.close() return
[docs]def create_subcommands(subparsers): """Create the subcommands from the subparsers.""" create_table_parser = subparsers.add_parser( 'create_table', help='create_table help', description='create table in the DB' ) delete_obj_parser = subparsers.add_parser( 'delete_obj', help='delete_obj help', description='delete an object in the DB' ) render_parser = subparsers.add_parser( 'render', help='render help', description='render and object from database w/o API' ) db_parser = subparsers.add_parser( 'dbsync', help='dbsync help', description='Update or Create the Database.' ) dbchk_parser = subparsers.add_parser( 'dbchk', help='dbchk help', description='Check database against current version.' ) return ( (render_parser, render_options), (create_table_parser, create_table_options), (delete_obj_parser, delete_obj_options), (db_parser, db_options), (dbchk_parser, dbchk_options) )
[docs]def db_options(db_parser): """Add the options for dbsync subcommand.""" db_parser.set_defaults(func=dbsync)
[docs]def dbchk_options(dbchk_parser): """Add the options for dbchk.""" dbchk_parser.add_argument( '--equal', default=False, dest='check_equal', action='store_true' ) dbchk_parser.set_defaults(func=dbchk)
[docs]def objstr_to_timedelta(obj_str): """Turn an object string of the format X unit ago into timedelta.""" value, unit, check = obj_str.split() assert check == 'ago' return timedelta(**{unit: float(value)})
[docs]def objstr_to_ormobj(obj_str): """Convert a string to an orm object or raise ValueError.""" obj_strs = [obj.__name__ for obj in ORM_OBJECTS] if str(obj_str) not in obj_strs: raise ValueError('{} not a valid Metadata Object.'.format(obj_str)) return [obj for obj in ORM_OBJECTS if obj.__name__ == str(obj_str)][0]
[docs]def objstr_to_whereclause(obj_str): """Convert a string to a where clause hash or raise ValueError.""" json_obj = loads(obj_str) if not isinstance(json_obj, dict): raise ValueError('{} should be a dict not {}'.format( json_obj, type(json_obj))) return json_obj
[docs]def delete_obj_options(delete_obj_parser): """Delete the object command line options.""" delete_obj_parser.add_argument( '--where-clause', dest='where_clause', type=objstr_to_whereclause, help='object where args query.', required=True ) delete_obj_parser.add_argument( '--object-name', dest='object', type=objstr_to_ormobj, help='object type to query.', required=True ) delete_obj_parser.add_argument( '--force', dest='force', default=False, action='store_true', help='force delete object.' ) delete_obj_parser.add_argument( '--recursive', dest='recursive', default=False, action='store_true', help='recursive delete related objects.' ) delete_obj_parser.set_defaults(func=delete_obj)
[docs]def create_table_options(create_table_parser): """Add the create object command line options.""" create_table_parser.add_argument( '--object-name', dest='object', type=objstr_to_ormobj, help='object table to create.', required=True ) create_table_parser.set_defaults(func=create_table)
[docs]def render_options(render_parser): """Add the essync command line options.""" render_parser.add_argument( '--object-name', dest='object', type=objstr_to_ormobj, help='object type to query.', required=True ) render_parser.add_argument( '--where-clause', dest='where_clause', type=objstr_to_whereclause, help='query to use.', required=True ) render_parser.add_argument( '--recursion', default=1, dest='recursion', type=int, help='recursive level to go', required=False ) render_parser.set_defaults(func=render_obj)
[docs]def main(*argv): """Main method for admin command line tool.""" parser = ArgumentParser() subparsers = parser.add_subparsers(help='sub-command help') for subparser, options_func in create_subcommands(subparsers): options_func(subparser) if not argv: # pragma: no cover argv = sys_argv[1:] args = parser.parse_args(argv) return args.func(args)