Add options_manager, rework bottom_bar to only render through closures, easy add of options to bottom bar

This commit is contained in:
Roland Thomas Jr 2025-04-20 16:28:24 -04:00
parent ebcd4b43c6
commit 6c72e22415
11 changed files with 310 additions and 148 deletions

0
falyx/.pytyped Normal file
View File

View File

@ -38,4 +38,5 @@ def build_falyx() -> Falyx:
if __name__ == "__main__":
logging.basicConfig(level=logging.WARNING)
falyx = build_falyx()
asyncio.run(falyx.cli())
asyncio.run(falyx.run())

View File

@ -471,8 +471,6 @@ class ProcessAction(BaseAction):
try:
import pickle
pickle.dumps(obj)
print("YES")
return True
except (pickle.PicklingError, TypeError):
print("NO")
return False

View File

@ -1,53 +1,92 @@
"""bottom_bar.py"""
from typing import Any, Callable, Optional
from typing import Any, Callable
from prompt_toolkit.formatted_text import HTML, merge_formatted_text
from prompt_toolkit.key_binding import KeyBindings
from rich.console import Console
from falyx.options_manager import OptionsManager
from falyx.themes.colors import OneColors
from falyx.utils import CaseInsensitiveDict
class BottomBar:
"""Bottom Bar class for displaying a bottom bar in the terminal."""
def __init__(self, columns: int = 3, key_bindings: KeyBindings | None = None):
"""
Bottom Bar class for displaying a bottom bar in the terminal.
Args:
columns (int): Number of columns in the bottom bar.
key_bindings (KeyBindings, optional): Key bindings for the bottom bar.
key_validator (Callable[[str], bool], optional): Function to validate toggle keys.
Must return True if key is available, otherwise False.
"""
def __init__(
self,
columns: int = 3,
key_bindings: KeyBindings | None = None,
key_validator: Callable[[str], bool] | None = None,
) -> None:
self.columns = columns
self.console = Console()
self._items: list[Callable[[], HTML]] = []
self._named_items: dict[str, Callable[[], HTML]] = {}
self._states: dict[str, Any] = CaseInsensitiveDict()
self.toggles: list[str] = []
self._value_getters: dict[str, Callable[[], Any]] = CaseInsensitiveDict()
self.toggle_keys: list[str] = []
self.key_bindings = key_bindings or KeyBindings()
self.key_validator = key_validator
def get_space(self) -> int:
@staticmethod
def default_render(label: str, value: Any, fg: str, bg: str, width: int) -> HTML:
return HTML(f"<style fg='{fg}' bg='{bg}'>{label}: {value:^{width}}</style>")
@property
def space(self) -> int:
return self.console.width // self.columns
def add_custom(
self,
name: str,
render_fn: Callable[[], HTML]
) -> None:
"""Add a custom render function to the bottom bar."""
if not callable(render_fn):
raise ValueError("`render_fn` must be callable")
self._add_named(name, render_fn)
def add_static(
self, name: str, text: str, fg: str = OneColors.BLACK, bg: str = OneColors.WHITE
self,
name: str,
text: str,
fg: str = OneColors.BLACK,
bg: str = OneColors.WHITE,
) -> None:
def render():
return HTML(
f"<style fg='{fg}' bg='{bg}'>{text:^{self.get_space()}}</style>"
f"<style fg='{fg}' bg='{bg}'>{text:^{self.space}}</style>"
)
self._add_named(name, render)
def add_counter(
def add_value_tracker(
self,
name: str,
label: str,
current: int,
get_value: Callable[[], Any],
fg: str = OneColors.BLACK,
bg: str = OneColors.WHITE,
) -> None:
self._states[name] = (label, current)
if not callable(get_value):
raise ValueError("`get_value` must be a callable returning any value")
self._value_getters[name] = get_value
def render():
label_, current_ = self._states[name]
text = f"{label_}: {current_}"
get_value_ = self._value_getters[name]
current_ = get_value_()
text = f"{label}: {current_}"
return HTML(
f"<style fg='{fg}' bg='{bg}'>{text:^{self.get_space()}}</style>"
f"<style fg='{fg}' bg='{bg}'>{text:^{self.space}}</style>"
)
self._add_named(name, render)
@ -56,23 +95,26 @@ class BottomBar:
self,
name: str,
label: str,
current: int,
get_current: Callable[[], int],
total: int,
fg: str = OneColors.BLACK,
bg: str = OneColors.WHITE,
) -> None:
self._states[name] = (label, current, total)
if not callable(get_current):
raise ValueError("`get_current` must be a callable returning int")
if current > total:
raise ValueError(
f"Current value {current} is greater than total value {total}"
)
self._value_getters[name] = get_current
def render():
label_, current_, text_ = self._states[name]
text = f"{label_}: {current_}/{text_}"
get_current_ = self._value_getters[name]
current_value = get_current_()
if current_value > total:
raise ValueError(
f"Current value {current_value} is greater than total value {total}"
)
text = f"{label}: {current_value}/{total}"
return HTML(
f"<style fg='{fg}' bg='{bg}'>{text:^{self.get_space()}}</style>"
f"<style fg='{fg}' bg='{bg}'>{text:^{self.space}}</style>"
)
self._add_named(name, render)
@ -81,24 +123,31 @@ class BottomBar:
self,
key: str,
label: str,
state: bool,
get_state: Callable[[], bool],
toggle_state: Callable[[], None],
fg: str = OneColors.BLACK,
bg_on: str = OneColors.GREEN,
bg_off: str = OneColors.DARK_RED,
) -> None:
if not callable(get_state):
raise ValueError("`get_state` must be a callable returning bool")
if not callable(toggle_state):
raise ValueError("`toggle_state` must be a callable")
key = key.upper()
if key in self.toggles:
if key in self.toggle_keys:
raise ValueError(f"Key {key} is already used as a toggle")
self._states[key] = (label, state)
self.toggles.append(key)
if self.key_validator and not self.key_validator(key):
raise ValueError(f"Key '{key}' conflicts with existing command, toggle, or reserved key.")
self._value_getters[key] = get_state
self.toggle_keys.append(key)
def render():
label_, state_ = self._states[key]
color = bg_on if state_ else bg_off
status = "ON" if state_ else "OFF"
text = f"({key.upper()}) {label_}: {status}"
get_state_ = self._value_getters[key]
color = bg_on if get_state_() else bg_off
status = "ON" if get_state_() else "OFF"
text = f"({key.upper()}) {label}: {status}"
return HTML(
f"<style bg='{color}' fg='{fg}'>{text:^{self.get_space()}}</style>"
f"<style bg='{color}' fg='{fg}'>{text:^{self.space}}</style>"
)
self._add_named(key, render)
@ -106,43 +155,43 @@ class BottomBar:
for k in (key.upper(), key.lower()):
@self.key_bindings.add(k)
def _(event, key=k):
self.toggle_state(key)
def _(event):
toggle_state()
def toggle_state(self, name: str) -> bool:
label, state = self._states.get(name, (None, False))
new_state = not state
self.update_toggle(name, new_state)
return new_state
def update_toggle(self, name: str, state: bool) -> None:
if name in self._states:
label, _ = self._states[name]
self._states[name] = (label, state)
def increment_counter(self, name: str) -> None:
if name in self._states:
label, current = self._states[name]
self._states[name] = (label, current + 1)
def increment_total_counter(self, name: str) -> None:
if name in self._states:
label, current, total = self._states[name]
if current < total:
self._states[name] = (label, current + 1, total)
def update_counter(
self, name: str, current: Optional[int] = None, total: Optional[int] = None
def add_toggle_from_option(
self,
key: str,
label: str,
options: OptionsManager,
option_name: str,
namespace_name: str = "cli_args",
fg: str = OneColors.BLACK,
bg_on: str = OneColors.GREEN,
bg_off: str = OneColors.DARK_RED,
) -> None:
if name in self._states:
label, c, t = self._states[name]
self._states[name] = (
label,
current if current is not None else c,
total if total is not None else t,
self.add_toggle(
key=key,
label=label,
get_state=options.get_value_getter(option_name, namespace_name),
toggle_state=options.get_toggle_function(option_name, namespace_name),
fg=fg,
bg_on=bg_on,
bg_off=bg_off,
)
@property
def values(self) -> dict[str, Any]:
"""Return the current computed values for all registered items."""
return {label: getter() for label, getter in self._value_getters.items()}
def get_value(self, name: str) -> Any:
if name not in self._value_getters:
raise ValueError(f"No value getter registered under name: '{name}'")
return self._value_getters[name]()
def _add_named(self, name: str, render_fn: Callable[[], HTML]) -> None:
if name in self._named_items:
raise ValueError(f"Bottom bar item '{name}' already exists")
self._named_items[name] = render_fn
self._items = list(self._named_items.values())

View File

@ -39,7 +39,8 @@ from falyx.exceptions import (CommandAlreadyExistsError, FalyxError,
InvalidActionError, NotAFalyxError)
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import Hook, HookManager, HookType
from falyx.parsers import FalyxParsers, get_arg_parsers
from falyx.options_manager import OptionsManager
from falyx.parsers import get_arg_parsers
from falyx.retry import RetryPolicy
from falyx.themes.colors import OneColors, get_nord_theme
from falyx.utils import CaseInsensitiveDict, async_confirm, chunks, logger
@ -78,7 +79,7 @@ class Falyx:
title: str | Markdown = "Menu",
prompt: str | AnyFormattedText = "> ",
columns: int = 3,
bottom_bar: BottomBar | str | Callable[[], None] | None = None,
bottom_bar: BottomBar | str | Callable[[], Any] | None = None,
welcome_message: str | Markdown | dict[str, Any] = "",
exit_message: str | Markdown | dict[str, Any] = "",
key_bindings: KeyBindings | None = None,
@ -88,6 +89,7 @@ class Falyx:
never_confirm: bool = False,
always_confirm: bool = False,
cli_args: Namespace | None = None,
options: OptionsManager | None = None,
custom_table: Callable[["Falyx"], Table] | Table | None = None,
) -> None:
"""Initializes the Falyx object."""
@ -104,12 +106,35 @@ class Falyx:
self.hooks: HookManager = HookManager()
self.last_run_command: Command | None = None
self.key_bindings: KeyBindings = key_bindings or KeyBindings()
self.bottom_bar: BottomBar | str | Callable[[], None] = bottom_bar or BottomBar(columns=columns, key_bindings=self.key_bindings)
self.bottom_bar: BottomBar | str | Callable[[], None] = bottom_bar
self.confirm_on_error: bool = confirm_on_error
self._never_confirm: bool = never_confirm
self._always_confirm: bool = always_confirm
self.cli_args: Namespace | None = cli_args
self.custom_table: Callable[["Falyx"], Table] | Table | None = custom_table
self.set_options(cli_args, options)
def set_options(
self,
cli_args: Namespace | None,
options: OptionsManager | None = None,
) -> None:
"""Checks if the options are set correctly."""
self.options: OptionsManager = options or OptionsManager()
if not cli_args and not options:
return
if options and not cli_args:
raise FalyxError("Options are set, but CLI arguments are not.")
if options is None:
self.options.from_namespace(cli_args, "cli_args")
if not isinstance(self.options, OptionsManager):
raise FalyxError("Options must be an instance of OptionsManager.")
if not isinstance(self.cli_args, Namespace):
raise FalyxError("CLI arguments must be a Namespace object.")
@property
def _name_map(self) -> dict[str, Command]:
@ -248,8 +273,8 @@ class Falyx:
keys.add(cmd.key.upper())
keys.update({alias.upper() for alias in cmd.aliases})
if isinstance(self.bottom_bar, BottomBar):
toggle_keys = {key.upper() for key in self.bottom_bar.toggles}
if isinstance(self._bottom_bar, BottomBar):
toggle_keys = {key.upper() for key in self._bottom_bar.toggle_keys}
else:
toggle_keys = set()
@ -277,57 +302,47 @@ class Falyx:
if hasattr(self, "session"):
del self.session
def add_toggle(self, key: str, label: str, state: bool) -> None:
"""Adds a toggle to the bottom bar."""
assert isinstance(self.bottom_bar, BottomBar), "Bottom bar must be an instance of BottomBar."
self.bottom_bar.add_toggle(key, label, state)
self._invalidate_session_cache()
def add_counter(self, name: str, label: str, current: int) -> None:
"""Adds a counter to the bottom bar."""
assert isinstance(self.bottom_bar, BottomBar), "Bottom bar must be an instance of BottomBar."
self.bottom_bar.add_counter(name, label, current)
self._invalidate_session_cache()
def add_total_counter(self, name: str, label: str, current: int, total: int) -> None:
"""Adds a counter to the bottom bar."""
assert isinstance(self.bottom_bar, BottomBar), "Bottom bar must be an instance of BottomBar."
self.bottom_bar.add_total_counter(name, label, current, total)
self._invalidate_session_cache()
def add_static(self, name: str, text: str) -> None:
"""Adds a static element to the bottom bar."""
assert isinstance(self.bottom_bar, BottomBar), "Bottom bar must be an instance of BottomBar."
self.bottom_bar.add_static(name, text)
self._invalidate_session_cache
def get_toggle_state(self, key: str) -> bool | None:
assert isinstance(self.bottom_bar, BottomBar), "Bottom bar must be an instance of BottomBar."
if key.upper() in self.bottom_bar._states:
"""Returns the state of a toggle."""
return self.bottom_bar._states[key.upper()][1]
return None
def add_help_command(self):
"""Adds a help command to the menu if it doesn't already exist."""
if not self.help_command:
self.help_command = self._get_help_command()
self._invalidate_session_cache()
def add_history_command(self):
"""Adds a history command to the menu if it doesn't already exist."""
if not self.history_command:
self.history_command = self._get_history_command()
@property
def bottom_bar(self) -> BottomBar | str | Callable[[], Any] | None:
"""Returns the bottom bar for the menu."""
return self._bottom_bar
@bottom_bar.setter
def bottom_bar(self, bottom_bar: BottomBar | str | Callable[[], Any] | None) -> None:
"""Sets the bottom bar for the menu."""
if bottom_bar is None:
self._bottom_bar = BottomBar(self.columns, self.key_bindings, key_validator=self.is_key_available)
elif isinstance(bottom_bar, BottomBar):
bottom_bar.key_validator = self.is_key_available
bottom_bar.key_bindings = self.key_bindings
self._bottom_bar = bottom_bar
elif (isinstance(bottom_bar, str) or
callable(bottom_bar)):
self._bottom_bar = bottom_bar
else:
raise FalyxError("Bottom bar must be a string, callable, or BottomBar instance.")
self._invalidate_session_cache()
def _get_bottom_bar(self) -> Callable[[], Any] | str | None:
def _get_bottom_bar_render(self) -> Callable[[], Any] | str | None:
"""Returns the bottom bar for the menu."""
if isinstance(self.bottom_bar, BottomBar) and self.bottom_bar._items:
return self.bottom_bar.render
elif callable(self.bottom_bar):
return self.bottom_bar
elif isinstance(self.bottom_bar, str):
return self.bottom_bar
return self._bottom_bar.render
elif callable(self._bottom_bar):
return self._bottom_bar
elif isinstance(self._bottom_bar, str):
return self._bottom_bar
elif self._bottom_bar is None:
return None
return None
@cached_property
@ -339,7 +354,7 @@ class Falyx:
completer=self._get_completer(),
reserve_space_for_menu=1,
validator=self._get_validator(),
bottom_toolbar=self._get_bottom_bar(),
bottom_toolbar=self._get_bottom_bar_render(),
key_bindings=self.key_bindings,
)
@ -382,10 +397,24 @@ class Falyx:
logger.debug(f"[Command '{key}'] after: {hook_names(command.hooks._hooks[HookType.AFTER])}")
logger.debug(f"[Command '{key}'] on_teardown: {hook_names(command.hooks._hooks[HookType.ON_TEARDOWN])}")
def is_key_available(self, key: str) -> bool:
key = key.upper()
toggles = self._bottom_bar.toggle_keys if isinstance(self._bottom_bar, BottomBar) else []
conflicts = (
key in self.commands,
key == self.exit_command.key.upper(),
self.history_command and key == self.history_command.key.upper(),
self.help_command and key == self.help_command.key.upper(),
key in toggles
)
return not any(conflicts)
def _validate_command_key(self, key: str) -> None:
"""Validates the command key to ensure it is unique."""
key = key.upper()
toggles = self.bottom_bar.toggles if isinstance(self.bottom_bar, BottomBar) else []
toggles = self._bottom_bar.toggle_keys if isinstance(self._bottom_bar, BottomBar) else []
collisions = []
if key in self.commands:
@ -423,7 +452,6 @@ class Falyx:
confirm=confirm,
confirm_message=confirm_message,
)
self._invalidate_session_cache()
def add_submenu(self, key: str, description: str, submenu: "Falyx", color: str = OneColors.CYAN) -> None:
"""Adds a submenu to the menu."""
@ -431,7 +459,6 @@ class Falyx:
raise NotAFalyxError("submenu must be an instance of Falyx.")
self._validate_command_key(key)
self.add_command(key, description, submenu.menu, color=color)
self._invalidate_session_cache()
def add_commands(self, commands: list[dict]) -> None:
"""Adds multiple commands to the menu."""
@ -511,7 +538,6 @@ class Falyx:
command.hooks.register(HookType.ON_TEARDOWN, hook)
self.commands[key] = command
self._invalidate_session_cache()
return command
def get_bottom_row(self) -> list[str]:
@ -755,10 +781,10 @@ class Falyx:
if self.exit_message:
self.print_message(self.exit_message)
async def run(self, parsers: FalyxParsers | None = None) -> None:
async def run(self) -> None:
"""Run Falyx CLI with structured subcommands."""
parsers = parsers or get_arg_parsers()
self.cli_args = parsers.root.parse_args()
if not self.cli_args:
self.cli_args = get_arg_parsers().root.parse_args()
if self.cli_args.verbose:
logging.getLogger("falyx").setLevel(logging.DEBUG)

View File

@ -2,8 +2,9 @@ import asyncio
import logging
from rich.markdown import Markdown
from falyx import Action, Falyx, HookType
from falyx.hooks import log_before, log_success, log_error, log_after
from falyx import Action, Falyx
from falyx.hook_manager import HookType
from falyx.debug import log_before, log_success, log_error, log_after
from falyx.themes.colors import OneColors
from falyx.utils import setup_logging
@ -76,7 +77,7 @@ def main():
spinner=True,
)
asyncio.run(menu.cli())
asyncio.run(menu.run())
if __name__ == "__main__":

72
falyx/options_manager.py Normal file
View File

@ -0,0 +1,72 @@
"""options_manager.py"""
from argparse import Namespace
from collections import defaultdict
from typing import Any, Callable
from falyx.utils import logger
class OptionsManager:
def __init__(self, namespaces: list[tuple[str, Namespace]] = None) -> None:
self.options = defaultdict(lambda: Namespace())
if namespaces:
for namespace_name, namespace in namespaces:
self.from_namespace(namespace, namespace_name)
def from_namespace(
self, namespace: Namespace, namespace_name: str = "cli_args"
) -> None:
self.options[namespace_name] = namespace
def get(
self, option_name: str, default: Any = None, namespace_name: str = "cli_args"
) -> Any:
"""Get the value of an option."""
return getattr(self.options[namespace_name], option_name, default)
def set(
self, option_name: str, value: Any, namespace_name: str = "cli_args"
) -> None:
"""Set the value of an option."""
setattr(self.options[namespace_name], option_name, value)
def has_option(self, option_name: str, namespace_name: str = "cli_args") -> bool:
"""Check if an option exists in the namespace."""
return hasattr(self.options[namespace_name], option_name)
def toggle(self, option_name: str, namespace_name: str = "cli_args") -> None:
"""Toggle a boolean option."""
current = self.get(option_name, namespace_name=namespace_name)
if not isinstance(current, bool):
raise TypeError(
f"Cannot toggle non-boolean option: '{option_name}' in '{namespace_name}'"
)
self.set(option_name, not current, namespace_name=namespace_name)
logger.debug(f"Toggled '{option_name}' in '{namespace_name}' to {not current}")
def get_value_getter(
self, option_name: str, namespace_name: str = "cli_args"
) -> Callable[[], Any]:
"""Get the value of an option as a getter function."""
def _getter() -> Any:
return self.get(option_name, namespace_name=namespace_name)
return _getter
def get_toggle_function(
self, option_name: str, namespace_name: str = "cli_args"
) -> Callable[[], None]:
"""Get the toggle function for a boolean option."""
def _toggle() -> None:
self.toggle(option_name, namespace_name=namespace_name)
return _toggle
def get_namespace_dict(self, namespace_name: str) -> Namespace:
"""Return all options in a namespace as a dictionary."""
if namespace_name not in self.options:
raise ValueError(f"Namespace '{namespace_name}' not found.")
return vars(self.options[namespace_name])

View File

@ -1,8 +1,9 @@
"""parsers.py
This module contains the argument parsers used for the Falyx CLI.
"""
from argparse import ArgumentParser, HelpFormatter, Namespace
from dataclasses import asdict, dataclass
from argparse import ArgumentParser
from typing import Any, Sequence
@dataclass
@ -15,6 +16,10 @@ class FalyxParsers:
list: ArgumentParser
version: ArgumentParser
def parse_args(self, args: Sequence[str] | None = None) -> Namespace:
"""Parse the command line arguments."""
return self.root.parse_args(args)
def as_dict(self) -> dict[str, ArgumentParser]:
"""Convert the FalyxParsers instance to a dictionary."""
return asdict(self)
@ -24,9 +29,37 @@ class FalyxParsers:
return self.as_dict().get(name)
def get_arg_parsers() -> FalyxParsers:
def get_arg_parsers(
prog: str |None = "falyx",
usage: str | None = None,
description: str | None = "Falyx CLI - Run structured async command workflows.",
epilog: str | None = None,
parents: Sequence[ArgumentParser] = [],
formatter_class: HelpFormatter = HelpFormatter,
prefix_chars: str = "-",
fromfile_prefix_chars: str | None = None,
argument_default: Any = None,
conflict_handler: str = "error",
add_help: bool = True,
allow_abbrev: bool = True,
exit_on_error: bool = True,
) -> FalyxParsers:
"""Returns the argument parser for the CLI."""
parser = ArgumentParser(prog="falyx", description="Falyx CLI - Run structured async command workflows.")
parser = ArgumentParser(
prog=prog,
usage=usage,
description=description,
epilog=epilog,
parents=parents,
formatter_class=formatter_class,
prefix_chars=prefix_chars,
fromfile_prefix_chars=fromfile_prefix_chars,
argument_default=argument_default,
conflict_handler=conflict_handler,
add_help=add_help,
allow_abbrev=allow_abbrev,
exit_on_error=exit_on_error,
)
parser.add_argument("-v", "--verbose", action="store_true", help="Enable debug logging for Falyx.")
parser.add_argument("--debug-hooks", action="store_true", help="Enable default lifecycle debug logging")
parser.add_argument("--version", action="store_true", help="Show Falyx version")

View File

@ -1 +1 @@
__version__ = "0.1.0"
__version__ = "0.1.5"

View File

@ -1,6 +1,6 @@
[tool.poetry]
name = "falyx"
version = "0.1.4"
version = "0.1.5"
description = "Reliable and introspectable async CLI action framework."
authors = ["Roland Thomas Jr <roland@rtj.dev>"]
license = "MIT"
@ -12,12 +12,12 @@ python = ">=3.10"
prompt_toolkit = "^3.0"
rich = "^13.0"
pydantic = "^2.0"
python-json-logger = "^3.3.0"
[tool.poetry.group.dev.dependencies]
pytest = "^7.0"
pytest-asyncio = "^0.20"
ruff = "^0.3"
python-json-logger = "^3.3.0"
[tool.poetry.scripts]
falyx = "falyx.cli.main:main"

View File

@ -1,18 +0,0 @@
from setuptools import setup, find_packages
setup(
name="falyx",
version="0.0.1",
description="Reserved package name for future CLI framework.",
long_description=open("README.md").read(),
long_description_content_type="text/markdown",
author="Roland Thomas Jr",
author_email="roland@rtj.dev",
packages=find_packages(),
python_requires=">=3.10",
classifiers=[
"Programming Language :: Python :: 3",
"License :: OSI Approved :: MIT License",
"Development Status :: 1 - Planning",
],
)