PICLE API reference

This page describes the configuration hooks PICLE reads from your Pydantic models and fields. It focuses on the parts you use when defining a shell (config, field metadata, execution, and output). For full API docs of App and built-in models, keep reading to the mkdocstrings reference at the bottom.

PicleConfig Model Level Configuration

Any Pydantic model may define an inner PicleConfig class. PICLE reads attributes from it (when present). Only a few are required; most are optional quality-of-life switches.

PicleConfig is intentionally “open-ended”: the core App reads a known set of attributes (documented below), and specific built-in models may honor additional PicleConfig keys (for example, ConfigModel).

Name	Meaning
ruler	Separator line char used by cmd help formatting (empty disables)
intro	Banner printed on shell start
prompt	Prompt string
use_rich	If True and Rich is installed, print via Rich console
newline	Output newline, default \r\n
completekey	Readline completion key name, default tab
pipe	Enables | and selects the pipe model ("self", import string, or model class)
processors	List of callables applied to the first command result
outputter	Callable used to render output when not overridden
outputter_kwargs	Extra kwargs passed into outputter
history_length	Length of commands history to store for history output, default 100
history_file	Filename to persistently store commands history, default ./picle_history.txt
subshell	If True, navigating to this model with no args enters a subshell (prompt changes, model is pushed onto a stack)
methods_override	Dict of {app_method_name: model_method_name} used to override App methods at runtime

json_schema_extra Field Level Configuration

PICLE reads extra behavior from fields definitions - Field(..., json_schema_extra={...}).

Note: command tokens come from the field name (or its alias / serialization_alias), not from the Pydantic class name.

Key	Meaning
function	Name of a model @staticmethod to call when run() is absent
presence	Constant value used when field is referenced without a value
processors	List of callables applied to the command result
outputter	Callable that formats output for this field (overrides model outputter)
outputter_kwargs	Extra kwargs passed into outputter
multiline	If True, the literal value input triggers multi-line collection
root_model	If True, pass the app root model as root_model=...
picle_app	If True, pass the App instance as picle_app=...
use_parent_run	If True (default), and the leaf model has no run(), PICLE searches parent models for a run() to execute. If False, the command errors unless the leaf model defines run() or function.
pkey	Primary key name to use for dynamic dictionary models
pkey_description	Description of dynamic dictionary model primary key

Handling of function Argument vs run() Method

Execution is resolved like this:

1. If the last referenced field sets json_schema_extra={"function": "method_name"}, PICLE calls getattr(model, method_name)(**kwargs).
2. If the current model has run, PICLE calls model.run(**kwargs).
3. if json_schema_extra={"use_parent_run": True} set on the field, backtracks parent models and executes first found run() method.

This lets small models define many “command -> staticmethod” fields, while larger models can centralize behavior in run().

Callable Signatures

PICLE builds callable **kwargs from collected field values and calls either run() or the field-level function. It can also inject extra context if callable declares a matching argument name:

• root_model - if callable signature includes root_model adds self.root model to callable arguments e.g. root_model=self.root
• picle_app - if callable signature includes picle_app adds self to callable arguments e.g. picle_app=self
• shell_command - if callable signature includes shell_command adds parsed command context for the current segment: a list of model dictionaries produced by parse_command method. This is useful when your function needs to inspect the command path, model defaults, or other parsing details.

PICLE App

picle.App(root: object, stdin=None, stdout=None)

Bases: Cmd

PICLE App class to construct shell.

Parameters:

• root (object) –

  Root/Top Pydantic model.

Source code in picle\picle.py

def __init__(self, root: object, stdin=None, stdout=None):
    self.root = root
    self.shell = self.root.model_construct()
    self.shell_defaults = {}
    self.shells = [self.shell]

    # extract configuration from shell model
    if hasattr(self.shell, "PicleConfig"):
        self.ruler = getattr(self.shell.PicleConfig, "ruler", self.ruler)
        self.intro = getattr(self.shell.PicleConfig, "intro", self.intro)
        self.prompt = getattr(self.shell.PicleConfig, "prompt", self.prompt)
        self.newline = getattr(self.shell.PicleConfig, "newline", self.newline)
        self.use_rich = getattr(self.shell.PicleConfig, "use_rich", self.use_rich)
        self.completekey = getattr(
            self.shell.PicleConfig, "completekey", self.completekey
        )
        self.history_length = getattr(
            self.shell.PicleConfig, "history_length", self.history_length
        )
        self.history_file = getattr(
            self.shell.PicleConfig, "history_file", self.history_file
        )

        # mount override methods
        if hasattr(self.shell.PicleConfig, "methods_override"):
            for (
                method_name,
                override,
            ) in self.shell.PicleConfig.methods_override.items():
                setattr(self, method_name, getattr(self.shell, override))

    # mount models
    self.model_mount(MAN, ["man"], "Manual/documentation functions")

    super(App, self).__init__(stdin=stdin, stdout=stdout)

    # configure readline history
    if HAS_READLINE:
        _readline.set_history_length(self.history_length)
        if self.history_file:
            history_path = os.path.expanduser(self.history_file)
            if os.path.exists(history_path):
                try:
                    _readline.read_history_file(history_path)
                except OSError:
                    pass

picle.App.build_command_data(models: list) -> dict

Build flat dictionary of command data from parsed models list.

Parameters:

• models (list) –

  List of parsed model dictionaries.

Returns:

• dict ( dict ) –

  Flat dictionary of command data.

Source code in picle\picle.py

def build_command_data(self, models: list) -> dict:
    """
    Build flat dictionary of command data from parsed models list.

    Args:
        models (list): List of parsed model dictionaries.

    Returns:
        dict: Flat dictionary of command data.
    """
    command_data = {}
    for index, model in enumerate(models):
        # collect data for dynamic dictionary keys
        if isinstance(model["model"], VirtualDictModel):
            vdm = model["model"]
            # collect next model parameter as a value for the dictionary key
            command_data[vdm.key] = models[index + 1]["parameter"]
        else:
            for f in model["fields"]:
                if f["values"] is not ...:
                    command_data[f["name"]] = f["values"]
    return command_data

picle.App.completedefault(text: str, line: str, begidx: int, endidx: int) -> list[str]

Return completions for every command parameter after the first one.

Called by cmd on a tab-key hit for arguments beyond the initial command keyword.

Parameters:

• text (str) –

  The current text being completed.

• line (str) –

  The current input line.

• begidx (int) –

  The beginning index of the completion.

• endidx (int) –

  The ending index of the completion.

Returns:

• list[str] –

  list[str]: List of completion suggestions.

Source code in picle\picle.py

def completedefault(
    self, text: str, line: str, begidx: int, endidx: int
) -> list[str]:
    """
    Return completions for every command parameter after the first one.

    Called by cmd on a tab-key hit for arguments beyond the initial command keyword.

    Args:
        text (str): The current text being completed.
        line (str): The current input line.
        begidx (int): The beginning index of the completion.
        endidx (int): The ending index of the completion.

    Returns:
        list[str]: List of completion suggestions.
    """
    fieldnames = []
    try:
        command_models = self.parse_command(line, is_help=True)
        last_model = command_models[-1][-1]["model"]
        # check if last model has fields collected
        if command_models[-1][-1]["fields"]:
            last_field_name = command_models[-1][-1]["fields"][-1]["name"]
            last_field = model_fields(last_model)[last_field_name]
            last_field_value = command_models[-1][-1]["fields"][-1]["values"]
            fparam = self._get_field_params(last_field)
            if isinstance(last_field_value, list):
                last_field_value = last_field_value[-1]
            elif last_field_value == ...:
                last_field_value = ""
            # check if need to extract enum values
            if isinstance(last_field.annotation, enum.EnumMeta):
                fieldnames = [
                    str(i.value)
                    for i in last_field.annotation
                    if str(i.value).startswith(last_field_value)
                    and i.value != last_field_value
                ]
            # check if model has method to source field choices
            elif hasattr(last_model, f"source_{last_field_name}"):
                source_callable = getattr(last_model, f"source_{last_field_name}")
                if callable_expects_argument(source_callable, "choice"):
                    fieldnames = source_callable(choice=text)
                else:
                    fieldnames = source_callable()
                # handle partial match
                if last_field_value not in fieldnames:
                    fieldnames = [
                        str(i)
                        for i in fieldnames
                        if str(i).startswith(last_field_value)
                    ]
                # remove already collected values from choice
                collected_values = command_models[-1][-1]["fields"][-1]["values"]
                if collected_values is not ...:
                    if isinstance(collected_values, list):
                        fieldnames = [
                            i for i in fieldnames if i not in collected_values
                        ]
                    else:
                        fieldnames = [
                            i for i in fieldnames if i != collected_values
                        ]
            # auto complete 'load-terminal' for multi-line input mode
            elif fparam.get("multiline") is True:
                if (
                    "load-terminal".startswith(last_field_value)
                    and last_field_value != "load-terminal"
                ):
                    fieldnames = ["load-terminal"]
        # return a list of all model fields
        else:
            if line.endswith(" "):
                for name, f in model_fields(last_model).items():
                    if f.alias:
                        fieldnames.append(f.alias)
                    elif f.serialization_alias:
                        fieldnames.append(f.serialization_alias)
                    else:
                        fieldnames.append(name)
            else:
                last_fieldname = command_models[-1][-1]["parameter"]
                fieldnames.append(last_fieldname)
    except FieldLooseMatchOnly as e:
        model, parameter = e.args
        for name, f in model_fields(model["model"]).items():
            # skip fields with already collected values from complete prompt
            if any(
                collected_field["name"] == name
                for collected_field in model["fields"]
                if collected_field["values"] is not ...
            ):
                continue
            # handle Enum fields options
            elif any(
                collected_field["name"] == name
                for collected_field in model["fields"]
            ) and isinstance(f.annotation, enum.EnumMeta):
                fieldnames = [
                    str(i.value)
                    for i in f.annotation
                    if str(i.value).startswith(parameter)
                ]
                break
            elif f.alias and f.alias.startswith(parameter):
                fieldnames.append(f.alias)
            elif f.serialization_alias and f.serialization_alias.startswith(
                parameter
            ):
                fieldnames.append(f.serialization_alias)
            elif name.startswith(parameter):
                fieldnames.append(name)
    except FieldKeyError:
        pass
    except:
        tb = traceback.format_exc()
        self.write(tb)

    return sorted([f"{i} " for i in fieldnames])

picle.App.completenames(text: str, line: str, begidx: int, endidx: int) -> list[str]

Return completions for the very first command parameter.

Called by cmd on a tab-key hit for the initial keyword.

Parameters:

• text (str) –

  The current text being completed.

• line (str) –

  The current input line.

• begidx (int) –

  The beginning index of the completion.

• endidx (int) –

  The ending index of the completion.

Returns:

• list[str] –

  list[str]: List of completion suggestions.

Source code in picle\picle.py

def completenames(
    self, text: str, line: str, begidx: int, endidx: int
) -> list[str]:
    """
    Return completions for the very first command parameter.

    Called by cmd on a tab-key hit for the initial keyword.

    Args:
        text (str): The current text being completed.
        line (str): The current input line.
        begidx (int): The beginning index of the completion.
        endidx (int): The ending index of the completion.

    Returns:
        list[str]: List of completion suggestions.
    """
    fieldnames = []
    # collect global methods
    for method_name in dir(self):
        if method_name.startswith("do_"):
            name = method_name.replace("do_", "")
            if name.startswith(line):
                fieldnames.append(name)
    # collect model arguments
    try:
        command_models = self.parse_command(line, is_help=True)
        fieldnames.extend(model_fields(command_models[-1][-1]["model"]))
    except FieldLooseMatchOnly as e:
        model, parameter = e.args
        for name, f in model_fields(model["model"]).items():
            display = f.alias or f.serialization_alias or name
            if display.startswith(parameter):
                fieldnames.append(display)
    except FieldKeyError:
        pass
    return sorted([f"{i} " for i in fieldnames])

picle.App.default(line: str) -> Optional[bool]

Process a command line when no matching do_* method is found.

Parameters:

• line (str) –

  Command line input.

Returns:

• Optional[bool] –

  Optional[bool]: True if the shell should exit, otherwise None.

Source code in picle\picle.py

@run_print_exception
def default(self, line: str) -> Optional[bool]:
    """
    Process a command line when no matching do_* method is found.

    Args:
        line (str): Command line input.

    Returns:
        Optional[bool]: True if the shell should exit, otherwise None.
    """
    ret = False
    outputter = True  # use default outputter - self.write
    outputter_kwargs = {}
    line = line.strip()

    if line.endswith("?"):
        self.process_help_command(line)
    else:
        try:
            command_models = self.parse_command(line, collect_multiline=True)
        except FieldLooseMatchOnly as e:
            model, parameter = e.args
            fields = [
                f.alias or f.serialization_alias or name
                for name, f in model_fields(model["model"]).items()
                if name.startswith(parameter)
                or (f.alias and f.alias.startswith(parameter))
                or (
                    f.serialization_alias
                    and f.serialization_alias.startswith(parameter)
                )
            ]
            self.write_error(
                f"Incomplete command, possible completions: {', '.join(fields)}"
            )
        except FieldKeyError as e:
            model, parameter = e.args
            candidates = [
                f.alias or f.serialization_alias or name
                for name, f in model_fields(model["model"]).items()
            ]
            close = difflib.get_close_matches(
                parameter, candidates, n=3, cutoff=0.6
            )
            msg = f"Incorrect command, '{parameter}' not part of '{self._get_model_name(model)}' model fields"
            if close:
                msg += f". Did you mean: {', '.join(close)}?"
            self.write_error(msg)
        except ValidationError as e:
            self.write_error(e)
        else:
            # go over collected commands separated by pipe
            for index, command in enumerate(command_models):
                json_schema_extra = {}
                method_name = None
                # collect arguments
                command_arguments = self.build_command_data(command)

                # collect command defaults
                command_defaults = {}
                for cmd in command:
                    command_defaults.update(cmd.get("defaults", {}))
                model = command[-1]["model"]
                picle_config = getattr(model, "PicleConfig", None)

                # check if model has subshell and no arguments provided - enter subshell
                if (
                    not command_arguments
                    and getattr(picle_config, "subshell", None) is True
                ):
                    for item in command[:-1]:
                        m = item["model"]
                        self.defaults_update(m)
                        if (
                            getattr(
                                getattr(m, "PicleConfig", None), "subshell", None
                            )
                            is True
                            and m not in self.shells
                        ):
                            self.shells.append(m)
                    self.prompt = getattr(picle_config, "prompt", self.prompt)
                    self.shell = model
                    self.shells.append(self.shell)
                    continue

                # resolve run function - prefer json_schema_extra "function", fallback to "run" method, search parents for "run"
                if command[-1]["fields"]:
                    json_schema_extra = command[-1]["fields"][-1][
                        "json_schema_extra"
                    ]
                if callable(json_schema_extra.get("function")):
                    run_function = json_schema_extra["function"]
                else:
                    method_name = json_schema_extra.get("function", "run")
                    if hasattr(model, method_name):
                        run_function = getattr(model, method_name)
                    elif method_name != "run":
                        ret = f"Model '{model.__name__}' has no '{method_name}' method defined"
                        break
                    elif json_schema_extra.get("use_parent_run", True):
                        run_function = self._find_parent_run(command)
                        if run_function is None:
                            self.defaults_pop(model)
                            ret = f"Incorrect command for '{model.__name__}', model parents have no 'run' method to execute command"
                            break
                    else:
                        self.defaults_pop(model)
                        ret = f"Incorrect command for '{model.__name__}', model has no method to execute command"
                        break

                # validate command data and exit if failed
                if not self._validate_values(command):
                    return

                # build kwargs and call the method
                kw = {}
                if callable_expects_argument(run_function, "shell_command"):
                    kw["shell_command"] = command
                if callable_expects_argument(run_function, "root_model"):
                    kw["root_model"] = self.root
                if callable_expects_argument(run_function, "picle_app"):
                    kw["picle_app"] = self
                if index == 0:
                    kw.update(self.shell_defaults)
                    kw.update(command_defaults)
                    kw.update(command_arguments)
                    ret = run_function(**kw)
                else:
                    kw.update(command_defaults)
                    kw.update(command_arguments)
                    ret = run_function(ret, **kw)

                # apply field-level processors
                for processor in json_schema_extra.get("processors", []):
                    if callable(processor):
                        ret = processor(ret)

                # apply PicleConfig processors for first command only
                if index == 0:
                    for processor in getattr(picle_config, "processors", []):
                        if callable(processor):
                            ret = processor(ret)

                # resolve outputter: from return tuple, field definition, or PicleConfig
                if isinstance(ret, tuple) and len(ret) == 2:
                    ret, outputter = ret
                    outputter_kwargs = {}
                elif isinstance(ret, tuple) and len(ret) == 3:
                    ret, outputter, outputter_kwargs = ret
                elif json_schema_extra.get("outputter"):
                    outputter = json_schema_extra["outputter"]
                    outputter_kwargs = json_schema_extra.get("outputter_kwargs", {})
                elif picle_config and hasattr(picle_config, "outputter"):
                    outputter = picle_config.outputter
                    outputter_kwargs = getattr(picle_config, "outputter_kwargs", {})

    # returning True will end the shell - exit
    if ret is True:
        return True

    if ret:
        if callable(outputter):
            self.write(outputter(ret, **outputter_kwargs))
        elif outputter is True:
            self.write(ret)

picle.App.defaults_pop(model: Any) -> None

Remove the given model's field names from shell_defaults.

Parameters:

• model (Any) –

  Pydantic model class or instance.

Source code in picle\picle.py

def defaults_pop(self, model: Any) -> None:
    """
    Remove the given model's field names from shell_defaults.

    Args:
        model: Pydantic model class or instance.
    """
    for name in model_fields(model).keys():
        self.shell_defaults.pop(name, None)

picle.App.defaults_set(model: Any) -> None

Replace shell_defaults with the given model's defaults.

Clears the existing defaults and populates them from model.

Parameters:

• model (Any) –

  Pydantic model class or instance.

Source code in picle\picle.py

def defaults_set(self, model: Any) -> None:
    """
    Replace shell_defaults with the given model's defaults.

    Clears the existing defaults and populates them from model.

    Args:
        model: Pydantic model class or instance.
    """
    self.shell_defaults.clear()
    self.defaults_update(model)

picle.App.defaults_update(model: Any) -> None

Merge the given model's default field values into shell_defaults.

Parameters:

• model (Any) –

  Pydantic model class or instance.

Source code in picle\picle.py

def defaults_update(self, model: Any) -> None:
    """
    Merge the given model's default field values into shell_defaults.

    Args:
        model: Pydantic model class or instance.
    """
    self.shell_defaults.update(self.extract_model_defaults(model))

picle.App.do_cls(arg: str) -> None

Clear the terminal screen.

Source code in picle\picle.py

def do_cls(self, arg: str) -> None:
    """Clear the terminal screen."""
    if "?" in arg:
        self.write(" cls    Clear shell Screen")
    else:
        if "LINUX" in platform.system().upper():
            os.system("clear")
        elif "WINDOWS" in platform.system().upper():
            os.system("cls")

picle.App.do_end(arg: str) -> Optional[bool]

Exit the application entirely.

Source code in picle\picle.py

def do_end(self, arg: str) -> Optional[bool]:
    """Exit the application entirely."""
    if "?" in arg:
        self.write(" end    Exit application")
    else:
        return True

picle.App.do_exit(arg: str) -> Optional[bool]

Exit current shell or terminate if at the top level.

Source code in picle\picle.py

def do_exit(self, arg: str) -> Optional[bool]:
    """Exit current shell or terminate if at the top level."""
    if "?" in arg:
        self.write(" exit    Exit current shell")
    else:
        # delete defaults for closing shell
        self.defaults_pop(self.shells[-1])
        _ = self.shells.pop(-1)
        if self.shells:
            self.shell = self.shells[-1]
            if hasattr(self.shell, "PicleConfig") and getattr(
                self.shell.PicleConfig, "prompt"
            ):
                self.prompt = self.shell.PicleConfig.prompt
            if len(self.shells) == 1:  # check if reached top shell
                self.defaults_set(self.shell)
        else:
            return True

picle.App.do_help(arg: str) -> None

Print help message for the given command or model.

Source code in picle\picle.py

def do_help(self, arg: str) -> None:
    """Print help message for the given command or model."""
    try:
        command_models = self.parse_command(arg.strip("?"), is_help=True)
    except FieldLooseMatchOnly as e:
        model, parameter = e.args
        self.print_model_help([[model]], verbose=..., match=parameter)
        return
    except FieldKeyError as e:
        model, parameter = e.args
        self.write_error(
            f"Incorrect command, '{parameter}' not part of "
            f"'{self._get_model_name(model)}' model fields"
        )
        return
    help_msg, width = self.print_model_help(
        command_models,
        verbose=arg.strip().endswith("?"),
        print_help=False,
    )
    # print help for global top commands
    if len(arg.strip().split(" ")) == 1:
        lines = {}  # dict of {cmd: cmd_help}
        for method_name in dir(self):
            if method_name.startswith("do_"):
                name = method_name.replace("do_", "")
                lines[name] = getattr(self, method_name).__doc__
                width = max(width, len(name))
        if lines:
            for k, v in lines.items():
                padding = " " * (width - len(k)) + (" " * 4)
                help_msg.append(f" {k}{padding}{v}")
    self.write(self.newline.join(help_msg))

picle.App.do_history(arg: str) -> None

Print command history.

Source code in picle\picle.py

def do_history(self, arg: str) -> None:
    """Print command history."""
    if "?" in arg:
        self.write(" history    Print command history")
    else:
        if HAS_READLINE:
            count = _readline.get_current_history_length()
            if count == 0:
                self.write("No command history")
            else:
                lines = []
                for i in range(count):
                    if _readline.get_history_item(i + 1):
                        line = f" {_readline.get_history_item(i + 1).strip()}"
                        if line.strip() in [
                            "history",
                            "exit",
                            "top",
                            "pwd",
                            "end",
                            "cls",
                            "help",
                        ]:
                            continue
                        if line.endswith("?"):
                            continue
                        if line in lines:
                            continue
                        lines.append(line)
                self.write(self.newline.join(lines))
        else:
            self.write("No command history")

picle.App.do_pwd(arg: str) -> None

Print the current shell path from root.

Source code in picle\picle.py

def do_pwd(self, arg: str) -> None:
    """Print the current shell path from root."""
    if "?" in arg:
        self.write(" pwd    Print current shell path")
    else:
        path = ["Root"]
        for shell in self.shells[1:]:
            path.append(shell.__name__)
        self.write("->".join(path))

picle.App.do_top(arg: str) -> None

Exit to top shell, resetting the shell stack.

Source code in picle\picle.py

def do_top(self, arg: str) -> None:
    """Exit to top shell, resetting the shell stack."""
    if "?" in arg:
        self.write(" top    Exit to top shell")
    else:
        self.shell = self.shells[0]
        if hasattr(self.shell, "PicleConfig") and getattr(
            self.shell.PicleConfig, "prompt"
        ):
            self.prompt = self.shell.PicleConfig.prompt
        while self.shells:
            _ = self.shells.pop()
        self.shells.append(self.shell)
        # set shell defaults
        self.defaults_set(self.shell)

picle.App.emptyline() -> None

Override empty line method to not run last command.

Source code in picle\picle.py

def emptyline(self) -> None:
    """
    Override empty line method to not run last command.
    """
    return None

picle.App.extract_model_defaults(model: Any) -> dict

Extract non-None default values from a Pydantic model's fields.

Parameters:

• model (Any) –

  Pydantic model class or instance to extract defaults from.

Returns:

• dict ( dict ) –

  Dictionary mapping field names to their default values.

Source code in picle\picle.py

def extract_model_defaults(self, model: Any) -> dict:
    """
    Extract non-None default values from a Pydantic model's fields.

    Args:
        model: Pydantic model class or instance to extract defaults from.

    Returns:
        dict: Dictionary mapping field names to their default values.
    """
    ret = {}
    # extract default values from model fields
    for name, field in model_fields(model).items():
        # skip non Field references e.g. to other models
        if not isinstance(field, FieldInfo):
            continue
        # skip required Fields
        if field.is_required():
            continue
        # ignore None default values
        if field.get_default() is None:
            continue
        default = field.get_default()
        # convert Enum defaults to their plain value
        if isinstance(default, enum.Enum):
            default = default.value
        ret[name] = default

    return ret

picle.App.model_mount(model: ModelMetaclass, path: Union[str, list[str]], description: str = None, default: Any = None, **kwargs: dict) -> None

Mount a Pydantic model at the provided path in relation to the root model.

Parameters:

• model (ModelMetaclass) –

  Pydantic model to mount.

• path (Union[str, list[str]]) –

  List of path segments to mount the model.

• description (str, default: None ) –

  Description of the model.

• default (Any, default: None ) –

  Default value for the model.

• **kwargs (dict, default: {} ) –

  Additional keyword arguments for the FieldInfo.

Source code in picle\picle.py

def model_mount(
    self,
    model: ModelMetaclass,
    path: Union[str, list[str]],
    description: str = None,
    default: Any = None,
    **kwargs: dict,
) -> None:
    """
    Mount a Pydantic model at the provided path in relation to the root model.

    Args:
        model: Pydantic model to mount.
        path: List of path segments to mount the model.
        description (str, optional): Description of the model.
        default: Default value for the model.
        **kwargs: Additional keyword arguments for the FieldInfo.
    """
    if isinstance(path, str):
        path = [path.strip()]
    parent_model = self.root
    while path:
        mount_name = path.pop(0)
        if mount_name in parent_model.model_fields:
            parent_model = parent_model.model_fields[mount_name].annotation
        else:
            # handle when not all path items before last one are in models tree
            if len(path) > 0:
                raise KeyError(
                    f"'{mount_name}' not part of '{parent_model}' model fields, but remaining path still not empty - {path}"
                )
            parent_model.model_fields[mount_name] = FieldInfo(
                annotation=model,
                required=False,
                description=description,
                default=default,
                **kwargs,
            )
            break

picle.App.model_remove(path: list[str]) -> None

Remove a Pydantic model at the provided path in relation to the root model.

Parameters:

• path (list[str]) –

  List of path segments to remove the model.

Source code in picle\picle.py

def model_remove(self, path: list[str]) -> None:
    """
    Remove a Pydantic model at the provided path in relation to the root model.

    Args:
        path (list[str]): List of path segments to remove the model.
    """
    if isinstance(path, str):
        path = [path.strip()]
    parent_model = self.root
    while path:
        mount_name = path.pop(0)
        if mount_name in parent_model.model_fields:
            if len(path) == 0:
                parent_model = parent_model.model_fields.pop(mount_name)
            else:
                parent_model = parent_model.model_fields[mount_name].annotation
        else:
            raise KeyError(
                f"Failed to remove model at path '{mount_name}', parent model: '{parent_model}'"
            )

picle.App.parse_command(command: str, collect_multiline: bool = False, is_help: bool = False) -> list

Parse command string and construct list of model references and field values.

Parameters:

• command (str) –

  Command string to parse through.

• collect_multiline (bool, default: False ) –

  Enables multiple input collection for fields.

• is_help (bool, default: False ) –

  Indicates that parsing help command or tab completion command; if set to True disables 'presence' argument handling for last field.

Returns:

• list ( list ) –

  List of lists of dictionaries with collected models details, each dictionary containing 'model', 'fields', and 'parameter' keys.

Source code in picle\picle.py

def parse_command(
    self, command: str, collect_multiline: bool = False, is_help: bool = False
) -> list:
    """
    Parse command string and construct list of model references and field values.

    Args:
        command (str): Command string to parse through.
        collect_multiline (bool): Enables multiple input collection for fields.
        is_help (bool): Indicates that parsing help command or tab completion command; if set to True disables 'presence' argument handling for last field.

    Returns:
        list: List of lists of dictionaries with collected models details, each dictionary containing 'model', 'fields', and 'parameter' keys.
    """
    current_model = {
        "model": self.shell,
        "fields": [],
        "parameter": ...,
        "defaults": self.extract_model_defaults(self.shell),
    }
    current_field = {}
    models = [current_model]
    parameters = [i for i in command.split(" ") if i.strip()]
    ret = [models]

    # iterate over command parameters and decide if its a reference
    # to a model or model's field value
    while parameters:
        parameter = parameters.pop(0)

        # handle pipe - "|"
        if parameter == "|":
            pipe_config = self._find_pipe_config(models)
            if pipe_config is False:
                log.error(
                    f"'{current_model['model'].__name__}' pipe handling disabled"
                )
                break
            elif pipe_config is None:
                log.error(f"'{current_model['model'].__name__}' pipe not found")
                break
            # resolve pipe model
            if pipe_config == "self":
                pipe_model = current_model["model"]
            # import pipe model from module path string
            elif isinstance(pipe_config, str):
                # rpartition - returns a tuple of (before_last_dot, dot, after_last_dot)
                module_path, _, class_name = pipe_config.rpartition(".")
                module = __import__(module_path, fromlist=[""])
                pipe_model = getattr(module, class_name)
            else:
                pipe_model = pipe_config
            current_model = {
                "model": pipe_model,
                "fields": [],
                "parameter": parameter,
            }
            models = [current_model]
            ret.append(models)

        # collect JSON dictionary or list string
        elif parameter.strip().startswith(("{", "[")) and current_field:
            close = "}" if parameter.strip().startswith("{") else "]"
            value_items = [parameter]
            while parameters:
                parameter = parameters.pop(0)
                value_items.append(parameter)
                if parameter.strip().endswith(close):
                    break
            self._save_collected_value(current_field, " ".join(value_items))

        # collect quoted field value (single or double quotes)
        elif ('"' in parameter or "'" in parameter) and current_field:
            quote = '"' if '"' in parameter else "'"
            value_items = [parameter.replace(quote, "")]
            if parameter.count(quote) != 2:
                while parameters:
                    parameter = parameters.pop(0)
                    value_items.append(parameter.replace(quote, ""))
                    if quote in parameter:
                        break
            self._save_collected_value(current_field, " ".join(value_items))

        # handle exact match to model field by name, alias, or serialization_alias
        elif resolved := self._resolve_field(current_model["model"], parameter):
            parameter, field = resolved
            # record presence for previous field before moving on
            if current_field.get(
                "values"
            ) is ... and "presence" in current_field.get("json_schema_extra", {}):
                self._save_collected_value(
                    current_field,
                    current_field["json_schema_extra"]["presence"],
                )
            # handle next level model reference
            if isinstance(field.annotation, ModelMetaclass):
                current_model = {
                    "model": field.annotation,
                    "fields": [],
                    "parameter": parameter,
                }
                models.append(current_model)
                current_field = {}
                if len(ret) == 1:
                    current_model["defaults"] = self.extract_model_defaults(
                        field.annotation
                    )
            # handle dictionary reference
            elif (
                get_origin(field.annotation) in (dict, Dict)
                and field.json_schema_extra
                and field.json_schema_extra.get("pkey")
            ):
                key_name = field.json_schema_extra["pkey"]
                key_desc = field.json_schema_extra.get(
                    "pkey_description", "Input key"
                )

                # Get value type
                args = get_args(field.annotation)
                value_type = args[1] if len(args) > 1 else Any

                current_model = {
                    "model": VirtualDictModel(
                        key=key_name, description=key_desc, value_type=value_type
                    ),
                    "fields": [],
                    "parameter": parameter,
                }
                models.append(current_model)
                current_field = {}
            # handle actual field reference
            elif isinstance(field, FieldInfo):
                current_field = {
                    "name": parameter,
                    "values": ...,
                    "field": field,
                    "json_schema_extra": field.json_schema_extra or {},
                }
                # find and replace default value if present
                for idx, f in enumerate(current_model["fields"]):
                    if f["name"] == current_field["name"]:
                        current_model["fields"][idx] = current_field
                        break
                else:
                    current_model["fields"].append(current_field)
            else:
                raise TypeError(
                    f"Unsupported pydantic field type: '{type(field.annotation)}', "
                    f"parameter: '{parameter}', command: '{command}', current model: "
                    f"'{current_model['model']}'"
                )

        # check if last field is an Enumerator
        elif current_field and isinstance(
            current_field["field"].annotation, enum.EnumMeta
        ):
            if any(
                str(i.value) == parameter for i in current_field["field"].annotation
            ):
                self._save_collected_value(current_field, parameter)
            elif any(
                str(i.value).startswith(parameter)
                for i in current_field["field"].annotation
            ):
                raise FieldLooseMatchOnly(current_model, parameter)

        # check if parameter partially matches any model field
        elif self._has_partial_match(current_model["model"], parameter):
            raise FieldLooseMatchOnly(current_model, parameter)

        # parameter is a value, save it to current field
        elif current_field:
            self._save_collected_value(current_field, parameter)
        else:
            raise FieldKeyError(current_model, parameter)
    # check presence for last parameter is not is_help
    if (
        is_help is False
        and current_field.get("values") is ...
        and "presence" in current_field["json_schema_extra"]
    ):
        value = current_field["json_schema_extra"]["presence"]
        self._save_collected_value(current_field, value)

    # iterate over collected models and fields to see
    # if need to collect multi-line input
    if collect_multiline:
        for command_models in ret:
            for model in command_models:
                for field in model["fields"]:
                    self._collect_multiline(field)

    return ret

picle.App.postloop() -> None

Save readline history to file on shell exit.

Source code in picle\picle.py

def postloop(self) -> None:
    """
    Save readline history to file on shell exit.
    """
    if HAS_READLINE and self.history_file:
        try:
            _readline.write_history_file(os.path.expanduser(self.history_file))
        except OSError:
            pass

picle.App.print_model_help(models: list, verbose: bool = False, match: Optional[str] = None, print_help: bool = True) -> Optional[tuple[list[str], int]]

Form and print help message for model fields.

Parameters:

• models (list) –

  List of model dictionaries.

• verbose (bool, default: False ) –

  If True, print verbose help.

• match (str, default: None ) –

  Only collect help for fields that start with this string.

• print_help (bool, default: True ) –

  If True, prints help; otherwise returns tuple of help lines list and width of longest line.

Returns:

• Optional[tuple[list[str], int]] –

  Optional[tuple[list[str], int]]: Help lines and width if print_help is False.

Source code in picle\picle.py

def print_model_help(
    self,
    models: list,
    verbose: bool = False,
    match: Optional[str] = None,
    print_help: bool = True,
) -> Optional[tuple[list[str], int]]:
    """
    Form and print help message for model fields.

    Args:
        models (list): List of model dictionaries.
        verbose (bool): If True, print verbose help.
        match (str, optional): Only collect help for fields that start with this string.
        print_help (bool): If True, prints help; otherwise returns tuple of help lines list and width of longest line.

    Returns:
        Optional[tuple[list[str], int]]: Help lines and width if print_help is False.
    """
    model = models[-1][-1]  # get last model
    last_field = model["fields"][-1] if model["fields"] else None
    fparam = self._get_field_params(last_field)
    lines = {}  # dict of {cmd: cmd_help}
    lines_mandatory = {}  # dict of mandatory commands {cmd: cmd_help}
    width = 0  # record longest command width for padding
    # print help message only for last collected field
    if last_field and last_field["values"] == ...:
        field = last_field["field"]
        json_schema_extra = last_field["json_schema_extra"]
        name = f"<'{last_field['name']}' value>"
        # check if field referencing function
        if json_schema_extra.get("function"):
            lines[name] = f"{field.description}"
            name = "<ENTER>"
            lines[name] = "Execute command"
        # add options for enumerations
        elif isinstance(field.annotation, enum.EnumMeta):
            options = [i.value for i in field.annotation]
            lines[name] = ", ".join([str(i) for i in options])
        # check if model has method to source field choices
        elif hasattr(model["model"], f"source_{last_field['name']}"):
            source_callable = getattr(
                model["model"], f"source_{last_field['name']}"
            )
            if callable_expects_argument(source_callable, "choice"):
                options = source_callable(choice="")
            else:
                options = source_callable()
            lines[name] = ", ".join([str(i) for i in options])
        else:
            lines[name] = f"{field.description}"
            # check if field supports multiline input
            if fparam.get("multiline") is True:
                lines["load-terminal"] = "Collect value using multi line input mode"
            if verbose:
                lines[name] += (
                    f"; default '{field.get_default()}', type '{str(field.annotation)}', "
                    f"is required - {field.is_required()}"
                )
    # collect help message for all fields of this model
    else:
        # check if model supports subshell
        if (
            hasattr(model["model"], "PicleConfig")
            and getattr(model["model"].PicleConfig, "subshell", None) is True
            # exclude <ENTER> if already in model's shell
            and not self.shells[-1] == model["model"]
        ):
            name = "<ENTER>"
            lines[name] = "Enter command subshell"
        # iterate over model fields
        for name, field in model_fields(model["model"]).items():
            # skip fields that already have values
            if any(f["name"] == name for f in model["fields"]):
                continue
            # check if field has alias
            if field.alias:
                name = field.alias
            # check if field has serialization alias
            if field.serialization_alias:
                name = field.serialization_alias
            # filter fields
            if match and not name.startswith(match):
                continue
            # make mandatory fields standing out
            if field.is_required():
                lines_mandatory[name] = f"{field.description}"
            else:
                lines[name] = f"{field.description}"
            if verbose:
                lines[name] += (
                    f"; default '{field.get_default()}', type '{str(field.annotation)}', "
                    f"is required - {field.is_required()}"
                )

        # collect VirtualDictModel description
        if isinstance(model["model"], VirtualDictModel):
            vdm = model["model"]
            k = f"<{vdm.key}>"
            if not match or k.startswith(match or ""):
                lines_mandatory[k] = vdm.description or ""
                width = max(width, len(k))

    # check if model has pipe defined
    if self._find_pipe_config(models[-1]):
        name = "|"
        lines[name] = "Execute pipe command"
    width = max((len(k) for k in lines), default=width)
    # form help lines for mandatory fields first
    help_msg = []
    for k in sorted(lines_mandatory.keys()):
        padding = " " * (width - len(k)) + (" " * 4)
        if self.use_rich and HAS_RICH:
            help_msg.append(f" [bold]{k}[/bold]{padding}{lines_mandatory[k]}")
        else:
            help_msg.append(f" \033[1m{k}\033[0m{padding}{lines_mandatory[k]}")
    # form help lines for non-mandatory fields
    for k in sorted(lines.keys()):
        padding = " " * (width - len(k)) + (" " * 4)
        help_msg.append(f" {k}{padding}{lines[k]}")
    # make sure ENTER is at the end of help message if subshell supported
    enter_line = [c for c in help_msg if "<ENTER>" in c]
    if enter_line:
        help_msg.remove(enter_line[0])
        help_msg.append(enter_line[0])

    if print_help:  # print help message
        self.write(self.newline.join(help_msg))
    else:
        return help_msg, width

picle.App.process_help_command(line: str) -> None

Process inline help triggered by '?' or '??' at the end of a command line.

Parameters:

• line (str) –

  Input command line string ending with '?' or '??'.

Source code in picle\picle.py

def process_help_command(self, line: str) -> None:
    """
    Process inline help triggered by '?' or '??' at the end of a command line.

    Args:
        line (str): Input command line string ending with '?' or '??'.
    """
    verbose = line.endswith("??")
    try:
        command_models = self.parse_command(line.rstrip("?"), is_help=True)
    except FieldLooseMatchOnly as e:
        model, parameter = e.args
        self.print_model_help([[model]], verbose=verbose, match=parameter)
    except FieldKeyError as e:
        model, parameter = e.args
        self.write_error(
            f"Incorrect command, '{parameter}' not part of "
            f"'{self._get_model_name(model)}' model fields"
        )
    else:
        self.print_model_help(command_models, verbose=verbose)

picle.App.write(output: str) -> None

Write output to stdout.

Parameters:

• output (str) –

  Output to write to stdout.

Source code in picle\picle.py

def write(self, output: str) -> None:
    """
    Write output to stdout.

    Args:
        output (str): Output to write to stdout.
    """
    if self.use_rich and HAS_RICH:
        RICHCONSOLE.print(output)
    else:
        if not isinstance(output, str):
            output = str(output)
        if not output.endswith(self.newline):
            output += self.newline
        self.stdout.write(output)
    self.stdout.flush()

picle.App.write_error(output: str) -> None

Write error output to stdout in red color.

Parameters:

• output (str) –

  Error message to write to stdout.

Source code in picle\picle.py

def write_error(self, output: str) -> None:
    """
    Write error output to stdout in red color.

    Args:
        output (str): Error message to write to stdout.
    """
    if self.use_rich and HAS_RICH:
        self.write(f"[red]{output}[/red]")
    else:
        self.write(f"\033[31m{output}\033[0m")

PICLE Build In Models

picle.models.Filters

Bases: BaseModel

picle.models.Filters.filter_exclude(data: Any, exclude: Any = None) -> str staticmethod

Filter data line by line using provided pattern. Returns only lines that do not contain the requested exclude pattern.

Parameters:

• data (Any) –

  Data to filter.

• exclude (Any, default: None ) –

  Pattern to filter data.

Returns:

• str ( str ) –

  Filtered lines joined by newline.

Source code in picle\models.py

@staticmethod
def filter_exclude(data: Any, exclude: Any = None) -> str:
    """
    Filter data line by line using provided pattern. Returns only lines that do not contain the requested exclude pattern.

    Args:
        data: Data to filter.
        exclude: Pattern to filter data.

    Returns:
        str: Filtered lines joined by newline.
    """
    exclude = str(exclude)
    return "\n".join(
        [line for line in str(data).splitlines() if exclude not in line]
    )

picle.models.Filters.filter_include(data: Any, include: Any = None) -> str staticmethod

Filter data line by line using provided pattern. Returns only lines that contain the requested include pattern.

Parameters:

• data (Any) –

  Data to filter.

• include (Any, default: None ) –

  Pattern to filter data.

Returns:

• str ( str ) –

  Filtered lines joined by newline.

Source code in picle\models.py

@staticmethod
def filter_include(data: Any, include: Any = None) -> str:
    """
    Filter data line by line using provided pattern. Returns only lines that contain the requested include pattern.

    Args:
        data: Data to filter.
        include: Pattern to filter data.

    Returns:
        str: Filtered lines joined by newline.
    """
    include = str(include)
    return "\n".join([line for line in str(data).splitlines() if include in line])

picle.models.Filters.filter_last(data: Any, last: int = None) -> str staticmethod

Returns only the last N lines.

Parameters:

• data (Any) –

  Data to filter.

• last (int, default: None ) –

  Number of lines to return from the end.

Returns:

• str ( str ) –

  Last N lines joined by newline.

Source code in picle\models.py

@staticmethod
def filter_last(data: Any, last: int = None) -> str:
    """
    Returns only the last N lines.

    Args:
        data: Data to filter.
        last: Number of lines to return from the end.

    Returns:
        str: Last N lines joined by newline.
    """
    lines = str(data).splitlines()
    return "\n".join(lines[-last:] if last else lines)

picle.models.Outputters

Bases: BaseModel

picle.models.Outputters.outputter_json(data: Union[dict, list, bytes], indent: int = 4, sort_keys: bool = True, ensure_ascii: bool = True, separators: tuple = None) -> Any staticmethod

Pretty print JSON string.

Parameters:

• data (dict, list, or bytes) –

  Data to print.

• indent (int, default: 4 ) –

  Indentation for JSON output.

• sort_keys (bool, default: True ) –

  Sort dictionary keys in output.

• ensure_ascii (bool, default: True ) –

  Escape non-ASCII characters in output.

• separators (tuple, default: None ) –

  Item and key separators, e.g. (',', ': ').

Returns:

• Any ( Any ) –

  JSON-formatted string or error message.

Source code in picle\models.py

@staticmethod
def outputter_json(
    data: Union[dict, list, bytes],
    indent: int = 4,
    sort_keys: bool = True,
    ensure_ascii: bool = True,
    separators: tuple = None,
) -> Any:
    """
    Pretty print JSON string.

    Args:
        data (dict, list, or bytes): Data to print.
        indent (int): Indentation for JSON output.
        sort_keys (bool): Sort dictionary keys in output.
        ensure_ascii (bool): Escape non-ASCII characters in output.
        separators (tuple): Item and key separators, e.g. (',', ': ').

    Returns:
        Any: JSON-formatted string or error message.
    """
    if isinstance(data, bytes):
        data = data.decode("utf-8")

    if isinstance(data, str):
        return data

    # data should be a json string
    try:
        data = json.dumps(
            data,
            indent=indent,
            sort_keys=sort_keys,
            ensure_ascii=ensure_ascii,
            separators=separators,
        )
    except Exception as e:
        print(
            f"ERROR: Failed to format data as JSON string:\n{data}\n\nError: '{e}'"
        )

    return data

picle.models.Outputters.outputter_kv(data: dict, parent_key='', separator='.', is_top=True) -> str staticmethod

Turn a nested structure (combination of lists/dictionaries) into a flattened dictionary.

This function is useful to explore deeply nested structures.

Source code in picle\models.py

@staticmethod
def outputter_kv(data: dict, parent_key="", separator=".", is_top=True) -> str:
    """
    Turn a nested structure (combination of lists/dictionaries) into a
    flattened dictionary.

    This function is useful to explore deeply nested structures.
    """
    items = []
    if isinstance(data, dict):
        for key, value in data.items():
            new_key = (
                "{}{}{}".format(parent_key, separator, key) if parent_key else key
            )
            items.extend(
                Outputters.outputter_kv(
                    value, new_key, separator, is_top=False
                ).items()
            )
    elif isinstance(data, list):
        for k, v in enumerate(data):
            new_key = "{}{}{}".format(parent_key, separator, k) if parent_key else k
            items.extend(
                Outputters.outputter_kv(
                    {str(new_key): v}, separator=separator, is_top=False
                ).items()
            )
    else:
        items.append((parent_key, data))

    if is_top:
        return "\n".join(f"{k}: {v}" if k else v for k, v in items)
    else:
        return dict(items)

picle.models.Outputters.outputter_nested(data: Union[dict, list], initial_indent: int = 0, with_tables: bool = False, tabulate_kwargs: dict = None) -> str staticmethod

Recursively formats and prints nested data structures (dictionaries and lists) in a human-readable format.

Parameters:

• data (dict or list) –

  Nested data structure to be formatted and printed.

• initial_indent (int, default: 0 ) –

  Initial indentation level.

• with_tables (bool, default: False ) –

  If True, will format flat lists as Tabulate tables.

• tabulate_kwargs (dict, default: None ) –

  Arguments for tabulate table outputter.

Returns:

• str ( str ) –

  Formatted nested string.

Source code in picle\models.py

@staticmethod
def outputter_nested(
    data: Union[dict, list],
    initial_indent: int = 0,
    with_tables: bool = False,
    tabulate_kwargs: dict = None,
) -> str:
    """
    Recursively formats and prints nested data structures (dictionaries and lists) in a human-readable format.

    Args:
        data (dict or list): Nested data structure to be formatted and printed.
        initial_indent (int): Initial indentation level.
        with_tables (bool): If True, will format flat lists as Tabulate tables.
        tabulate_kwargs (dict, optional): Arguments for tabulate table outputter.

    Returns:
        str: Formatted nested string.
    """
    tabulate_kwargs = tabulate_kwargs or {"tablefmt": "simple"}

    key_styles = {
        1: "bold green",
        2: "bold blue",
        3: "bold yellow",
        4: "bold magenta",
    }

    def is_dictionary_list(data):
        for item in data:
            if not isinstance(item, Mapping):
                return False
            for i in item.values():
                if isinstance(i, (list, tuple, Mapping)):
                    return False
        return True

    def ustring(indent, msg, prefix="", suffix=""):
        indent *= " "
        fmt = "{0}{1}{2}{3}"
        return fmt.format(indent, prefix, msg, suffix)

    def style_key(key, level):
        key = str(key)
        style = key_styles.get(level)
        if HAS_RICH and style:
            return f"[{style}]{key}[/{style}]"
        return key

    def nest(ret, indent, prefix, out, key_level=0):
        if isinstance(ret, bytes):
            try:
                ret = ret.decode("utf-8")
            except UnicodeDecodeError:
                ret = str(ret)

        if ret is None or ret is True or ret is False:
            out.append(ustring(indent, ret, prefix=prefix))
        elif isinstance(ret, Number):
            out.append(ustring(indent, repr(ret), prefix=prefix))
        elif isinstance(ret, str):
            first_line = True
            for line in ret.splitlines():
                line_prefix = " " * len(prefix) if not first_line else prefix
                out.append(ustring(indent, line, prefix=line_prefix))
                first_line = False
        elif isinstance(ret, (list, tuple)):
            # make a text table if it is a flat list
            if with_tables and is_dictionary_list(ret):
                table = Outputters.outputter_tabulate_table(ret, **tabulate_kwargs)
                nest(table, indent + 2, prefix, out, key_level)
            else:
                for ind in ret:
                    if isinstance(ind, (list, tuple, Mapping)):
                        out.append(ustring(indent, "|_"))
                        prefix = "" if isinstance(ind, Mapping) else "- "
                        nest(ind, indent + 2, prefix, out, key_level)
                    else:
                        nest(ind, indent, "- ", out, key_level)
        elif isinstance(ret, Mapping):
            if indent:
                out.append(ustring(indent, "----------"))

            for key in ret.keys():
                val = ret[key]
                styled_key = style_key(key, key_level + 1)
                out.append(ustring(indent, styled_key, suffix=":", prefix=prefix))
                # Dict depth controls key coloring; list depth does not.
                nest(val, indent + 4, "", out, key_level + 1)

        return out

    # make sure data is sorted
    try:
        if isinstance(data, dict):
            data = dict(sorted(data.items()))
        elif isinstance(data, list):
            if data and isinstance(data[0], dict):
                first_key = next(iter(data[0]))
                data = list(sorted(data, key=lambda x: x.get(first_key, "")))
            else:
                data = list(sorted(data))
    except Exception as e:
        log.warning(f"Nested outputter data sorting failed: '{e}'")

    lines = nest(data, initial_indent, "", [], key_level=0)
    lines = "\n".join(lines)

    return lines

picle.models.Outputters.outputter_pprint(data: Any, indent: int = 4, width: int = 80, depth: int = None, compact: bool = False, sort_dicts: bool = True) -> str staticmethod

Pretty-print results using Python's pprint module.

Parameters:

• data (Any) –

  Any data to pretty-print.

• indent (int, default: 4 ) –

  Indentation added for each nesting level.

• width (int, default: 80 ) –

  Maximum number of characters per line.

• depth (int, default: None ) –

  Maximum depth of nested structures to display.

• compact (bool, default: False ) –

  Fit as many items as possible on each line.

• sort_dicts (bool, default: True ) –

  Sort dictionary keys before display.

Returns:

• str ( str ) –

  Nicely formatted string representation.

Source code in picle\models.py

@staticmethod
def outputter_pprint(
    data: Any,
    indent: int = 4,
    width: int = 80,
    depth: int = None,
    compact: bool = False,
    sort_dicts: bool = True,
) -> str:
    """
    Pretty-print results using Python's pprint module.

    Args:
        data: Any data to pretty-print.
        indent (int): Indentation added for each nesting level.
        width (int): Maximum number of characters per line.
        depth (int): Maximum depth of nested structures to display.
        compact (bool): Fit as many items as possible on each line.
        sort_dicts (bool): Sort dictionary keys before display.

    Returns:
        str: Nicely formatted string representation.
    """
    if isinstance(data, str):
        return data
    return pprint.pformat(
        data,
        indent=indent,
        width=width,
        depth=depth,
        compact=compact,
        sort_dicts=sort_dicts,
    )

picle.models.Outputters.outputter_rich_markdown(data: Any) -> Any staticmethod

Print markdown output using Rich library.

Parameters:

• data (Any) –

  Any data to print.

Returns:

• Any ( Any ) –

  Rich Markdown object or string.

Source code in picle\models.py

@staticmethod
def outputter_rich_markdown(data: Any) -> Any:
    """
    Print markdown output using Rich library.

    Args:
        data: Any data to print.

    Returns:
        Any: Rich Markdown object or string.
    """
    if not isinstance(data, str):
        data = str(data)

    if HAS_RICH:
        return Markdown(data)
    else:
        return data

picle.models.Outputters.outputter_rich_table(data: list[dict], headers: list = None, title: str = None, sortby: str = None) -> Any staticmethod

Format a list of dictionaries as a Rich table.

Parameters:

• data (list[dict]) –

  List of dictionaries to display.

• headers (list, default: None ) –

  Column headers; defaults to the keys of the first row.

• title (str, default: None ) –

  Table title.

• sortby (str, default: None ) –

  Key name to sort rows by.

Returns:

• Any ( Any ) –

  A Rich Table object, or the original data if Rich is unavailable.

Source code in picle\models.py

@staticmethod
def outputter_rich_table(
    data: list[dict], headers: list = None, title: str = None, sortby: str = None
) -> Any:
    """
    Format a list of dictionaries as a Rich table.

    Args:
        data (list[dict]): List of dictionaries to display.
        headers (list, optional): Column headers; defaults to the keys of the first row.
        title (str, optional): Table title.
        sortby (str, optional): Key name to sort rows by.

    Returns:
        Any: A Rich Table object, or the original data if Rich is unavailable.
    """
    if not HAS_RICH or not isinstance(data, list):
        return data

    if not data:
        return data

    headers = headers or list(data[0].keys())
    table = RICHTABLE(title=title, box=False)

    # add table columns
    for h in headers:
        table.add_column(h, justify="left", no_wrap=True)

    # sort the table
    if sortby:
        sorted_data = sorted(data, key=lambda d: d[sortby])
    else:
        sorted_data = data

    # add table rows
    for item in sorted_data:
        cells = [str(item.get(h, "")) for h in headers]
        table.add_row(*cells)

    return table

picle.models.Outputters.outputter_save(data: Any, save: str) -> Any staticmethod

Output data into a file.

Parameters:

• data (Any) –

  Any data to print.

• save (str) –

  File path to save data.

Returns:

• Any ( Any ) –

  The data that was saved.

Source code in picle\models.py

@staticmethod
def outputter_save(data: Any, save: str) -> Any:
    """
    Output data into a file.

    Args:
        data: Any data to print.
        save (str): File path to save data.

    Returns:
        Any: The data that was saved.
    """
    # create directories
    abspath = os.path.abspath(save)
    dirs = os.path.split(abspath)[0]
    os.makedirs(dirs, exist_ok=True)

    # save data to file
    with open(save, "w") as f:
        if isinstance(data, str):
            f.write(data)
        else:
            f.write(str(data))

    return data

picle.models.Outputters.outputter_tabulate_table(data: list, headers_exclude: list = None, sortby: str = None, reverse: bool = False, tablefmt: str = 'grid', headers: list = None, showindex: bool = True, maxcolwidths: int = None) -> Any staticmethod

Format and output data as a text table using the tabulate library.

Parameters:

• data (list) –

  A list of dictionaries or list of lists to be formatted into a table. If it is a list of lists, the function merges nested lists.

• headers_exclude (list, default: None ) –

  A list or comma-separated string of headers to exclude from the table.

• sortby (str, default: None ) –

  The key name to sort the table by. If None, no sorting is applied.

• reverse (bool, default: False ) –

  If True, reverses the sort order. Defaults to False.

• tablefmt (str, default: 'grid' ) –

  Table format style.

• headers (list or str, default: None ) –

  Specifies the table headers. Can be a list, a comma-separated string, or 'keys' to use dictionary keys as headers.

• showindex (bool, default: True ) –

  If True, includes an index column in the table.

• maxcolwidths (int, default: None ) –

  Maximum width of the column before wrapping text.

Returns:

• Any ( Any ) –

  Tabulated table string or error message.

Source code in picle\models.py

@staticmethod
def outputter_tabulate_table(
    data: list,
    headers_exclude: list = None,
    sortby: str = None,
    reverse: bool = False,
    tablefmt: str = "grid",
    headers: list = None,
    showindex: bool = True,
    maxcolwidths: int = None,
) -> Any:
    """
    Format and output data as a text table using the tabulate library.

    Args:
        data (list): A list of dictionaries or list of lists to be formatted into a table. If it is a list of lists, the function merges nested lists.
        headers_exclude (list, optional): A list or comma-separated string of headers to exclude from the table.
        sortby (str, optional): The key name to sort the table by. If None, no sorting is applied.
        reverse (bool, optional): If True, reverses the sort order. Defaults to False.
        tablefmt (str): Table format style.
        headers (list or str, optional): Specifies the table headers. Can be a list, a comma-separated string, or 'keys' to use dictionary keys as headers.
        showindex (bool, optional): If True, includes an index column in the table.
        maxcolwidths (int, optional): Maximum width of the column before wrapping text.

    Returns:
        Any: Tabulated table string or error message.
    """
    if not HAS_TABULATE:
        log.error(
            "PICLE Table outputter tabulate library import failed, install: pip install tabulate"
        )
        return data
    if not isinstance(data, list):
        log.error("PICLE Table outputter data is not a list")
        return data

    # transform headers to exclude argument
    headers_exclude = headers_exclude or []
    if isinstance(headers_exclude, str) and "," in headers_exclude:
        headers_exclude = [i.strip() for i in headers_exclude.split(",")]

    # form base tabulate arguments
    if isinstance(headers, str):
        headers = [i.strip() for i in headers.split(",")]
    elif headers is None:
        headers = "keys"

    tabulate_kw = {
        "headers": headers,
        "tablefmt": tablefmt,
        "maxcolwidths": maxcolwidths,
    }

    # form singe table out of list of lists
    table_ = []
    while data:
        item = data.pop(0)
        if isinstance(item, list):
            table_.extend(item)
        else:
            table_.append(item)
    data = table_

    # sort results
    if sortby:
        data = sorted(
            data,
            reverse=reverse,
            key=lambda item: str(item.get(sortby, "")),
        )

    # filter table headers if requested to do so
    if headers_exclude:
        data = [
            {k: v for k, v in res.items() if k not in headers_exclude}
            for res in data
        ]

    # transform data content to match headers
    if isinstance(tabulate_kw["headers"], list):
        data = [[item.get(i, "") for i in tabulate_kw["headers"]] for item in data]

    # start index with 1 instead of 0
    if showindex is True:
        showindex = range(1, len(data) + 1)
        tabulate_kw["showindex"] = showindex

    return tabulate_lib.tabulate(data, **tabulate_kw)

picle.models.Outputters.outputter_yaml(data: Union[dict, list, bytes], absolute_indent: int = 0, indent: int = 2, sort_keys: bool = True, allow_unicode: bool = True, width: int = None, default_flow_style: bool = False) -> Any staticmethod

Format structured data as a YAML string.

Parameters:

• data (dict, list, or bytes) –

  Data to print.

• absolute_indent (int, default: 0 ) –

  Indentation to prepend for entire output.

• indent (int, default: 2 ) –

  Indentation for YAML output.

• sort_keys (bool, default: True ) –

  Sort dictionary keys in output.

• allow_unicode (bool, default: True ) –

  Allow Unicode characters instead of escaping them.

• width (int, default: None ) –

  Maximum line width before wrapping.

• default_flow_style (bool, default: False ) –

  Use flow style for collections.

Returns:

• Any ( Any ) –

  YAML-formatted string or error message.

Source code in picle\models.py

@staticmethod
def outputter_yaml(
    data: Union[dict, list, bytes],
    absolute_indent: int = 0,
    indent: int = 2,
    sort_keys: bool = True,
    allow_unicode: bool = True,
    width: int = None,
    default_flow_style: bool = False,
) -> Any:
    """
    Format structured data as a YAML string.

    Args:
        data (dict, list, or bytes): Data to print.
        absolute_indent (int): Indentation to prepend for entire output.
        indent (int): Indentation for YAML output.
        sort_keys (bool): Sort dictionary keys in output.
        allow_unicode (bool): Allow Unicode characters instead of escaping them.
        width (int): Maximum line width before wrapping.
        default_flow_style (bool): Use flow style for collections.

    Returns:
        Any: YAML-formatted string or error message.
    """
    if isinstance(data, bytes):
        data = data.decode("utf-8")

    if isinstance(data, str):
        return data

    # data should be a YAML string
    try:
        if HAS_YAML:
            data = yaml.safe_dump(
                data,
                default_flow_style=default_flow_style,
                sort_keys=sort_keys,
                indent=indent,
                allow_unicode=allow_unicode,
                width=width,
            )
            # add indent
            if absolute_indent:
                data = "\n".join(
                    [f"{' ' * absolute_indent}{i}" for i in data.splitlines()]
                )
        else:
            log.error(
                "PICLE YAML outputter yaml library import failed, install: pip install pyyaml"
            )
    except Exception as e:
        print(
            f"ERROR: Failed to format data as YAML string:\n{data}\n\nError: '{e}'"
        )

    return data

picle.models.PipeFunctionsModel

Bases: Filters, Outputters

Collection of common pipe functions to use in PICLE shell models

picle.models.MAN

Bases: BaseModel

Manual and documentation related functions

picle.models.MAN.print_model_json_schema(root_model: object, **kwargs: dict) -> str staticmethod

Print model JSON schema for shell model specified by dot separated path (e.g. model.shell.command).

Parameters:

• root_model (object) –

  PICLE App root model to print JSON schema for.

• **kwargs (dict, default: {} ) –

  Additional keyword arguments (expects 'json_schema' for path).

Returns:

• str ( str ) –

  JSON schema as a formatted string.

Source code in picle\models.py

@staticmethod
def print_model_json_schema(root_model: object, **kwargs: dict) -> str:
    """
    Print model JSON schema for shell model specified by dot separated path (e.g. model.shell.command).

    Args:
        root_model: PICLE App root model to print JSON schema for.
        **kwargs: Additional keyword arguments (expects 'json_schema' for path).

    Returns:
        str: JSON schema as a formatted string.
    """

    class MyGenerateJsonSchema(GenerateJsonSchema):
        def handle_invalid_for_json_schema(
            self, schema: core_schema.CoreSchema, error_info: str
        ) -> JsonSchemaValue:
            raise PydanticOmit

        def callable_schema(self, schema):
            print(schema)
            raise PydanticOmit

        def render_warning_message(kind, detail: str) -> None:
            print(kind, detail)

    path = kwargs["json_schema"].split(".") if kwargs.get("json_schema") else []
    model = MAN._recurse_to_model(root_model, path=path)
    return json.dumps(
        model.model_json_schema(schema_generator=MyGenerateJsonSchema),
        indent=4,
        sort_keys=True,
    )

picle.models.MAN.print_model_tree(root_model: object, **kwargs: dict) -> None staticmethod

Print model tree for shell model specified by dot separated path (e.g. model.shell.command).

Parameters:

• root_model (object) –

  PICLE App root model to print tree for.

• **kwargs (dict, default: {} ) –

  Additional keyword arguments (expects 'tree' for path).

Source code in picle\models.py

@staticmethod
def print_model_tree(root_model: object, **kwargs: dict) -> None:
    """
    Print model tree for shell model specified by dot separated path (e.g. model.shell.command).

    Args:
        root_model: PICLE App root model to print tree for.
        **kwargs: Additional keyword arguments (expects 'tree' for path).
    """
    if HAS_RICH:
        RICHCONSOLE.print(
            "\n[bold]R[/bold] - required field, "
            + "[bold]M[/bold] - supports multiline input, "
            + "[bold]D[/bold] - dynamic key\n"
        )
        path = kwargs["tree"].split(".") if kwargs.get("tree") else []
        rich_tree = RICHTREE("[bold]root[/bold]")
        RICHCONSOLE.print(
            MAN._construct_model_tree(
                model=root_model.model_construct(), tree=rich_tree, path=path
            )
        )
    else:
        log.error(
            "PICLE model tree outputter requires Rich library, install: pip install rich"
        )