CLIWrapper represents calls to CLI tools as an object with native python function calls.\n\nExamples
\n\nfrom json import loads # or any other parser\nfrom cli_wrapper import CLIWrapper\nkubectl = CLIWrapper('kubectl')\nkubectl._update_command(\"get\", default_flags={\"output\": \"json\"}, parse=loads)\n# this will run `kubectl get pods --namespace kube-system --output json`\nresult = kubectl.get(\"pods\", namespace=\"kube-system\")\nprint(result)\n\nkubectl = CLIWrapper('kubectl', async_=True)\nkubectl._update_command(\"get\", default_flags={\"output\": \"json\"}, parse=loads)\nresult = await kubectl.get(\"pods\", namespace=\"kube-system\") # same thing but async\nprint(result)\n
\n\nYou can also override argument names and provide input validators:
\n\nfrom json import loads\nfrom cli_wrapper import CLIWrapper\nkubectl = CLIWrapper('kubectl')\nkubectl._update_command(\"get_all\", cli_command=\"get\", default_flags={\"output\": \"json\", \"A\": None}, parse=loads)\nresult = kubectl.get_all(\"pods\") # this will run `kubectl get pods -A --output json`\nprint(result)\n\ndef validate_pod_name(name):\n return all(\n len(name) < 253,\n name[0].isalnum() and name[-1].isalnum(),\n all(c.isalnum() or c in ['-', '.'] for c in name[1:-1])\n )\nkubectl._update_command(\"get\", validators={1: validate_pod_name})\nresult = kubectl.get(\"pod\", \"my-pod!!\") # raises ValueError\n
\n\nCallable serialization
\n\nArgument validation and parser configuration are not straightforward to serialize. To get around this, CLI Wrapper uses\nCallableRegistry and CallableChain. These make it somewhat more straightforward to create more serializable wrapper\nconfigurations.
\n\nTL;DR
\n\n\n\nImplementation
\n\nHere's how these work:
\n\nCallableRegistry
\n\nCallable registries form the basis of serializing callables by mapping strings to functions. If you are doing custom\nparsers and validators and you want these to be serializable, you will use their respective callable registries to\nassociate the code with the serializable name.
\n\n\n
def greater_than(a, b):\n return a > b\n\n\nregistry = CallableRegistry(\n {\n "core" = {}\n }\n)\nregistry.register("gt", greater_than)\n\nx = registry.get("gt", [2])\n\nassert(not x(1))\nassert(x(3))\n
\n
\n\nCallableChain
\n\nA callable chain is a serializable structure that gets converted to a sequence of calls to things in a\ncli_wrapper.util.callable_registry.CallableRegistry. It is an abstract base class, and so shouldn't be created directly; subclasses are expected to\nimplement __call__. We'll use the .validators.Validator class as an example. validators is a CallableRegistry with all of\nthe base validators (is_dict, is_list, is_str, startswith...)
\n\n\n
# Say we have these validators that we want to run:\ndef every_letter_is(v, l):\n return all((x == l.lower()) or (x == l.upper()) for x in v)\n\nvalidators.register("every_letter_is", every_letter_is)\n\nmy_validation = ["is_str", {"every_letter_is": "a"}]\n\nstraight_as = Validator(my_validation)\nassert(straight_as("aaaaAAaa"))\nassert(not straight_as("aaaababa"))\n
\n
\n\nValidator.__call__ just checks that every validation returns true. Elsewhere, Parser pipes inputs in sequence:
\n\n\n
parser:\n - yaml\n - extract: result \n
\n
\n\nThis would first parse the output as yaml and then extract the \"result\" key from the dictionary returned by the yaml\nstep.
\n\nfrom curses import wrapper
\n\nValidators
\n\nValidators are used to validate argument values. They are implemented as a\ncli_wrapper.util.callable_chain.CallableChain for serialization. Callables in the chain are called with the value\nsequentially, stopping at the first callable that returns False.
\n\nDefault Validators
\n\nThe default validators are:
\n\n\nis_dict \nis_list \nis_str \nis_str_or_list \nis_int \nis_float \nis_bool \nis_path - is a pathlib.Path \nis_alnum - is alphanumeric \nis_alpha - is alphabetic \nstarts_alpha - first digit is a letter \nstartswith - checks if the string starts with a given prefix \n
\n\nCustom Validators
\n\nYou can register your own validators in cli_wrapper.validators.validators:
\n\n\n- Takes at most one positional argument
\n- When configuring the validator, additional arguments can be supplied using a dictionary:
\n
\n\n\n
wrapper.update_command_("cmd", validators={"arg":["is_str", {"startswith": {"prefix": "prefix"}}]})\n# or\nwrapper.update_command_("cmd", validators={"arg": ["is_str", {"startswith": "prefix"}]})\n
\n
\n\nExample
\n\n\n
from cli_wrapper import CLIWrapper\nfrom cli_wrapper.validators import validators\n\ndef is_alnum_or_dash(value):\n return all(c.isalnum() or c == "-" for c in value)\nvalidators.register("is_alnum_or_dash", is_alnum_or_dash)\n\nkubectl = CLIWrapper("kubectl")\n# 1 refers to the first positional argument, so in `kubectl.get("pods", "my-pod")` it would refer to `"my-pod"`\nkubectl.update_command_("get", validators={\n 1: ["is_str", "is_alnum_or_dash", "starts_alpha"],\n})\n\nassert kubectl.get("pods", "my-pod")\nthrew = False\ntry:\n kubectl.get("pods", "level-9000-pod!!")\nexcept ValueError:\n threw = True\nassert threw\n
\n
\n\nParsers
\n\nParsers provide a mechanism to convert the output of a CLI tool into a usable structure. They make use of\ncli_wrapper.util.callable_chain.CallableChain to be serializable-ish.
\n\nDefault Parsers
\n\n\njson: uses json.loads to parse stdout \nextract: extracts data from the raw output, using the args as a list of nested keys. \nyaml: if ruamel.yaml is installed, uses YAML().load_all to read stdout. If load_all only returns one\ndocument, it returns that document. Otherwise, it returns a list of documents. pyyaml is also supported. \ndotted_dict: if dotted_dict is installed, converts an input dict or list to a PreserveKeysDottedDict or \na list of them. This lets you refer to most dictionary keys as a.b.c instead of a[\"b\"][\"c\"]. \n
\n\nThese can be combined in a list in the parse argument to cli_wrapper.cli_wrapper.CLIWrapper.update_command_,\nallowing the result of the call to be immediately usable.
\n\nYou can also register your own parsers in cli_wrapper.parsers.parsers, which is a \ncli_wrapper.util.callable_registry.CallableRegistry.
\n\nExamples
\n\n\n
from cli_wrapper import CLIWrapper\n\ndef skip_lists(result): \n if result["kind"] == "List":\n return result["items"]\n return result\n\nkubectl = CLIWrapper("kubectl")\n# you can use the parser directly, but you won't be able to serialize the\n# wrapper to json\nkubectl.update_command_(\n "get",\n parse=["json", skip_lists, "dotted_dict"],\n default_flags=["--output", "json"]\n)\n\na = kubectl.get("pods", namespace="kube-system")\nassert isinstance(a, list)\nb = kubectl.get("pods", a[0].metadata.name, namespace="kube-system")\nassert isinstance(b, dict)\nassert b.metadata.name == a[0].metadata.name\n
\n
\n\n\n\nArgument transformers receive an argument (either a numbered positional argument or a string keywork argument/flag) and\na value. They return a tuple of argument and value that replace the original.
\n\nThe main transformer used by cli-wrapper is cli_wrapper.transformers.snake2kebab, which converts a an_argument_like_this to\nan-argument-like-this and returns the value unchanged. This is the default transformer for all keyword arguments.
\n\nTransformers are added to a callable registry, so they can be refernced as a string after they're registered.\nTransformers are not currently chained.
\n\n\n\n1. Write dictionaries to files and return a flag referencing a file
\n\nConsider a command like kubectl create: the primary argument is a filename or list of files. Say you have your \nmanifest to create as a dictionary:
\n\n\n
from pathlib import Path\nfrom ruamel.yaml import YAML\nfrom cli_wrapper import transformers, CLIWrapper\n\nmanifest_count = 0\nbase_filename = "my_manifest"\nbase_dir = Path()\ny = YAML()\ndef write_manifest(manifest: dict | list[dict]):\n global manifest_count\n manifest_count += 1\n file = base_dir / f"{base_filename}_{manifest_count}.yaml"\n with file.open("w") as f:\n if isinstance(manifest, list):\n y.dump_all(manifest, f)\n else:\n y.dump(manifest, f)\n return file.as_posix()\n\ndef manifest_transformer(arg, value, writer=write_manifest):\n return "filename", writer(value)\n\ntransformers.register("manifest", manifest_transformer)\n\n# If you had different writer functions (e.g., different base name), you could register those as partials:\nfrom functools import partial\ntransformers.register("other_manifest", partial(manifest_transformer, writer=my_other_writer))\n\nkubectl = CLIWrapper('kubectl')\nkubectl.update_command_("create", args={"data": {"transformer": "manifest"}})\n\n# will write the manifest to "my_manifest_1.yaml" and execute `kubectl create -f my_manifest_1.yaml`\nkubectl.create(data=my_kubernetes_manifest)\n
\n
\n\nPossible future changes
\n\n\n- it might make sense to make transformers a
CallableChain similar to parser so a sequence of things can be done on an arg \n- it might also make sense to support transformers that break individual args into multiple args with separate values
\n
\n"}, {"fullname": "cli_wrapper.cli_wrapper", "modulename": "cli_wrapper.cli_wrapper", "kind": "module", "doc": "\n"}, {"fullname": "cli_wrapper.cli_wrapper.Argument", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Argument", "kind": "class", "doc": "Argument represents a command line argument to be passed to the cli_wrapper
\n"}, {"fullname": "cli_wrapper.cli_wrapper.Argument.__init__", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Argument.__init__", "kind": "function", "doc": "Method generated by attrs for class Argument.
\n", "signature": "(\tliteral_name: str | None = None,\tdefault: str = None,\tvalidator=None,\ttransformer: Union[Callable, str, dict, list[str | dict]] = 'snake2kebab')"}, {"fullname": "cli_wrapper.cli_wrapper.Argument.from_dict", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Argument.from_dict", "kind": "function", "doc": "Create an Argument from a dictionary
\n\nParameters
\n\n\n- arg_dict: the dictionary to be converted
\n
\n\nReturns
\n\n\n Argument object
\n
\n", "signature": "(cls, arg_dict):", "funcdef": "def"}, {"fullname": "cli_wrapper.cli_wrapper.Argument.to_dict", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Argument.to_dict", "kind": "function", "doc": "Convert the Argument to a dictionary
\n\nReturns
\n\n\n the dictionary representation of the Argument
\n
\n", "signature": "(self):", "funcdef": "def"}, {"fullname": "cli_wrapper.cli_wrapper.Argument.is_valid", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Argument.is_valid", "kind": "function", "doc": "Validate the value of the argument
\n\nParameters
\n\n\n- value: the value to be validated
\n
\n\nReturns
\n\n\n True if valid, False otherwise
\n
\n", "signature": "(self, value):", "funcdef": "def"}, {"fullname": "cli_wrapper.cli_wrapper.Argument.transform", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Argument.transform", "kind": "function", "doc": "Transform the name and value of the argument
\n\nParameters
\n\n\n- name: the name of the argument
\n- value: the value to be transformed
\n
\n\nReturns
\n\n\n the transformed value
\n
\n", "signature": "(self, name, value, **kwargs):", "funcdef": "def"}, {"fullname": "cli_wrapper.cli_wrapper.Command", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Command", "kind": "class", "doc": "Command represents a command to be run with the cli_wrapper
\n"}, {"fullname": "cli_wrapper.cli_wrapper.Command.__init__", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Command.__init__", "kind": "function", "doc": "Method generated by attrs for class Command.
\n", "signature": "(\tcli_command: str | list[str],\tdefault_flags: dict = {},\targs: dict = NOTHING,\tparse=None,\tdefault_transformer: str = 'snake2kebab',\tshort_prefix: str = '-',\tlong_prefix: str = '--',\targ_separator: str = '=')"}, {"fullname": "cli_wrapper.cli_wrapper.Command.from_dict", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Command.from_dict", "kind": "function", "doc": "Create a Command from a dictionary
\n\nParameters
\n\n\n- command_dict: the dictionary to be converted
\n
\n\nReturns
\n\n\n Command object
\n
\n", "signature": "(cls, command_dict, **kwargs):", "funcdef": "def"}, {"fullname": "cli_wrapper.cli_wrapper.Command.to_dict", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Command.to_dict", "kind": "function", "doc": "Convert the Command to a dictionary.\nExcludes prefixes/separators, because they are set in the CLIWrapper
\n\nReturns
\n\n\n the dictionary representation of the Command
\n
\n", "signature": "(self):", "funcdef": "def"}, {"fullname": "cli_wrapper.cli_wrapper.Command.validate_args", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Command.validate_args", "kind": "function", "doc": "\n", "signature": "(self, *args, **kwargs):", "funcdef": "def"}, {"fullname": "cli_wrapper.cli_wrapper.Command.build_args", "modulename": "cli_wrapper.cli_wrapper", "qualname": "Command.build_args", "kind": "function", "doc": "\n", "signature": "(self, *args, **kwargs):", "funcdef": "def"}, {"fullname": "cli_wrapper.cli_wrapper.CLIWrapper", "modulename": "cli_wrapper.cli_wrapper", "qualname": "CLIWrapper", "kind": "class", "doc": "Parameters
\n\n\n- path: The path to the CLI tool. This will be passed to subprocess directly, and does not require a full path\nunless the tool is not in the system path.
\n- env: A dict of environment variables to be set in the subprocess environment, in addition to and overriding\nthose in os.environ.
\n- trusting: If True, the wrapper will accept any command and pass them to the cli with default configuration.\nOtherwise, it will only allow commands that have been defined with
update_command_ \n- raise_exc: If True, the wrapper will raise an exception if a command returns a non-zero exit code.
\n- async_: If true, the wrapper will return coroutines that must be awaited.
\n- default_transformer: The transformer configuration to apply to all arguments. The default of snake2kebab will\nconvert pythonic_snake_case_kwargs to kebab-case-arguments
\n- short_prefix: The string prefix for single-letter arguments
\n- long_prefix: The string prefix for arguments longer than 1 letter
\n- arg_separator: The character that separates argument values from names. Defaults to '=', so\nwrapper.command(arg=value) would become \"wrapper command --arg=value\"
\n
\n"}, {"fullname": "cli_wrapper.cli_wrapper.CLIWrapper.__init__", "modulename": "cli_wrapper.cli_wrapper", "qualname": "CLIWrapper.__init__", "kind": "function", "doc": "Method generated by attrs for class CLIWrapper.
\n", "signature": "(\tpath: str,\tenv: dict[str, str] = None,\tcommands: dict[str, cli_wrapper.cli_wrapper.Command] = {},\ttrusting: bool = True,\traise_exc: bool = False,\tasync_: bool = False,\tdefault_transformer: str = 'snake2kebab',\tshort_prefix: str = '-',\tlong_prefix: str = '--',\targ_separator: str = '=')"}, {"fullname": "cli_wrapper.cli_wrapper.CLIWrapper.update_command_", "modulename": "cli_wrapper.cli_wrapper", "qualname": "CLIWrapper.update_command_", "kind": "function", "doc": "update the command to be run with the cli_wrapper
\n\nParameters
\n\n\n- command: the command name for the wrapper
\n- cli_command: the command to be run, if different from the command name
\n- args: the arguments passed to the command
\n- default_flags: default flags to be used with the command
\n- parse: function to parse the output of the command
\n
\n\nReturns
\n", "signature": "(\tself,\tcommand: str,\t*,\tcli_command: str | list[str] = None,\targs: dict[str | int, any] = None,\tdefault_flags: dict = None,\tparse=None):", "funcdef": "def"}, {"fullname": "cli_wrapper.cli_wrapper.CLIWrapper.from_dict", "modulename": "cli_wrapper.cli_wrapper", "qualname": "CLIWrapper.from_dict", "kind": "function", "doc": "Create a CLIWrapper from a dictionary
\n\nParameters
\n\n\n- cliwrapper_dict: the dictionary to be converted
\n
\n\nReturns
\n\n\n CLIWrapper object
\n
\n", "signature": "(cls, cliwrapper_dict):", "funcdef": "def"}, {"fullname": "cli_wrapper.cli_wrapper.CLIWrapper.to_dict", "modulename": "cli_wrapper.cli_wrapper", "qualname": "CLIWrapper.to_dict", "kind": "function", "doc": "Convert the CLIWrapper to a dictionary
\n\nReturns
\n\n\n a dictionary that can be used to recreate the wrapper using from_dict
\n
\n", "signature": "(self):", "funcdef": "def"}, {"fullname": "cli_wrapper.parsers", "modulename": "cli_wrapper.parsers", "kind": "module", "doc": "\n"}, {"fullname": "cli_wrapper.parsers.extract", "modulename": "cli_wrapper.parsers", "qualname": "extract", "kind": "function", "doc": "Extracts a sub-dictionary from a source dictionary based on a given path.\nTODO: this
\n\nParameters
\n\n\n- src: The source dictionary to extract from.
\n- path: A list of keys representing the path to the sub-dictionary.
\n
\n\nReturns
\n\n\n The extracted sub-dictionary.
\n
\n", "signature": "(src: dict, *args) -> dict:", "funcdef": "def"}, {"fullname": "cli_wrapper.parsers.core_parsers", "modulename": "cli_wrapper.parsers", "qualname": "core_parsers", "kind": "variable", "doc": "\n", "default_value": "{'extract': <function extract>, 'json': <function loads>, 'yaml': <function yaml_loads>, 'dotted_dict': <function dotted_dictify>}"}, {"fullname": "cli_wrapper.parsers.parsers", "modulename": "cli_wrapper.parsers", "qualname": "parsers", "kind": "variable", "doc": "A CallableRegistry of parsers. These can be chained in sequence to perform \noperations on input.
\n\nDefaults:\ncore parsers:
\n\n\n- json - parses the input as json, returns the result
\n- extract - extracts the specified sub-dictionary from the source dictionary
\n- yaml - parses the input as yaml, returns the result (requires ruamel.yaml or pyyaml)
\n- dotted_dict - converts an input dictionary to a dotted_dict (requires dotted_dict)
\n
\n", "default_value": "CallableRegistry(_all={'core': {'extract': <function extract>, 'json': <function loads>, 'yaml': <function yaml_loads>, 'dotted_dict': <function dotted_dictify>}}, callable_name='Parser')"}, {"fullname": "cli_wrapper.parsers.Parser", "modulename": "cli_wrapper.parsers", "qualname": "Parser", "kind": "class", "doc": "@public\nParser class that allows for the chaining of multiple parsers. Callables in the configuration are run as a\npipeline, with the output of one parser being passed as input to the next.
\n", "bases": "cli_wrapper.util.callable_chain.CallableChain"}, {"fullname": "cli_wrapper.parsers.Parser.__init__", "modulename": "cli_wrapper.parsers", "qualname": "Parser.__init__", "kind": "function", "doc": "@public
\n\nParameters
\n\n\n- config: a callable, a string, a dictionary with one key and config, or a list of the previous
\n- source: a
CallableRegistry to get callables from \n
\n", "signature": "(config)"}, {"fullname": "cli_wrapper.parsers.yaml_loads", "modulename": "cli_wrapper.parsers", "qualname": "yaml_loads", "kind": "function", "doc": "\n", "signature": "(src: str) -> dict:", "funcdef": "def"}, {"fullname": "cli_wrapper.parsers.dotted_dictify", "modulename": "cli_wrapper.parsers", "qualname": "dotted_dictify", "kind": "function", "doc": "\n", "signature": "(src, *args, **kwargs):", "funcdef": "def"}, {"fullname": "cli_wrapper.pre_packaged", "modulename": "cli_wrapper.pre_packaged", "kind": "module", "doc": "\n"}, {"fullname": "cli_wrapper.pre_packaged.get_wrapper", "modulename": "cli_wrapper.pre_packaged", "qualname": "get_wrapper", "kind": "function", "doc": "Gets a wrapper defined in the beta/stable folders as json.
\n\nParameters
\n\n\n- name: the name of the wrapper to retrieve
\n- status: stable/beta/None. None will search stable and beta
\n
\n\nReturns
\n\n\n the requested wrapper
\n
\n", "signature": "(name, status=None):", "funcdef": "def"}, {"fullname": "cli_wrapper.transformers", "modulename": "cli_wrapper.transformers", "kind": "module", "doc": "\n"}, {"fullname": "cli_wrapper.transformers.snake2kebab", "modulename": "cli_wrapper.transformers", "qualname": "snake2kebab", "kind": "function", "doc": "snake.gravity = 0
\n\nconverts a snake_case argument to a kebab-case one
\n", "signature": "(arg: str, value: <built-in function any>) -> tuple[str, any]:", "funcdef": "def"}, {"fullname": "cli_wrapper.transformers.transformers", "modulename": "cli_wrapper.transformers", "qualname": "transformers", "kind": "variable", "doc": "A callable registry of transformers.
\n\nDefaults:\ncore group:
\n\n\n", "default_value": "CallableRegistry(_all={'core': {'snake2kebab': <function snake2kebab>}}, callable_name='Callable thing')"}, {"fullname": "cli_wrapper.util", "modulename": "cli_wrapper.util", "kind": "module", "doc": "\n"}, {"fullname": "cli_wrapper.util.CallableRegistry", "modulename": "cli_wrapper.util", "qualname": "CallableRegistry", "kind": "class", "doc": "Stores collections of callables. @public
\n\n\n- callables are registered by name
\n- they are retrieved by name with args and kwargs
\n- calling the callable with positional arguments will call the callable\nwith the args in the call, plus any args and kwargs passed to get()
\n
\n"}, {"fullname": "cli_wrapper.util.CallableRegistry.__init__", "modulename": "cli_wrapper.util", "qualname": "CallableRegistry.__init__", "kind": "function", "doc": "Method generated by attrs for class CallableRegistry.
\n", "signature": "(\tall: dict[str, dict[str, typing.Callable]],\tcallable_name: str = 'Callable thing')"}, {"fullname": "cli_wrapper.util.CallableRegistry.callable_name", "modulename": "cli_wrapper.util", "qualname": "CallableRegistry.callable_name", "kind": "variable", "doc": "a name of the things in the registry to use in error messages
\n", "annotation": ": str"}, {"fullname": "cli_wrapper.util.CallableRegistry.get", "modulename": "cli_wrapper.util", "qualname": "CallableRegistry.get", "kind": "function", "doc": "Retrieves a callable function based on the specified parser name.
\n\nParameters
\n\n\n- name: The name of the callable to retrieve.
\n
\n\nReturns
\n\n\n The corresponding callable function.
\n
\n\nRaises
\n\n\n- KeyError: If the specified callable name is not found.
\n
\n", "signature": "(self, name: Union[str, Callable], args=None, kwargs=None) -> Callable:", "funcdef": "def"}, {"fullname": "cli_wrapper.util.CallableRegistry.register", "modulename": "cli_wrapper.util", "qualname": "CallableRegistry.register", "kind": "function", "doc": "Registers a new callable function with the specified name.
\n\nParameters
\n\n\n- name: The name to associate with the callable.
\n- callable_: The callable function to register.
\n
\n", "signature": "(\tself,\tname: str,\tcallable_: <built-in function callable>,\tgroup='core'):", "funcdef": "def"}, {"fullname": "cli_wrapper.util.CallableRegistry.register_group", "modulename": "cli_wrapper.util", "qualname": "CallableRegistry.register_group", "kind": "function", "doc": "Registers a new callable group with the specified name.
\n\nParameters
\n\n\n- name: The name to associate with the callable group.
\n- callables: A dictionary of callables to register in the group.
\n
\n", "signature": "(self, name: str, callables: dict = None):", "funcdef": "def"}, {"fullname": "cli_wrapper.util.CallableChain", "modulename": "cli_wrapper.util", "qualname": "CallableChain", "kind": "class", "doc": "A callable object representing a collection of callables.
\n", "bases": "abc.ABC"}, {"fullname": "cli_wrapper.util.CallableChain.__init__", "modulename": "cli_wrapper.util", "qualname": "CallableChain.__init__", "kind": "function", "doc": "@public
\n\nParameters
\n\n\n- config: a callable, a string, a dictionary with one key and config, or a list of the previous
\n- source: a
CallableRegistry to get callables from \n
\n", "signature": "(config, source)"}, {"fullname": "cli_wrapper.util.CallableChain.chain", "modulename": "cli_wrapper.util", "qualname": "CallableChain.chain", "kind": "variable", "doc": "\n", "annotation": ": list[callable]"}, {"fullname": "cli_wrapper.util.CallableChain.config", "modulename": "cli_wrapper.util", "qualname": "CallableChain.config", "kind": "variable", "doc": "\n", "annotation": ": list"}, {"fullname": "cli_wrapper.util.CallableChain.to_dict", "modulename": "cli_wrapper.util", "qualname": "CallableChain.to_dict", "kind": "function", "doc": "\n", "signature": "(self):", "funcdef": "def"}, {"fullname": "cli_wrapper.validators", "modulename": "cli_wrapper.validators", "kind": "module", "doc": "\n"}, {"fullname": "cli_wrapper.validators.core_validators", "modulename": "cli_wrapper.validators", "qualname": "core_validators", "kind": "variable", "doc": "\n", "default_value": "{'is_dict': <function <lambda>>, 'is_list': <function <lambda>>, 'is_str': <function <lambda>>, 'is_str_or_list': <function <lambda>>, 'is_int': <function <lambda>>, 'is_bool': <function <lambda>>, 'is_float': <function <lambda>>, 'is_alnum': <function <lambda>>, 'is_alpha': <function <lambda>>, 'is_digit': <function <lambda>>, 'is_path': <function <lambda>>, 'starts_alpha': <function <lambda>>, 'startswith': <function <lambda>>}"}, {"fullname": "cli_wrapper.validators.validators", "modulename": "cli_wrapper.validators", "qualname": "validators", "kind": "variable", "doc": "\n", "default_value": "CallableRegistry(_all={'core': {'is_dict': <function <lambda>>, 'is_list': <function <lambda>>, 'is_str': <function <lambda>>, 'is_str_or_list': <function <lambda>>, 'is_int': <function <lambda>>, 'is_bool': <function <lambda>>, 'is_float': <function <lambda>>, 'is_alnum': <function <lambda>>, 'is_alpha': <function <lambda>>, 'is_digit': <function <lambda>>, 'is_path': <function <lambda>>, 'starts_alpha': <function <lambda>>, 'startswith': <function <lambda>>}}, callable_name='Validator')"}, {"fullname": "cli_wrapper.validators.Validator", "modulename": "cli_wrapper.validators", "qualname": "Validator", "kind": "class", "doc": "@public
\n\nA class that provides a validation mechanism for input data.\nIt uses a list of validators to check if the input data is valid.\nThey are executed in sequence until one fails.
\n", "bases": "cli_wrapper.util.callable_chain.CallableChain"}, {"fullname": "cli_wrapper.validators.Validator.__init__", "modulename": "cli_wrapper.validators", "qualname": "Validator.__init__", "kind": "function", "doc": "@public
\n\nParameters
\n\n\n- config: a callable, a string, a dictionary with one key and config, or a list of the previous
\n- source: a
CallableRegistry to get callables from \n
\n", "signature": "(config)"}, {"fullname": "cli_wrapper.validators.Validator.config", "modulename": "cli_wrapper.validators", "qualname": "Validator.config", "kind": "variable", "doc": "\n"}, {"fullname": "cli_wrapper.validators.Validator.to_dict", "modulename": "cli_wrapper.validators", "qualname": "Validator.to_dict", "kind": "function", "doc": "Converts the validator configuration to a dictionary.
\n", "signature": "(self):", "funcdef": "def"}];
+
+ // mirrored in build-search-index.js (part 1)
+ // Also split on html tags. this is a cheap heuristic, but good enough.
+ elasticlunr.tokenizer.setSeperator(/[\s\-.;&_'"=,()]+|<[^>]*>/);
+
+ let searchIndex;
+ if (docs._isPrebuiltIndex) {
+ console.info("using precompiled search index");
+ searchIndex = elasticlunr.Index.load(docs);
+ } else {
+ console.time("building search index");
+ // mirrored in build-search-index.js (part 2)
+ searchIndex = elasticlunr(function () {
+ this.pipeline.remove(elasticlunr.stemmer);
+ this.pipeline.remove(elasticlunr.stopWordFilter);
+ this.addField("qualname");
+ this.addField("fullname");
+ this.addField("annotation");
+ this.addField("default_value");
+ this.addField("signature");
+ this.addField("bases");
+ this.addField("doc");
+ this.setRef("fullname");
+ });
+ for (let doc of docs) {
+ searchIndex.addDoc(doc);
+ }
+ console.timeEnd("building search index");
+ }
+
+ return (term) => searchIndex.search(term, {
+ fields: {
+ qualname: {boost: 4},
+ fullname: {boost: 2},
+ annotation: {boost: 2},
+ default_value: {boost: 2},
+ signature: {boost: 2},
+ bases: {boost: 2},
+ doc: {boost: 1},
+ },
+ expand: true
+ });
+})();
\ No newline at end of file
diff --git a/src/cli_wrapper/__init__.py b/src/cli_wrapper/__init__.py
index d032bbc..aa802b0 100644
--- a/src/cli_wrapper/__init__.py
+++ b/src/cli_wrapper/__init__.py
@@ -1,5 +1,45 @@
-from .cli_wrapper import CLIWrapper
-from .transformers import transformers
-from .parsers import parsers
+"""
+CLIWrapper represents calls to CLI tools as an object with native python function calls.
-__all__ = ["CLIWrapper", "transformers", "parsers"]
+# Examples
+
+```
+from json import loads # or any other parser
+from cli_wrapper import CLIWrapper
+kubectl = CLIWrapper('kubectl')
+kubectl._update_command("get", default_flags={"output": "json"}, parse=loads)
+# this will run `kubectl get pods --namespace kube-system --output json`
+result = kubectl.get("pods", namespace="kube-system")
+print(result)
+
+kubectl = CLIWrapper('kubectl', async_=True)
+kubectl._update_command("get", default_flags={"output": "json"}, parse=loads)
+result = await kubectl.get("pods", namespace="kube-system") # same thing but async
+print(result)
+```
+
+You can also override argument names and provide input validators:
+```
+from json import loads
+from cli_wrapper import CLIWrapper
+kubectl = CLIWrapper('kubectl')
+kubectl._update_command("get_all", cli_command="get", default_flags={"output": "json", "A": None}, parse=loads)
+result = kubectl.get_all("pods") # this will run `kubectl get pods -A --output json`
+print(result)
+
+def validate_pod_name(name):
+ return all(
+ len(name) < 253,
+ name[0].isalnum() and name[-1].isalnum(),
+ all(c.isalnum() or c in ['-', '.'] for c in name[1:-1])
+ )
+kubectl._update_command("get", validators={1: validate_pod_name})
+result = kubectl.get("pod", "my-pod!!") # raises ValueError
+```
+.. include:: ../../doc/callable_serialization.md
+
+.. include:: ../../doc/validators.md
+.. include:: ../../doc/parsers.md
+.. include:: ../../doc/transformers.md
+
+"""
diff --git a/src/cli_wrapper/cli_wrapper.py b/src/cli_wrapper/cli_wrapper.py
index d8c03fd..b343770 100644
--- a/src/cli_wrapper/cli_wrapper.py
+++ b/src/cli_wrapper/cli_wrapper.py
@@ -1,57 +1,10 @@
-"""
-CLIWrapper represents calls to CLI tools as an object with native python function calls.
-For example:
-``` python
-from json import loads # or any other parser
-from cli_wrapper import CLIWrapper
-kubectl = CLIWrapper('kubectl')
-kubectl._update_command("get", default_flags={"output": "json"}, parse=loads)
-# this will run `kubectl get pods --namespace kube-system --output json`
-result = kubectl.get("pods", namespace="kube-system")
-print(result)
-
-kubectl = CLIWrapper('kubectl', async_=True)
-kubectl._update_command("get", default_flags={"output": "json"}, parse=loads)
-result = await kubectl.get("pods", namespace="kube-system") # same thing but async
-print(result)
-```
-
-You can also override argument names and provide input validators:
-``` python
-from json import loads
-from cli_wrapper import CLIWrapper
-kubectl = CLIWrapper('kubectl')
-kubectl._update_command("get_all", cli_command="get", default_flags={"output": "json", "A": None}, parse=loads)
-result = kubectl.get_all("pods") # this will run `kubectl get pods -A --output json`
-print(result)
-
-def validate_pod_name(name):
- return all(
- len(name) < 253,
- name[0].isalnum() and name[-1].isalnum(),
- all(c.isalnum() or c in ['-', '.'] for c in name[1:-1])
- )
-kubectl._update_command("get", validators={1: validate_pod_name})
-result = kubectl.get("pod", "my-pod!!") # raises ValueError
-```
-
-Attributes:
- trusting: if false, only run defined commands, and validate any arguments that have validation. If true, run
- any command. This is useful for cli tools that have a lot of commands that you probably won't use, or for
- YOLO development.
- default_converter: if an argument for a command isn't defined, it will be passed to this. By default, it will
- just convert the name to kebab-case. This is useful for commands that have a lot of (rarely-used) arguments
- that you don't want to bother defining.
- arg_separator: what to put between a flag and its value. default is '=', so `command(arg=val)` would translate
- to `command --arg=val`. If you want to use spaces instead, set this to ' '
-"""
-
import asyncio.subprocess
import logging
import os
import subprocess
from copy import copy
from itertools import chain
+from typing import Callable
from attrs import define, field
@@ -59,7 +12,7 @@ def validate_pod_name(name):
from .transformers import transformers
from .validators import validators, Validator
-logger = logging.getLogger(__name__)
+_logger = logging.getLogger(__name__)
@define
@@ -69,9 +22,13 @@ class Argument:
"""
literal_name: str | None = None
+ """ @private """
default: str = None
+ """ @private """
validator: Validator | str | dict | list[str | dict] = field(converter=Validator, default=None)
- transformer: str = "snake2kebab"
+ """ @private """
+ transformer: Callable | str | dict | list[str | dict] = "snake2kebab"
+ """ @private """
@classmethod
def from_dict(cls, arg_dict):
@@ -92,7 +49,7 @@ def to_dict(self):
Convert the Argument to a dictionary
:return: the dictionary representation of the Argument
"""
- logger.debug(f"Converting argument {self.literal_name} to dict")
+ _logger.debug(f"Converting argument {self.literal_name} to dict")
return {
"literal_name": self.literal_name,
"default": self.default,
@@ -105,12 +62,12 @@ def is_valid(self, value):
:param value: the value to be validated
:return: True if valid, False otherwise
"""
- logger.debug(f"Validating {self.literal_name} with value {value}")
+ _logger.debug(f"Validating {self.literal_name} with value {value}")
return validators.get(self.validator)(value) if self.validator is not None else True
def transform(self, name, value, **kwargs):
"""
- Transform the value of the argument
+ Transform the name and value of the argument
:param name: the name of the argument
:param value: the value to be transformed
:return: the transformed value
@@ -120,7 +77,7 @@ def transform(self, name, value, **kwargs):
)
-def cli_command_converter(value: str | list[str]):
+def _cli_command_converter(value: str | list[str]):
if value is None:
return []
if isinstance(value, str):
@@ -128,7 +85,7 @@ def cli_command_converter(value: str | list[str]):
return value
-def arg_converter(value: dict):
+def _arg_converter(value: dict):
"""
Convert the value of the argument to a string
:param value: the value to be converted
@@ -154,14 +111,22 @@ class Command: # pylint: disable=too-many-instance-attributes
Command represents a command to be run with the cli_wrapper
"""
- cli_command: list[str] | str = field(converter=cli_command_converter)
+ cli_command: list[str] | str = field(converter=_cli_command_converter)
+ """ @private """
default_flags: dict = {}
- args: dict[str | int, any] = field(factory=dict, converter=arg_converter)
+ """ @private """
+ args: dict[str | int, any] = field(factory=dict, converter=_arg_converter)
+ """ @private """
parse: Parser = field(converter=Parser, default=None)
+ """ @private """
default_transformer: str = "snake2kebab"
+ """ @private """
short_prefix: str = field(repr=False, default="-")
+ """ @private """
long_prefix: str = field(repr=False, default="--")
+ """ @private """
arg_separator: str = field(repr=False, default="=")
+ """ @private """
@classmethod
def from_dict(cls, command_dict, **kwargs):
@@ -192,7 +157,7 @@ def to_dict(self):
Excludes prefixes/separators, because they are set in the CLIWrapper
:return: the dictionary representation of the Command
"""
- logger.debug(f"Converting command {self.cli_command} to dict")
+ _logger.debug(f"Converting command {self.cli_command} to dict")
return {
"cli_command": self.cli_command,
"default_flags": self.default_flags,
@@ -203,9 +168,9 @@ def to_dict(self):
def validate_args(self, *args, **kwargs):
# TODO: validate everything and raise comprehensive exception instead of just the first one
for name, arg in chain(enumerate(args), kwargs.items()):
- logger.debug(f"Validating arg {name} with value {arg}")
+ _logger.debug(f"Validating arg {name} with value {arg}")
if name in self.args:
- logger.debug("Argument found in args")
+ _logger.debug("Argument found in args")
v = self.args[name].is_valid(arg)
if isinstance(name, int):
name += 1 # let's call positional arg 0, "Argument 1"
@@ -222,13 +187,13 @@ def build_args(self, *args, **kwargs):
for arg, value in chain(
enumerate(args), kwargs.items(), [(k, v) for k, v in self.default_flags.items() if k not in kwargs]
):
- logger.debug(f"arg: {arg}, value: {value}")
+ _logger.debug(f"arg: {arg}, value: {value}")
if arg in self.args:
literal_arg = self.args[arg].literal_name if self.args[arg].literal_name is not None else arg
arg, value = self.args[arg].transform(literal_arg, value)
else:
arg, value = transformers.get(self.default_transformer)(arg, value)
- logger.debug(f"after: arg: {arg}, value: {value}")
+ _logger.debug(f"after: arg: {arg}, value: {value}")
if isinstance(arg, str):
prefix = self.long_prefix if len(arg) > 1 else self.short_prefix
if value is not None and not isinstance(value, bool):
@@ -241,23 +206,50 @@ def build_args(self, *args, **kwargs):
else:
positional.append(value)
result = positional + params
- logger.debug(result)
+ _logger.debug(result)
return result
@define
class CLIWrapper: # pylint: disable=too-many-instance-attributes
+ """
+ :param path: The path to the CLI tool. This will be passed to subprocess directly, and does not require a full path
+ unless the tool is not in the system path.
+ :param env: A dict of environment variables to be set in the subprocess environment, in addition to and overriding
+ those in os.environ.
+ :param trusting: If True, the wrapper will accept any command and pass them to the cli with default configuration.
+ Otherwise, it will only allow commands that have been defined with `update_command_`
+ :param raise_exc: If True, the wrapper will raise an exception if a command returns a non-zero exit code.
+ :param async_: If true, the wrapper will return coroutines that must be awaited.
+ :param default_transformer: The transformer configuration to apply to all arguments. The default of snake2kebab will
+ convert pythonic_snake_case_kwargs to kebab-case-arguments
+ :param short_prefix: The string prefix for single-letter arguments
+ :param long_prefix: The string prefix for arguments longer than 1 letter
+ :param arg_separator: The character that separates argument values from names. Defaults to '=', so
+ wrapper.command(arg=value) would become "wrapper command --arg=value"
+ """
+
path: str
+ """ @private """
env: dict[str, str] = None
- commands: dict[str, Command] = {}
+ """ @private """
+ _commands: dict[str, Command] = {}
+ """ @private """
trusting: bool = True
+ """ @private """
raise_exc: bool = False
+ """ @private """
async_: bool = False
+ """ @private """
default_transformer: str = "snake2kebab"
+ """ @private """
short_prefix: str = "-"
+ """ @private """
long_prefix: str = "--"
+ """ @private """
arg_separator: str = "="
+ """ @private """
def _get_command(self, command: str):
"""
@@ -265,7 +257,7 @@ def _get_command(self, command: str):
:param command: the command to be run
:return:
"""
- if command not in self.commands:
+ if command not in self._commands:
if not self.trusting:
raise ValueError(f"Command {command} not found in {self.path}")
c = Command(
@@ -276,7 +268,7 @@ def _get_command(self, command: str):
arg_separator=self.arg_separator,
)
return c
- return self.commands[command]
+ return self._commands[command]
def update_command_( # pylint: disable=too-many-arguments
self,
@@ -291,11 +283,12 @@ def update_command_( # pylint: disable=too-many-arguments
update the command to be run with the cli_wrapper
:param command: the command name for the wrapper
:param cli_command: the command to be run, if different from the command name
+ :param args: the arguments passed to the command
:param default_flags: default flags to be used with the command
:param parse: function to parse the output of the command
:return:
"""
- self.commands[command] = Command(
+ self._commands[command] = Command(
cli_command=command if cli_command is None else cli_command,
args=args if args is not None else {},
default_flags=default_flags if default_flags is not None else {},
@@ -307,18 +300,11 @@ def update_command_( # pylint: disable=too-many-arguments
)
def _run(self, command: str, *args, **kwargs):
- """
- run the command with the cli_wrapper
- :param command: the subcommand for the cli tool
- :param args: arguments to be passed to the command
- :param kwargs: flags to be passed to the command
- :return:
- """
command_obj = self._get_command(command)
command_obj.validate_args(*args, **kwargs)
command_args = [self.path] + command_obj.build_args(*args, **kwargs)
env = os.environ.copy().update(self.env if self.env is not None else {})
- logger.debug(f"Running command: {' '.join(command_args)}")
+ _logger.debug(f"Running command: {' '.join(command_args)}")
# run the command
result = subprocess.run(command_args, capture_output=True, text=True, env=env, check=self.raise_exc)
if result.returncode != 0:
@@ -330,7 +316,7 @@ async def _run_async(self, command: str, *args, **kwargs):
command_obj.validate_args(*args, **kwargs)
command_args = [self.path] + list(command_obj.build_args(*args, **kwargs))
env = os.environ.copy().update(self.env if self.env is not None else {})
- logger.debug(f"Running command: {', '.join(command_args)}")
+ _logger.debug(f"Running command: {', '.join(command_args)}")
proc = await asyncio.subprocess.create_subprocess_exec( # pylint: disable=no-member
*command_args,
stdout=asyncio.subprocess.PIPE,
@@ -354,6 +340,14 @@ def __getattr__(self, item, *args, **kwargs):
return lambda *args, **kwargs: self._run(item, *args, **kwargs)
def __call__(self, *args, **kwargs):
+ """
+ Invokes the wrapper with no extra arguments. e.g., for the kubectl wrapper, calls bare kubectl.
+ `kubectl(help=True)` will be interpreted as "kubectl --help".
+ :param args: positional arguments to be passed to the command
+ :param kwargs: kwargs will be treated as `--options`. Boolean values will be bare flags, others will be
+ passed as `--kwarg=value` (where `=` is the wrapper's arg_separator)
+ :return:
+ """
return (self.__getattr__(None))(*args, **kwargs)
@classmethod
@@ -388,12 +382,12 @@ def from_dict(cls, cliwrapper_dict):
def to_dict(self):
"""
Convert the CLIWrapper to a dictionary
- :return:
+ :return: a dictionary that can be used to recreate the wrapper using `from_dict`
"""
return {
"path": self.path,
"env": self.env,
- "commands": {k: v.to_dict() for k, v in self.commands.items()},
+ "commands": {k: v.to_dict() for k, v in self._commands.items()},
"trusting": self.trusting,
"async_": self.async_,
"default_transformer": self.default_transformer,
diff --git a/src/cli_wrapper/parsers.py b/src/cli_wrapper/parsers.py
index 90c55cc..5400cc3 100644
--- a/src/cli_wrapper/parsers.py
+++ b/src/cli_wrapper/parsers.py
@@ -3,7 +3,7 @@
from .util.callable_chain import CallableChain
from .util.callable_registry import CallableRegistry
-logger = logging.getLogger(__name__)
+_logger = logging.getLogger(__name__)
def extract(src: dict, *args) -> dict:
@@ -37,7 +37,10 @@ def extract(src: dict, *args) -> dict:
def yaml_loads(src: str) -> dict: # pragma: no cover
# pylint: disable=missing-function-docstring
yaml = YAML(typ="safe")
- return yaml.load(src)
+ result = list(yaml.load_all(src))
+ if len(result) == 1:
+ return result[0]
+ return result
core_parsers["yaml"] = yaml_loads
except ImportError: # pragma: no cover
@@ -68,11 +71,24 @@ def dotted_dictify(src, *args, **kwargs):
pass
parsers = CallableRegistry({"core": core_parsers}, callable_name="Parser")
+"""
+A `CallableRegistry` of parsers. These can be chained in sequence to perform
+operations on input.
+
+Defaults:
+core parsers:
+ - json - parses the input as json, returns the result
+ - extract - extracts the specified sub-dictionary from the source dictionary
+ - yaml - parses the input as yaml, returns the result (requires ruamel.yaml or pyyaml)
+ - dotted_dict - converts an input dictionary to a dotted_dict (requires dotted_dict)
+"""
class Parser(CallableChain):
"""
- Parser class that allows for the chaining of multiple parsers.
+ @public
+ Parser class that allows for the chaining of multiple parsers. Callables in the configuration are run as a
+ pipeline, with the output of one parser being passed as input to the next.
"""
def __init__(self, config):
@@ -82,6 +98,6 @@ def __call__(self, src):
# For now, parser expects to be called with one input.
result = src
for parser in self.chain:
- logger.debug(result)
+ _logger.debug(result)
result = parser(result)
return result
diff --git a/src/cli_wrapper/pre_packaged/__init__.py b/src/cli_wrapper/pre_packaged/__init__.py
index c52df80..5adeedd 100644
--- a/src/cli_wrapper/pre_packaged/__init__.py
+++ b/src/cli_wrapper/pre_packaged/__init__.py
@@ -1,10 +1,16 @@
from json import loads
from pathlib import Path
-from cli_wrapper import CLIWrapper
+from ..cli_wrapper import CLIWrapper
def get_wrapper(name, status=None):
+ """
+ Gets a wrapper defined in the beta/stable folders as json.
+ :param name: the name of the wrapper to retrieve
+ :param status: stable/beta/None. None will search stable and beta
+ :return: the requested wrapper
+ """
if status is None:
status = ["stable", "beta"]
if isinstance(status, str):
diff --git a/src/cli_wrapper/transformers.py b/src/cli_wrapper/transformers.py
index a8c6752..9528cfd 100644
--- a/src/cli_wrapper/transformers.py
+++ b/src/cli_wrapper/transformers.py
@@ -3,7 +3,9 @@
def snake2kebab(arg: str, value: any) -> tuple[str, any]:
"""
- snake.gravity == 0
+ `snake.gravity = 0`
+
+ converts a snake_case argument to a kebab-case one
"""
if isinstance(arg, str):
return arg.replace("_", "-"), value
@@ -14,5 +16,13 @@ def snake2kebab(arg: str, value: any) -> tuple[str, any]:
core_transformers = {
"snake2kebab": snake2kebab,
}
+""" @private """
transformers = CallableRegistry({"core": core_transformers})
+"""
+A callable registry of transformers.
+
+Defaults:
+core group:
+ - snake2kebab
+"""
diff --git a/src/cli_wrapper/util/__init__.py b/src/cli_wrapper/util/__init__.py
index e69de29..210a6aa 100644
--- a/src/cli_wrapper/util/__init__.py
+++ b/src/cli_wrapper/util/__init__.py
@@ -0,0 +1,4 @@
+from cli_wrapper.util.callable_chain import CallableChain
+from cli_wrapper.util.callable_registry import CallableRegistry
+
+__all__ = [CallableRegistry.__name__, CallableChain.__name__]
diff --git a/src/cli_wrapper/util/callable_chain.py b/src/cli_wrapper/util/callable_chain.py
index 57df3a2..e0b6cc8 100644
--- a/src/cli_wrapper/util/callable_chain.py
+++ b/src/cli_wrapper/util/callable_chain.py
@@ -2,10 +2,19 @@
class CallableChain(ABC):
+ """
+ A callable object representing a collection of callables.
+ """
+
chain: list[callable]
config: list
def __init__(self, config, source):
+ """
+ @public
+ :param config: a callable, a string, a dictionary with one key and config, or a list of the previous
+ :param source: a `CallableRegistry` to get callables from
+ """
self.chain = []
self.config = config
if callable(config):
@@ -30,7 +39,8 @@ def to_dict(self):
@abstractmethod
def __call__(self, value):
"""
- Calls the chain of functions with the given value.
+ This function should be overridden by subclasses to determine how the
+ callable chain is handled.
"""
raise NotImplementedError()
diff --git a/src/cli_wrapper/util/callable_registry.py b/src/cli_wrapper/util/callable_registry.py
index 96612ac..7f1c871 100644
--- a/src/cli_wrapper/util/callable_registry.py
+++ b/src/cli_wrapper/util/callable_registry.py
@@ -5,16 +5,25 @@
@define
class CallableRegistry:
+ """
+ Stores collections of callables. @public
+ - callables are registered by name
+ - they are retrieved by name with args and kwargs
+ - calling the callable with positional arguments will call the callable
+ with the args in the call, plus any args and kwargs passed to get()
+ """
+
_all: dict[str, dict[str, Callable]]
callable_name: str = "Callable thing"
+ """ a name of the things in the registry to use in error messages """
def get(self, name: str | Callable, args=None, kwargs=None) -> Callable:
"""
- Retrieves a parser function based on the specified parser name.
+ Retrieves a callable function based on the specified parser name.
- :param name: The name of the parser to retrieve.
- :return: The corresponding parser function.
- :raises KeyError: If the specified parser name is not found.
+ :param name: The name of the callable to retrieve.
+ :return: The corresponding callable function.
+ :raises KeyError: If the specified callable name is not found.
"""
if args is None:
args = []
@@ -27,10 +36,10 @@ def get(self, name: str | Callable, args=None, kwargs=None) -> Callable:
if group is not None:
if group not in self._all:
raise KeyError(f"{self.callable_name} group '{group}' not found.")
- parser_group = self._all[group]
- if name not in parser_group:
+ callable_group = self._all[group]
+ if name not in callable_group:
raise KeyError(f"{self.callable_name} '{name}' not found.")
- callable_ = parser_group[name]
+ callable_ = callable_group[name]
else:
for _, v in self._all.items():
if name in v:
@@ -42,9 +51,9 @@ def get(self, name: str | Callable, args=None, kwargs=None) -> Callable:
def register(self, name: str, callable_: callable, group="core"):
"""
- Registers a new parser function with the specified name.
+ Registers a new callable function with the specified name.
- :param name: The name to associate with the parser.
+ :param name: The name to associate with the callable.
:param callable_: The callable function to register.
"""
ngroup, name = self._parse_name(name)
@@ -57,31 +66,31 @@ def register(self, name: str, callable_: callable, group="core"):
raise KeyError(f"{self.callable_name} '{name}' already registered.")
self._all[group][name] = callable_
- def register_group(self, name: str, parsers: dict = None):
+ def register_group(self, name: str, callables: dict = None):
"""
- Registers a new parser group with the specified name.
+ Registers a new callable group with the specified name.
- :param name: The name to associate with the parser group.
- :param parsers: A dictionary of parsers to register in the group.
+ :param name: The name to associate with the callable group.
+ :param callables: A dictionary of callables to register in the group.
"""
if name in self._all:
raise KeyError(f"{self.callable_name} group '{name}' already registered.")
if "." in name:
raise KeyError(f"{self.callable_name} group name '{name}' is not valid.")
- parsers = {} if parsers is None else parsers
- bad_parser_names = [x for x in parsers.keys() if "." in x]
- if bad_parser_names:
+ callables = {} if callables is None else callables
+ bad_callable_names = [x for x in callables.keys() if "." in x]
+ if bad_callable_names:
raise KeyError(
- f"{self.callable_name} group '{name}' contains invalid parser names: {', '.join(bad_parser_names)}"
+ f"{self.callable_name} group '{name}' contains invalid callable names: {', '.join(bad_callable_names)}"
)
- self._all[name] = parsers
+ self._all[name] = callables
def _parse_name(self, name: str) -> tuple[str, str]:
"""
- Parses a name into a group and parser name.
+ Parses a name into a group and callable name.
:param name: The name to parse.
- :return: A tuple containing the group and parser name.
+ :return: A tuple containing the group and callable name.
"""
if "." not in name:
return None, name
diff --git a/src/cli_wrapper/validators.py b/src/cli_wrapper/validators.py
index 61e6ae3..6990223 100644
--- a/src/cli_wrapper/validators.py
+++ b/src/cli_wrapper/validators.py
@@ -5,7 +5,7 @@
from .util.callable_chain import CallableChain
from .util.callable_registry import CallableRegistry
-logger = logging.getLogger(__name__)
+_logger = logging.getLogger(__name__)
core_validators = {
"is_dict": lambda x: isinstance(x, dict),
@@ -15,12 +15,12 @@
"is_int": lambda x: isinstance(x, int),
"is_bool": lambda x: isinstance(x, bool),
"is_float": lambda x: isinstance(x, float),
- "is_alnum": lambda x: isinstance(x, str) and x.isalnum(),
- "is_alpha": lambda x: isinstance(x, str) and x.isalpha(),
- "is_digit": lambda x: isinstance(x, str) and x.isdigit(),
+ "is_alnum": lambda x: x.isalnum(),
+ "is_alpha": lambda x: x.isalpha(),
+ "is_digit": lambda x: x.isdigit(),
"is_path": lambda x: isinstance(x, Path),
- "starts_alpha": lambda x: isinstance(x, str) and x[0].isalpha(),
- "startswith": lambda x, prefix: isinstance(x, str) and x.startswith(prefix),
+ "starts_alpha": lambda x: len(x) and x[0].isalpha(),
+ "startswith": lambda x, prefix: x.startswith(prefix),
}
validators = CallableRegistry({"core": core_validators}, callable_name="Validator")
@@ -28,8 +28,11 @@
class Validator(CallableChain):
"""
+ @public
+
A class that provides a validation mechanism for input data.
It uses a list of validators to check if the input data is valid.
+ They are executed in sequence until one fails.
"""
def __init__(self, config):
@@ -45,13 +48,17 @@ def __call__(self, value):
config = [self.config] if not isinstance(self.config, list) else self.config
for x, c in zip(self.chain, config):
validator_result = x(value)
- logger.debug(f"Validator {c} result: {validator_result}")
+ _logger.debug(f"Validator {c} result: {validator_result}")
result = result and validator_result
+ if isinstance(result, str) or not result:
+ # don't bother doing other validations once one has failed
+ _logger.debug("...failed")
+ break
return result
def to_dict(self):
"""
Converts the validator configuration to a dictionary.
"""
- logger.debug(f"returning validator config: {self.config}")
+ _logger.debug(f"returning validator config: {self.config}")
return self.config
diff --git a/src/help_parser/__main__.py b/src/help_parser/__main__.py
index d3be84f..8bce0e7 100644
--- a/src/help_parser/__main__.py
+++ b/src/help_parser/__main__.py
@@ -34,13 +34,14 @@ def parse_args(argv):
default="help",
help="The flag to use for getting help (default: 'help').",
)
- parser.add_argument(
- "--style",
- type=str,
- choices=["golang", "argparse"],
- default="golang",
- help="The style of cli help output (default: 'golang').",
- )
+ # Re-add this when some other help format is implemented
+ # parser.add_argument(
+ # "--style",
+ # type=str,
+ # choices=["golang", "argparse"],
+ # default="golang",
+ # help="The style of cli help output (default: 'golang').",
+ # )
parser.add_argument(
"--default-flags",
type=str,
@@ -78,6 +79,7 @@ def parse_args(argv):
)
config = parser.parse_args(argv)
+ config.style = "golang"
config.default_flags_dict = {}
for f in config.default_flags:
if "=" not in f:
diff --git a/tests/pre_packaged/test_kubectl.py b/tests/pre_packaged/test_kubectl.py
index 1f31f9e..41fd578 100644
--- a/tests/pre_packaged/test_kubectl.py
+++ b/tests/pre_packaged/test_kubectl.py
@@ -1,3 +1,4 @@
+from pathlib import Path
from shutil import which
import pytest
@@ -11,3 +12,12 @@ def test_kubectl_wrapper():
result = kubectl.config_get_contexts()
assert result is not None
+
+
+@pytest.mark.skipif(which("kubectl") is not None, reason="skipping fake kubectl test because real kubectl exists")
+def test_kubectl_wrapper_fake():
+ kubectl = get_wrapper("kubectl")
+ kubectl.path = (Path(__file__).parent.parent / "data" / "fake_kubectl").as_posix()
+
+ result = kubectl.get_pods()
+ assert result is not None
diff --git a/tests/test_cli_wrapper.py b/tests/test_cli_wrapper.py
index af127ae..a3cebd4 100644
--- a/tests/test_cli_wrapper.py
+++ b/tests/test_cli_wrapper.py
@@ -21,6 +21,17 @@ def test_argument(self):
with pytest.raises(KeyError):
Argument("test", validator="not callable")
+ def is_invalid(value):
+ return value == "invalid"
+
+ validators.register("is_invalid", is_invalid)
+
+ arg = Argument("test", default="default", validator=is_invalid)
+ assert arg.is_valid("valid") is False
+ assert arg.is_valid("invalid") is True
+
+ validators._all["core"].pop("is_invalid")
+
def test_argument_from_dict(self):
arg = Argument.from_dict({"literal_name": "test", "default": "default", "validator": lambda x: x == "valid"})
@@ -150,8 +161,8 @@ def test_cliwrapper(self):
r = kubectl.get("pods", namespace="kube-system")
assert isinstance(r, str)
- kubectl.commands["get"].default_flags = {"output": "json"}
- kubectl.commands["get"].parse = ["json"]
+ kubectl._commands["get"].default_flags = {"output": "json"}
+ kubectl._commands["get"].parse = ["json"]
r = kubectl.get("pods", "-A")
assert r["kind"] == "List"
@@ -170,7 +181,7 @@ def test_cliwrapper(self):
with pytest.raises(ValueError):
kubectl.describe("pods", namespace="kube-system")
logger.info("no parser")
- kubectl.commands["get"].parse = None
+ kubectl._commands["get"].parse = None
r = kubectl.get("pods", namespace="kube-system")
assert isinstance(r, str)
@@ -185,7 +196,7 @@ async def test_subprocessor_async(self):
with pytest.raises(RuntimeError):
await kubectl.fake("pods", namespace="kube-system")
- kubectl.commands["get"].parse = None
+ kubectl._commands["get"].parse = None
r = await kubectl.get("pods", namespace="kube-system")
assert isinstance(r, str)
@@ -216,11 +227,11 @@ def validate_resource_name(name):
assert cliwrapper.path == "kubectl"
assert cliwrapper.trusting is True
- assert cliwrapper.commands["get"].cli_command == ["get"]
- assert cliwrapper.commands["get"].default_flags == {"output": "json"}
- assert cliwrapper.commands["get"].parse('"some json"') == "some json"
+ assert cliwrapper._commands["get"].cli_command == ["get"]
+ assert cliwrapper._commands["get"].default_flags == {"output": "json"}
+ assert cliwrapper._commands["get"].parse('"some json"') == "some json"
with pytest.raises(ValueError):
- cliwrapper.commands["get"].validate_args("pods", "my_cool_pod!!")
+ cliwrapper._commands["get"].validate_args("pods", "my_cool_pod!!")
with pytest.raises(ValueError):
cliwrapper.get("pods", "my_cool_pod!!")
diff --git a/tests/test_serialization.py b/tests/test_serialization.py
index 3f92c04..8cd6831 100644
--- a/tests/test_serialization.py
+++ b/tests/test_serialization.py
@@ -1,5 +1,4 @@
-from cli_wrapper import CLIWrapper
-from cli_wrapper.cli_wrapper import Argument
+from cli_wrapper.cli_wrapper import Argument, CLIWrapper
class TestSerialization: