Source code for oio.cli

# Copyright (C) 2018-2019 OpenIO SAS, as part of OpenIO SDS
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero 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
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <>.

import argparse
import os
from cliff import command, lister, show

from oio.common.logger import get_logger


[docs]def make_logger_args_parser(): """Create an ArgumentParser for logger configuration.""" log_parser = argparse.ArgumentParser(add_help=False) log_parser.add_argument('--log-level', choices=LOG_LEVELS, help="Log level") log_parser.add_argument('--log-syslog-prefix', help="Syslog prefix") log_parser.add_argument('--log-facility', help="Log facility") log_parser.add_argument('--log-address', help="Log address") log_parser.add_argument('-q', '--quiet', action='store_true', help="Don't print logs on console") return log_parser
[docs]def get_logger_from_args(args, default_conf=None): """Build a Logger instance from parsed args.""" conf = default_conf or {} if args.log_level is not None: conf['log_level'] = args.log_level if args.log_facility is not None: conf['log_facility'] = args.log_facility if args.log_address is not None: conf['log_address'] = args.log_address if args.log_syslog_prefix is not None: conf['syslog_prefix'] = args.log_syslog_prefix return get_logger(conf, 'log', not args.quiet)
[docs]def add_common_parser_options(parser): """ Add optional parameters common to all openio CLIs to parser. """ parser.add_argument( "--admin-flag", "--admin", dest='admin_mode', action='store_true', help="Add 'admin mode' flag to all requests to oio-proxy.") parser.add_argument( '-a', '--account', '--oio-account', metavar='<account>', dest='account', default=os.environ.get('OIO_ACCOUNT', ''), help='Account name (Env: OIO_ACCOUNT).') parser.add_argument( '--ns', '--oio-ns', metavar='<namespace>', dest='ns', default=os.environ.get('OIO_NS', ''), help='Namespace name (Env: OIO_NS).') parser.add_argument( '--oio-proxy', '--oio-proxyd-url', metavar='<proxy-url>', dest='proxyd_url', default=os.environ.get('OIO_PROXYD_URL', ''), help='URL of an oio-proxy service (Env: OIO_PROXYD_URL).') parser.add_argument( '--request-id', '--req-id', metavar='<request-id>', help=('Set a request ID. Maximum 63 characters. ' 'For looping commands, a suffix may be appended.')) parser.add_argument( "--profile", help=("Profile code, save profiling data in the specified file. " "'%%(pid)s' in the name will be replaced by the PID.")) parser.add_argument( "--profiler", default='cProfile', help=("Which profiler to use (default: cProfile, " "supported: GreenletProfiler, cProfile, profile).")) parser.add_argument( "--profile-early", action='store_true', help=("Start profiling early, before subcommand loading."))
[docs]def flat_dict_from_dict(parsed_args, dict_): """ Create a dictionary without depth. { 'depth0': { 'depth1': { 'depth2': 'test' } } } => { 'depth0.depth1.depth2': 'test' } """ flat_dict = dict() for key, value in dict_.items(): if not isinstance(value, dict): if isinstance(value, list) and parsed_args.formatter == 'table': value = '\n'.join(value) flat_dict[key] = value continue _flat_dict = flat_dict_from_dict(parsed_args, value) for _key, _value in _flat_dict.items(): flat_dict[key + '.' + _key] = _value return flat_dict
[docs]class Command(command.Command): """ Wraps cliff's command.Command and sets the process' return code according to the class' "success" field. """ success = True
[docs] def run(self, parsed_args): return_code = super(Command, self).run(parsed_args) if return_code == 0: return_code = int(not self.success) return return_code
[docs]class Lister(lister.Lister): """ Wraps cliff's lister.Lister and sets the process' return code according to the class' "success" field. """ success = True
[docs] def run(self, parsed_args): super(Lister, self).run(parsed_args) return int(not self.success)
[docs]class ShowOne(show.ShowOne): """ Wraps cliff's show.ShowOne and sets the process' return code according to the class' "success" field. """ success = True
[docs] def run(self, parsed_args): super(ShowOne, self).run(parsed_args) return int(not self.success)