A cli used to run python and inject kwargs. Helpful for remote exection and simple reception of key value pairs.

README.md

The following is a CLI used to be able to run a python callable on a remote machine. Its purpose is the be the recieving end of a list of key=value pairs=:) and will inject this into the **kwargs of the specified callable.

This function was used after deploying code to the remote machine and configurations had been marshalled into the form given above.

Notes on functionality

The main CLI would construct this class by passing in a subparser named run to the parser value. With the following line, this class is automatically pointing the main CLI towards the do_it function when the run subparser is used.

self.parser.set_defaults(func=self.do_it)

An example usage is cli run python.py my=arg hello=bye where run would cause argparse to load the subparser and thus the func. The main CLI would have the following logic to run whatever code is associated with the subparser.

def main():
    args = parser.parse_args()

    # If no sub command is specified, print help and exit
    if not hasattr(args, 'func'):
        print("Error: please use at least one positional argument\n", file=sys.stderr)

        parser.print_help()
        exit(1)

    # Otherwise trigger sub command
    args.func(args)

run_cli.py

import os, sys
import importlib.util
from argparse import ArgumentParser, Namespace

from .util import parse_kv_pairs, patch_dynamic_sys_path


class RunCLI:
    def __init__(self, parser: ArgumentParser):
        self.parser = parser

        self.parser.set_defaults(func=self.do_it)
        self.parser.add_argument(
            'python_file',
            metavar='python-file',
            default=None,
            help="Path to the python file which the callable is located in."
        )
        self.parser.add_argument(
            'python_callable',
            metavar='python-callable',
            default=None,
            help="Name of the python callable inside the `python-file` file to be called."
        )
        self.parser.add_argument(
            '--args',
            metavar="KEY=VALUE",
            nargs="+",
            required=False,
            default=None,
            help="Use --args followed by any number of KEY=VALUE pairs which will be passed to the `**kwargs` of the python callable."
        )

    def do_it(self, args: Namespace):
        # See DeployCLI's do_it for comments on this
        if not isinstance(args, dict):
            args = vars(args)
        
        assert os.path.isfile(args['python_file']), f'Python file provided to `agnt run` does not exist or is directory: {args["python_file"]}'

        file_path = os.path.abspath(args['python_file'])
        file_spec = importlib.util.spec_from_file_location(
            'dynamic_callable',
            file_path
        )
        patch_dynamic_sys_path(file_path)
        file_module = importlib.util.module_from_spec(file_spec)
        file_spec.loader.exec_module(file_module)

        try:
            callable_attr = getattr(file_module, args['python_callable'])
        except AttributeError:
            print(f'Error: python callable "{args["python_callable"]}" does not exist on file "{args["python_file"]}"', file=sys.stderr)
            exit(1)

        callable_attr(**parse_kv_pairs(args['args']))

util.py

def parse_kv_pairs(pairs):
    pairs_dict = {}
    if pairs:
        for pair in pairs:
            items = pair.split('=')
            if len(items) < 0:
                raise RuntimeError(f'Found non key-value pair: "{pair}" please use "=" to delimit key-value pairs (no spaces allowed in keys or pairs).')
            key = items[0].strip('')
            value = '='.join(items[1:])
            
            pairs_dict[key] = value
    return pairs_dict


def patch_dynamic_sys_path(file_path):
    '''
    Python will usually add the directory of an executed file to the `sys.path`. Because both `agnt deploy` and `agnt run` are dynamically loading files to be executed / interpreted this is done manually through this function.
    **NOTE:** This should be called before the module is dynamically imported.
    '''
    sys.path.append(os.path.abspath(os.path.dirname(file_path)))