plpipes.runner

Module: plpipes.runner

The purpose of the runner is to offer a unified entry point for the project actions and pipelines. It extracts information from a set of environment variables and parses command line arguments in a standard way.

For more information, refer to Runner.

arg_parser(**kwargs)

Initializes the argument parser for the runner.

Parameters:

Name	Type	Description	Default
kwargs		Additional keyword arguments for customizing the parser.	{}

Returns:

Type	Description
	An instance of argparse.ArgumentParser configured for the runner.

Source code in src\plpipes\runner.py

def arg_parser(**kwargs):
    """
    Initializes the argument parser for the runner.

    Parameters:
        kwargs: Additional keyword arguments for customizing the parser.

    Returns:
        An instance of argparse.ArgumentParser configured for the runner.
    """
    parser = argparse.ArgumentParser(**{**_default_arg_parser_args, **kwargs})
    parser.add_argument('-d', '--debug',
                        help="Turns on debugging",
                        action='store_true')
    parser.add_argument('-c', '--config',
                        action="append",
                        metavar="CFG_FN",
                        help="Additional configuration file",
                        default=[])
    parser.add_argument('-s', '--set',
                        action=_PairAction,
                        metavar="CFG_KEY=VAL",
                        help="Set configuration entry",
                        default=[])
    parser.add_argument('-S', '--set-json',
                        action=_PairAction,
                        metavar="CFG_KEY=JSON_VAL",
                        unpack="json",
                        dest="set",
                        help="Set configuration entry (value is parsed as JSON)")
    parser.add_argument('-e', '--env',
                        metavar="ENVIRONMENT",
                        help="Select environment (dev, pre, pro, etc.)")
    return parser

main(args=None)

Main entry point for the runner. Parses arguments and executes specified actions.

Parameters:

Name	Type	Description	Default
args		The command line arguments to parse (defaults to None).	None

Source code in src\plpipes\runner.py

def main(args=None):
    """
    Main entry point for the runner. Parses arguments and executes specified actions.

    Parameters:
        args: The command line arguments to parse (defaults to None).
    """
    parser = arg_parser()
    parser.add_argument('actions', nargs="*",
                        metavar="ACTION", default=["default"])
    opts = parse_args_and_init(parser, args)

    for action in opts.actions:
        logging.info(f"Executing action {action}")
        plpipes.action.run(action)

parse_args_and_init(arg_parser, args=None)

Parses command line arguments and initializes the PLPipes framework.

Parameters:

Name	Type	Description	Default
arg_parser		The argument parser instance to use for parsing.	required
args		The command line arguments to parse (defaults to sys.argv).	None

Returns:

Type	Description
	The parsed options as an object with attributes corresponding to the arguments.

Raises:

Type	Description
Exception	If the program name is missing from the argument list.

Source code in src\plpipes\runner.py

def parse_args_and_init(arg_parser, args=None):
    """
    Parses command line arguments and initializes the PLPipes framework.

    Parameters:
        arg_parser: The argument parser instance to use for parsing.
        args: The command line arguments to parse (defaults to sys.argv).

    Returns:
        The parsed options as an object with attributes corresponding to the arguments.

    Raises:
        Exception: If the program name is missing from the argument list.
    """
    if args is None:
        args = sys.argv

    if len(args) < 1:
        raise Exception("Unable to infer config stem. Program name missing from argument list")

    prog_path = pathlib.Path(args[0])
    if "PLPIPES_ROOT_DIR" in os.environ:
        root_dir = pathlib.Path(os.environ["PLPIPES_ROOT_DIR"]).resolve(strict=True)
    else:
        root_dir = prog_path.parent.parent
    root_dir = root_dir.absolute()

    opts = arg_parser.parse_args(args[1:])

    config_extra = [{'fs': {'stem': str(prog_path.stem),
                            'root': str(root_dir),
                            'project': str(root_dir.stem)}}]

    config_extra += opts.set

    if opts.env is not None:
        config_extra["env"] = opts.env
    if opts.debug:
        config_extra.append({"logging.level": "debug"})
        config_extra.append({"logging.level_file": "debug"})

    plpipes.init.init(*config_extra, config_files=opts.config)

    sys.path.append(plpipes.config.cfg["fs.lib"])

    os.environ.setdefault("PLPIPES_ROOT_DIR", plpipes.config.cfg["fs.root"])

    return opts

simple_init(args=None)

A simplified initialization function that uses a default argument parser.

Parameters:

Name	Type	Description	Default
args		The command line arguments to parse (defaults to None).	None

Returns:

Type	Description
	The parsed options as an object with attributes corresponding to the arguments.

Source code in src\plpipes\runner.py

def simple_init(args=None):
    """
    A simplified initialization function that uses a default argument parser.

    Parameters:
        args: The command line arguments to parse (defaults to None).

    Returns:
        The parsed options as an object with attributes corresponding to the arguments.
    """
    return parse_args_and_init(arg_parser(), args)