Skip to content

Documentation for Thermite

A package for easily creating CLIs.

CliParamKind

Bases: Enum

OPTION instance-attribute class-attribute

OPTION = 'OPTION'

ARGUMENT instance-attribute class-attribute

ARGUMENT = 'ARGUMENT'

Event

Bases: Enum

START_ARGS_PRE_PROCESS instance-attribute class-attribute

START_ARGS_PRE_PROCESS = 'START_ARGS_PRE_PROCESS'

SIG_EXTRACT instance-attribute class-attribute

SIG_EXTRACT = 'SIG_EXTRACT'

CMD_POST_CREATE instance-attribute class-attribute

CMD_POST_CREATE = 'CMD_POST_CREATE'

CMD_POST_PROCESS instance-attribute class-attribute

CMD_POST_PROCESS = 'CMD_POST_PROCESS'

PG_POST_CREATE instance-attribute class-attribute

PG_POST_CREATE = 'PG_POST_CREATE'

CMD_FINISH instance-attribute class-attribute

CMD_FINISH = 'CMD_FINISH'

ParameterSignature

name instance-attribute

name: str

python_kind instance-attribute

python_kind: inspect._ParameterKind

cli_kind instance-attribute

cli_kind: CliParamKind

descr instance-attribute

descr: Optional[str]

default_value instance-attribute

default_value: Any

annot instance-attribute

annot: Type

ParameterGroup

Bases: MutableMapping

name instance-attribute class-attribute

name: str = ''

short_descr instance-attribute class-attribute

short_descr: Optional[str] = None

long_descr instance-attribute class-attribute

long_descr: Optional[str] = None

return_annot instance-attribute

return_annot: Type

obj instance-attribute class-attribute

obj: Any = None

default_value instance-attribute class-attribute

default_value: Any = field(default=Ellipsis)

python_kind instance-attribute

python_kind: Optional[inspect._ParameterKind]

params instance-attribute class-attribute

params: Dict[str, Union[Parameter, ParameterGroup]] = field(
    factory=dict
)

posargs property

posargs: List[Union[Parameter, ParameterGroup]]

varposargs property

varposargs: List[Union[Parameter, ParameterGroup]]

kwargs property

kwargs: Dict[str, Union[Parameter, ParameterGroup]]

py_args_values property

py_args_values: Tuple[Any, ...]

py_args_values_with_excs property

py_args_values_with_excs: Tuple[Any, ...]

py_kwargs_values property

py_kwargs_values: Dict[str, Any]

py_kwargs_values_with_excs property

py_kwargs_values_with_excs: Dict[str, Any]

unset property

unset: bool

is_required property

is_required: bool

value property

value: Any

cli_args property

cli_args: Dict[str, Argument]

cli_opts property

cli_opts: Dict[str, Option]

cli_pgs property

cli_pgs: Dict[str, ParameterGroup]

cli_opts_recursive property

cli_opts_recursive: Dict[str, Option]

cli_args_recursive property

cli_args_recursive: Dict[str, Argument]

process

process(input_args)
Source code in thermite/parameters/group.py
def process(self, input_args: Sequence[str]) -> Sequence[str]:
    if len(input_args) == 0:
        return []

    if input_args[0].startswith("-"):
        opts_by_trigger = self.cli_opts_recursive
        if input_args[0] in opts_by_trigger:
            self._num_bound += 1
            opt = opts_by_trigger[input_args[0]]
            bind_res = opt.process(input_args)

            return bind_res
        else:
            raise TriggerError(f"No option with trigger {input_args[0]}")

    else:
        for argument in self.cli_args_recursive.values():
            if argument.unset:
                self._num_bound += 1
                args_use, args_remain = split_args_by_nargs(
                    input_args, argument.type_converter.num_req_args
                )
                argument.process(args_use)
                return args_remain

    return input_args

ConstantTriggerProcessor

Bases: TriggerProcessor

constant instance-attribute

constant: Any

bind

bind(args)
Source code in thermite/parameters/processors.py
def bind(self, args: Sequence[str]) -> Sequence[str]:
    if len(args) == 0:
        raise TriggerError("A trigger is expected.")
    if args[0] not in self.triggers:
        raise TriggerError(f"Trigger {args[0]} not an allowed trigger.")
    return args[1:]

process

process(value)
Source code in thermite/parameters/processors.py
def process(self, value: Any) -> Any:
    del value
    return self.constant

ObjSignature

short_descr instance-attribute

short_descr: Optional[str]

long_descr instance-attribute

long_descr: Optional[str]

params instance-attribute

params: Dict[str, ParameterSignature]

return_annot instance-attribute

return_annot: Type

Parameter

Bases: ABC, ParameterSignature

Base class for Parameters.

unset property

unset: bool

is_required property

is_required: bool

value property

value: Any

process abstractmethod

process(args)

Get a list of arguments and returns any unused ones.

Source code in thermite/parameters/base.py
@abstractmethod
def process(self, args: Sequence[str]) -> Sequence[str]:
    """Get a list of arguments and returns any unused ones."""
    ...

CliCallback

callback instance-attribute

callback: Callable

triggers instance-attribute

triggers: List[str]

descr instance-attribute

descr: str

num_req_args instance-attribute class-attribute

num_req_args: Union[int, slice] = 0

execute

execute(cmd, args)
Source code in thermite/command.py
def execute(self, cmd: "Command", args: Sequence[str]) -> Optional[Sequence[str]]:
    if args[0] in self.triggers:
        num_args_use = args_used(
            num_offered=len(args) - 1, num_req=self.num_req_args
        )
        self.callback(cmd, *args[1 : (1 + num_args_use)])
        return args[(1 + num_args_use) :]
    else:
        raise Exception("Callback was raised without appropriate trigger.")

ConvertTriggerProcessor

Bases: TriggerProcessor

type_converter instance-attribute

type_converter: CLIArgConverterBase

bound_args instance-attribute class-attribute

bound_args: Sequence[str] = field(factory=list, init=False)

allow_replace instance-attribute class-attribute

allow_replace: bool = False

bind

bind(args)
Source code in thermite/parameters/processors.py
def bind(self, args: Sequence[str]) -> Sequence[str]:
    if len(args) == 0:
        raise TriggerError("A trigger is expected.")
    if args[0] not in self.triggers:
        raise TriggerError(f"Trigger {args[0]} not an allowed trigger.")

    bound_args, ret_args = split_args_by_nargs(
        args[1:], num_req_args=self.type_converter.num_req_args
    )

    self.bound_args = bound_args
    return ret_args

process

process(value)
Source code in thermite/parameters/processors.py
def process(self, value: Any) -> Any:
    if value != ... and not self.allow_replace:
        raise TriggerError("Trigger already used once.")
    return self.type_converter.convert(self.bound_args)

Config

cli_callbacks instance-attribute class-attribute

cli_callbacks: List[CliCallback] = field(factory=list)

cli_callbacks_top_level instance-attribute class-attribute

cli_callbacks_top_level: List[CliCallback] = field(
    factory=list
)

cli_args_store instance-attribute class-attribute

cli_args_store: CLIArgConverterStore = field(
    factory=partial(CLIArgConverterStore, add_defaults=True)
)

event_callbacks instance-attribute class-attribute

event_callbacks: EventCallbacks = field(
    factory=partial(EventCallbacks, add_defaults=True)
)

SplitterClass instance-attribute class-attribute

SplitterClass: Type = EagerCliArgsSplitter

add_cli_callback

add_cli_callback(cb)
Source code in thermite/config.py
def add_cli_callback(self, cb: "CliCallback"):
    self.cli_callbacks.append(cb)

get_event_cbs

get_event_cbs(event)
Source code in thermite/config.py
def get_event_cbs(self, event: Union[str, Event]) -> List[Callable]:
    return self.event_callbacks.event_cb_dict[event]

event_cb_deco

event_cb_deco(event, obj=None)
Source code in thermite/config.py
def event_cb_deco(self, event: Union[str, Event], obj: Any = None):
    def add_event_cb_inner(cb: Callable):
        self.event_callbacks.add_event_cb(event=event, cb=cb, obj=obj)
        return cb

    return add_event_cb_inner

MultiConvertTriggerProcessor

Bases: TriggerProcessor

type_converter instance-attribute

type_converter: CLIArgConverterBase

bound_args instance-attribute class-attribute

bound_args: Sequence[str] = field(factory=list, init=False)

bind

bind(args)
Source code in thermite/parameters/processors.py
def bind(self, args: Sequence[str]) -> Sequence[str]:
    if len(args) == 0:
        raise TriggerError("A trigger is expected.")
    if args[0] not in self.triggers:
        raise TriggerError(f"Trigger {args[0]} not an allowed trigger.")

    bound_args, ret_args = split_args_by_nargs(
        args[1:], num_req_args=self.type_converter.num_req_args
    )

    self.bound_args = bound_args
    return ret_args

process

process(value)
Source code in thermite/parameters/processors.py
def process(self, value: Any) -> Any:
    append_val = self.type_converter.convert(self.bound_args)
    if not isinstance(value, list):
        return [append_val]
    else:
        value = value.copy()
        value.append(append_val)
        return value

to_convert_trigger_processor

to_convert_trigger_processor()
Source code in thermite/parameters/processors.py
def to_convert_trigger_processor(self) -> ConvertTriggerProcessor:
    inner_converter = self.type_converter
    if not isinstance(inner_converter.num_req_args, int):
        raise Exception(
            "Inner type converter needs ask for constant number " "of arguments"
        )
    return ConvertTriggerProcessor(
        type_converter=ListCLIArgConverter(
            target_type=List[inner_converter.target_type],  # type: ignore
            inner_converter=inner_converter,
        ),
        triggers=self.triggers,
        res_type=self.res_type,
    )

Option

Bases: Parameter

Base class for Options.

processors instance-attribute

processors: List[TriggerProcessor]

final_triggers property

final_triggers: Sequence[str]

process

process(args)

Implement of general argument processing.

Source code in thermite/parameters/base.py
def process(self, args: Sequence[str]) -> Sequence[str]:
    """Implement of general argument processing."""
    trigger_by_processor = self._final_trigger_by_processor
    if len(args) == 0:
        raise TriggerError("A trigger is expected for options.")

    if args[0] not in trigger_by_processor:
        raise TriggerError(f"Option {args[0]} not registered as a trigger.")

    processor = trigger_by_processor[args[0]]
    try:
        ret_args = processor.bind(args)
    except Exception as e:
        self._exceptions.append(e)
        return []

    try:
        self._value = processor.process(self._value)
    except Exception as e:
        self._exceptions.append(e)

    return ret_args

to_argument

to_argument()
Source code in thermite/parameters/base.py
def to_argument(self) -> "Argument":
    # to convert it to an argument, we need a processor of subclass
    # ConvertTriggerProcessor
    type_converter = None
    for proc in self.processors:
        if isinstance(proc, MultiConvertTriggerProcessor):
            type_converter = proc.to_convert_trigger_processor().type_converter
            res_type = proc.res_type
            break
        elif isinstance(proc, ConvertTriggerProcessor):
            type_converter = proc.type_converter  # type: ignore
            res_type = proc.res_type
            break

    if type_converter is None:
        raise Exception("Can't convert option to argument")

    res = Argument(
        name=self.name,
        python_kind=self.python_kind,
        cli_kind=self.cli_kind,
        descr=self.descr,
        default_value=self.default_value,
        annot=self.annot,
        res_type=res_type,
        type_converter=type_converter,
    )

    return res

BasicCLIArgConverter

Bases: CLIArgConverterBase

supported_type instance-attribute

supported_type: Type

conv_func instance-attribute

conv_func: Callable

target_type instance-attribute

target_type: Type

factory classmethod

factory(target_type, store, supported_type, conv_func=None)
Source code in thermite/type_converters.py
@classmethod
def factory(
    cls,
    target_type: Type,
    store: "CLIArgConverterStore",
    supported_type: Type,
    conv_func: Optional[Callable] = None,
):
    del store
    if conv_func is None:
        conv_func = supported_type
    return cls(
        supported_type=supported_type, target_type=target_type, conv_func=conv_func
    )

process_function_to_obj_signature

process_function_to_obj_signature(func)
Source code in thermite/signatures.py
def process_function_to_obj_signature(func: Callable) -> ObjSignature:
    descriptions = extract_descriptions(func)
    func_sig = inspect.signature(func)

    return ObjSignature(
        params=create_params_sig_dict(func_sig.parameters, descriptions.args_doc_dict),
        return_annot=func_sig.return_annotation,
        short_descr=descriptions.short_descr,
        long_descr=descriptions.long_descr,
    )

process_class_to_obj_signature

process_class_to_obj_signature(klass)
Source code in thermite/signatures.py
def process_class_to_obj_signature(klass: Type) -> ObjSignature:
    descriptions = extract_descriptions(klass)
    if klass.__init__ != object.__init__:
        init_sig = inspect.signature(klass.__init__)
        return ObjSignature(
            params=create_params_sig_dict(
                init_sig.parameters, descriptions.args_doc_dict
            ),
            return_annot=klass,
            short_descr=descriptions.short_descr,
            long_descr=descriptions.long_descr,
        )

    else:
        return process_instance_to_obj_signature(klass())

process_instance_to_obj_signature

process_instance_to_obj_signature(obj)
Source code in thermite/signatures.py
def process_instance_to_obj_signature(obj: Any) -> ObjSignature:
    # get the documentation
    klass_doc = inspect.getdoc(obj.__class__)

    if klass_doc is not None:
        klass_doc_parsed = parse(klass_doc)
        short_descr = klass_doc_parsed.short_description
        long_descr = klass_doc_parsed.long_description
    else:
        short_descr = None
        long_descr = None

    # as it is an instance, there are no things to call
    return ObjSignature(
        params={},
        return_annot=obj.__class__,
        short_descr=short_descr,
        long_descr=long_descr,
    )