X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=config.py;h=574449f54f7dc57bce107f071affa126d15a00da;hb=b10d30a46e601c9ee1f843241f2d69a1f90f7a94;hp=bc0dcdf14f23b5edd8582a4fd8afcd9425a5bd36;hpb=497fb9e21f45ec08e1486abaee6dfa7b20b8a691;p=python_utils.git diff --git a/config.py b/config.py index bc0dcdf..574449f 100644 --- a/config.py +++ b/config.py @@ -55,53 +55,48 @@ different modules). Usage: --dry_run Should we really do the thing? - Arguments themselves should be accessed via config.config['arg_name']. e.g. + Arguments themselves should be accessed via + config.config['arg_name']. e.g. if not config.config['dry_run']: module.do_the_thing() + """ import argparse +import logging +import os import pprint -import re import sys -from typing import Dict, Any - -# Note: at this point in time, logging hasn't been configured and -# anything we log will come out the root logger. - - -class LoadFromFile(argparse.Action): - """Helper to load a config file into argparse.""" - def __call__ (self, parser, namespace, values, option_string = None): - with values as f: - buf = f.read() - argv = [] - for line in buf.split(','): - line = line.strip() - line = line.strip('{') - line = line.strip('}') - m = re.match(r"^'([a-zA-Z_\-]+)'\s*:\s*(.*)$", line) - if m: - key = m.group(1) - value = m.group(2) - value = value.strip("'") - if value not in ('None', 'True', 'False'): - argv.append(f'--{key}') - argv.append(value) - parser.parse_args(argv, namespace) +from typing import Any, Dict, List, Optional + +# This module is commonly used by others in here and should avoid +# taking any unnecessary dependencies back on them. + +# Defer logging messages until later when logging has been initialized. +saved_messages: List[str] = [] + +# Make a copy of the original program arguments. +program_name = os.path.basename(sys.argv[0]) +original_argv = [arg for arg in sys.argv] # A global parser that we will collect arguments into. args = argparse.ArgumentParser( - description=f"This program uses config.py ({__file__}) for global, cross-module configuration.", + description=None, formatter_class=argparse.ArgumentDefaultsHelpFormatter, - fromfile_prefix_chars="@" + fromfile_prefix_chars="@", + epilog=f'------------------------------------------------------------------------------\n{program_name} uses config.py ({__file__}) for global, cross-module configuration setup and parsing.\n------------------------------------------------------------------------------' ) + +# Keep track of if we've been called and prevent being called more +# than once. config_parse_called = False -# A global configuration dictionary that will contain parsed arguments -# It is also this variable that modules use to access parsed arguments +# A global configuration dictionary that will contain parsed arguments. +# It is also this variable that modules use to access parsed arguments. +# This is the data that is most interesting to our callers; it will hold +# the configuration result. config: Dict[str, Any] = {} @@ -116,11 +111,9 @@ group = add_commandline_args( ) group.add_argument( '--config_loadfile', - type=open, - action=LoadFromFile, metavar='FILENAME', default=None, - help='Config file from which to read args in lieu or in addition to commandline.', + help='Config file (populated via --config_savefile) from which to read args in lieu or in addition to commandline.', ) group.add_argument( '--config_dump', @@ -133,27 +126,141 @@ group.add_argument( type=str, metavar='FILENAME', default=None, - help='Populate config file compatible --config_loadfile to save config for later use.', + help='Populate config file compatible with --config_loadfile to save config for later use.', ) -def parse() -> Dict[str, Any]: - """Main program should call this early in main()""" - global config_parse_called - config_parse_called = True - config.update(vars(args.parse_args())) +def is_flag_already_in_argv(var: str): + """Is a particular flag passed on the commandline?""" + for _ in sys.argv: + if var in _: + return True + return False - if config['config_savefile']: - with open(config['config_savefile'], 'w') as wf: - wf.write("\n".join(sys.argv[1:])) +def parse(entry_module: Optional[str]) -> Dict[str, Any]: + """Main program should call this early in main()""" + global config_parse_called + if config_parse_called: + return config + + global saved_messages + + # If we're about to do the usage message dump, put the main module's + # argument group last in the list (if possible) so that when the user + # passes -h or --help, it will be visible on the screen w/o scrolling. + reordered_action_groups = [] + global prog + for arg in sys.argv: + if arg == '--help' or arg == '-h': + for group in args._action_groups: + if entry_module is not None and entry_module in group.title: + reordered_action_groups.append(group) + elif program_name in group.title: + reordered_action_groups.append(group) + else: + reordered_action_groups.insert(0, group) + args._action_groups = reordered_action_groups + + # Examine the environment variables that match known flags. For a + # flag called --example_flag the corresponding environment + # variable would be called EXAMPLE_FLAG. + usage_message = args.format_usage() + optional = False + var = '' + for x in usage_message.split(): + if x[0] == '[': + optional = True + if optional: + var += f'{x} ' + if x[-1] == ']': + optional = False + var = var.strip() + var = var.strip('[') + var = var.strip(']') + chunks = var.split() + if len(chunks) > 1: + var = var.split()[0] + + # Environment vars the same as flag names without + # the initial -'s and in UPPERCASE. + env = var.strip('-').upper() + if env in os.environ: + if not is_flag_already_in_argv(var): + value = os.environ[env] + saved_messages.append( + f'Initialized from environment: {var} = {value}' + ) + from string_utils import to_bool + if len(chunks) == 1 and to_bool(value): + sys.argv.append(var) + elif len(chunks) > 1: + sys.argv.append(var) + sys.argv.append(value) + var = '' + env = '' + else: + next + + # Look for loadfile and read/parse it if present. + loadfile = None + saw_other_args = False + grab_next_arg = False + for arg in sys.argv[1:]: + if 'config_loadfile' in arg: + pieces = arg.split('=') + if len(pieces) > 1: + loadfile = pieces[1] + else: + grab_next_arg = True + elif grab_next_arg: + loadfile = arg + else: + saw_other_args = True + + if loadfile is not None: + if saw_other_args: + msg = f'Augmenting commandline arguments with those from {loadfile}.' + print(msg, file=sys.stderr) + saved_messages.append(msg) + if not os.path.exists(loadfile): + print(f'ERROR: --config_loadfile argument must be a file, {loadfile} not found.', + file=sys.stderr) + sys.exit(-1) + with open(loadfile, 'r') as rf: + newargs = rf.readlines() + newargs = [arg.strip('\n') for arg in newargs if 'config_savefile' not in arg] + sys.argv += newargs + + # Parse (possibly augmented, possibly completely overwritten) + # commandline args with argparse normally and populate config. + known, unknown = args.parse_known_args() + config.update(vars(known)) + + # Reconstruct the argv with unrecognized flags for the benefit of + # future argument parsers. For example, unittest_main in python + # has some of its own flags. If we didn't recognize it, maybe + # someone else will. + sys.argv = sys.argv[:1] + unknown + + # Check for savefile and populate it if requested. + savefile = config['config_savefile'] + if savefile and len(savefile) > 0: + with open(savefile, 'w') as wf: + wf.write( + "\n".join(original_argv[1:]) + ) + + # Also dump the config on stderr if requested. if config['config_dump']: dump_config() + config_parse_called = True return config def has_been_parsed() -> bool: + """Has the global config been parsed yet?""" global config_parse_called return config_parse_called @@ -162,3 +269,12 @@ def dump_config(): """Print the current config to stdout.""" print("Global Configuration:", file=sys.stderr) pprint.pprint(config, stream=sys.stderr) + print() + + +def late_logging(): + """Log messages saved earlier now that logging has been initialized.""" + logger = logging.getLogger(__name__) + global saved_messages + for _ in saved_messages: + logger.debug(_)