roland/falyx
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: add recursive namespace routing and standalone runner polish

Browse Source 
- introduce namespace-aware routing with RootParseResult, RouteResult, and InvocationContext
- register submenus as FalyxNamespace entries and resolve them through _entry_map
- refactor FalyxParser to parse only root options and leave recursive routing to Falyx
- add prepare_route, resolve_route, and route dispatch flow to Falyx
- update validator and completer to understand namespace entries and route results
- unify help/TLDR rendering APIs and add custom_tldr support on Command
- tighten Command.resolve_args error handling and parser type validation
- improve CommandRunner dependency validation and argv handling
- add BottomBar.has_items and improve wrapped executor error messages
- add tests for execution options, resolve_args, command runner, and route-aware validation
This commit is contained in:
Roland Thomas Jr
2026-04-11 11:57:03 -04:00
parent 5d8f3aa603
commit 30cb8b97b5
26 changed files with 1658 additions and 493 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
falyx/bottom_bar.py
View File

@@ -72,6 +72,11 @@ class BottomBar:
self.toggle_keys: list[str] = []
self.key_bindings = key_bindings or KeyBindings()
@property
def has_items(self) -> bool:
"""Check if the bottom bar has any registered items."""
return bool(self._named_items)
@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>")

112
falyx/command.py
View File

@@ -53,13 +53,12 @@ from falyx.action.base_action import BaseAction
from falyx.console import console
from falyx.context import ExecutionContext
from falyx.debug import register_debug_hooks
from falyx.exceptions import NotAFalyxError
from falyx.exceptions import CommandArgumentError, NotAFalyxError
from falyx.execution_option import ExecutionOption
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import HookManager, HookType
from falyx.hooks import spinner_before_hook, spinner_teardown_hook
from falyx.logger import logger
from falyx.mode import FalyxMode
from falyx.options_manager import OptionsManager
from falyx.parser.command_argument_parser import CommandArgumentParser
from falyx.parser.signature import infer_args_from_func
@@ -149,6 +148,8 @@ class Command(BaseModel):
Override parser logic entirely.
custom_help (Callable[[], str | None] | None):
Override help rendering.
custom_tldr (Callable[[], str | None] | None):
Override TLDR rendering.
auto_args (bool): Auto-generate arguments from action signature.
arg_metadata (dict[str, Any], optional): Metadata for arguments.
simple_help_signature (bool): Use simplified help formatting.
@@ -199,7 +200,8 @@ class Command(BaseModel):
arguments: list[dict[str, Any]] = Field(default_factory=list)
argument_config: Callable[[CommandArgumentParser], None] | None = None
custom_parser: ArgParserProtocol | None = None
custom_help: Callable[[], str | None] | None = None
custom_help: Callable[[], None] | None = None
custom_tldr: Callable[[], None] | None = None
auto_args: bool = True
arg_metadata: dict[str, str | dict[str, Any]] = Field(default_factory=dict)
simple_help_signature: bool = False
@@ -236,7 +238,7 @@ class Command(BaseModel):
- Handles help/preview signals raised during parsing.
Args:
args (list[str] | None): CLI-style argument tokens.
args (list[str] | str | None): CLI-style argument tokens or a single string.
from_validate (bool): Whether parsing is occurring in validation mode
(e.g. prompt_toolkit validator). When True, may suppress eager
resolution or defer certain errors.
@@ -257,35 +259,38 @@ class Command(BaseModel):
- This method is the canonical boundary between CLI parsing and
execution semantics.
"""
if callable(self.custom_parser):
if self.custom_parser is not None:
if not callable(self.custom_parser):
raise NotAFalyxError(
"custom_parser must be a callable that implements ArgParserProtocol."
)
if isinstance(raw_args, str):
try:
raw_args = shlex.split(raw_args)
except ValueError:
logger.warning(
"[Command:%s] Failed to split arguments: %s",
self.key,
raw_args,
)
return ((), {}, {})
except ValueError as error:
raise CommandArgumentError(
f"[{self.key}] Failed to parse arguments: {error}"
) from error
return self.custom_parser(raw_args)
if isinstance(raw_args, str):
try:
raw_args = shlex.split(raw_args)
except ValueError:
logger.warning(
"[Command:%s] Failed to split arguments: %s",
self.key,
raw_args,
)
return ((), {}, {})
if not isinstance(self.arg_parser, CommandArgumentParser):
logger.warning(
"[Command:%s] No argument parser configured, using default parsing.",
self.key,
except ValueError as error:
raise CommandArgumentError(
f"[{self.key}] Failed to parse arguments: {error}"
) from error
if self.arg_parser is None:
raise NotAFalyxError(
"Command has no parser configured. "
"Provide a custom_parser or CommandArgumentParser."
)
return ((), {}, {})
if not isinstance(self.arg_parser, CommandArgumentParser):
raise NotAFalyxError(
"arg_parser must be an instance of CommandArgumentParser"
)
return await self.arg_parser.parse_args_split(
raw_args, from_validate=from_validate
)
@@ -506,19 +511,15 @@ class Command(BaseModel):
tuple:
- str: Usage string (e.g. "falyx D | deploy [--help] region")
- str: Command description
- str | None: Optional tag/category label
- str: Optional tag/category label
Notes:
- This is the primary interface used by help menus, CLI help output,
and command listings.
- Formatting may vary depending on CLI vs menu mode.
"""
is_cli_mode = self.options_manager.get("mode") != FalyxMode.MENU
program = f"{self.program} " if is_cli_mode else ""
if self.arg_parser and not self.simple_help_signature:
usage = f"[{self.style}]{program}[/]{self.arg_parser.get_usage()}"
usage = self.arg_parser.get_usage()
description = f"[dim]{self.help_text or self.description}[/dim]"
if self.tags:
tags = f"[dim]Tags: {', '.join(self.tags)}[/dim]"
@@ -531,7 +532,7 @@ class Command(BaseModel):
+ [f"[{self.style}]{alias}[/{self.style}]" for alias in self.aliases]
)
return (
f"[{self.style}]{program}[/]{command_keys}",
f"{command_keys}",
f"[dim]{self.help_text or self.description}[/dim]",
"",
)
@@ -552,6 +553,18 @@ class Command(BaseModel):
return True
return False
def render_tldr(self) -> bool:
"""Display the TLDR message for the command."""
if callable(self.custom_tldr):
output = self.custom_tldr()
if output:
console.print(output)
return True
if isinstance(self.arg_parser, CommandArgumentParser):
self.arg_parser.render_tldr()
return True
return False
async def preview(self) -> None:
label = f"[{OneColors.GREEN_b}]Command:[/] '{self.key}' — {self.description}"
@@ -623,6 +636,7 @@ class Command(BaseModel):
execution_options: list[ExecutionOption | str] | None = None,
custom_parser: ArgParserProtocol | None = None,
custom_help: Callable[[], str | None] | None = None,
custom_tldr: Callable[[], str | None] | None = None,
auto_args: bool = True,
arg_metadata: dict[str, str | dict[str, Any]] | None = None,
simple_help_signature: bool = False,
@@ -697,6 +711,8 @@ class Command(BaseModel):
implementation that overrides normal parser behavior.
custom_help (Callable[[], str | None] | None): Optional custom help
renderer.
custom_tldr (Callable[[], str | None] | None): Optional custom TLDR
renderer.
auto_args (bool): Whether to infer arguments automatically from the action
signature when explicit definitions are not provided.
arg_metadata (dict[str, str | dict[str, Any]] | None): Optional metadata
@@ -722,12 +738,23 @@ class Command(BaseModel):
- This method is the canonical command-construction path used by higher-
level APIs such as `Falyx.add_command()` and `CommandRunner.build()`.
"""
if arg_parser:
if not isinstance(arg_parser, CommandArgumentParser):
raise NotAFalyxError(
"arg_parser must be an instance of CommandArgumentParser."
)
arg_parser = arg_parser
if arg_parser and not isinstance(arg_parser, CommandArgumentParser):
raise NotAFalyxError(
"arg_parser must be an instance of CommandArgumentParser."
)
arg_parser = arg_parser
if options_manager and not isinstance(options_manager, OptionsManager):
raise NotAFalyxError("options_manager must be an instance of OptionsManager.")
options_manager = options_manager or OptionsManager()
if hooks and not isinstance(hooks, HookManager):
raise NotAFalyxError("hooks must be an instance of HookManager.")
hooks = hooks or HookManager()
if retry_policy and not isinstance(retry_policy, RetryPolicy):
raise NotAFalyxError("retry_policy must be an instance of RetryPolicy.")
retry_policy = retry_policy or RetryPolicy()
if execution_options:
parsed_execution_options = frozenset(
@@ -737,8 +764,6 @@ class Command(BaseModel):
else:
parsed_execution_options = frozenset()
options_manager = options_manager or OptionsManager()
command = Command(
key=key,
description=description,
@@ -760,9 +785,10 @@ class Command(BaseModel):
spinner_speed=spinner_speed,
tags=tags if tags else [],
logging_hooks=logging_hooks,
hooks=hooks,
retry=retry,
retry_all=retry_all,
retry_policy=retry_policy or RetryPolicy(),
retry_policy=retry_policy,
options_manager=options_manager,
arg_parser=arg_parser,
execution_options=parsed_execution_options,
@@ -770,6 +796,7 @@ class Command(BaseModel):
argument_config=argument_config,
custom_parser=custom_parser,
custom_help=custom_help,
custom_tldr=custom_tldr,
auto_args=auto_args,
arg_metadata=arg_metadata or {},
simple_help_signature=simple_help_signature,
@@ -777,11 +804,6 @@ class Command(BaseModel):
program=program,
)
if hooks:
if not isinstance(hooks, HookManager):
raise NotAFalyxError("hooks must be an instance of HookManager.")
command.hooks = hooks
for hook in before_hooks or []:
command.hooks.register(HookType.BEFORE, hook)
for hook in success_hooks or []:

4
falyx/command_executor.py
View File

@@ -320,7 +320,9 @@ class CommandExecutor:
await self.hooks.trigger(HookType.ON_ERROR, context)
await self._handle_action_error(command, error)
if wrap_errors:
raise FalyxError(f"[execute] '{command.description}' failed.") from error
raise FalyxError(
f"[execute] '{command.description}' failed: {error}"
) from error
if raise_on_error:
raise error
finally:

58
falyx/command_runner.py
View File

@@ -87,7 +87,7 @@ class CommandRunner:
command (Command): The command executed by this runner.
options (OptionsManager): Shared options manager used by the command,
parser, and executor.
hooks (HookManager): Executor-level hooks used during execution.
runner_hooks (HookManager): Executor-level hooks used during execution.
console (Console): Rich console used for user-facing output.
executor (CommandExecutor): Shared execution engine used to run the
bound command.
@@ -98,7 +98,7 @@ class CommandRunner:
command: Command,
*,
options: OptionsManager | None = None,
hooks: HookManager | None = None,
runner_hooks: HookManager | None = None,
console: Console | None = None,
) -> None:
"""Initialize a `CommandRunner` for a single command.
@@ -111,28 +111,52 @@ class CommandRunner:
command (Command): The command to execute.
options (OptionsManager | None): Optional shared options manager. If
omitted, a new `OptionsManager` is created.
hooks (HookManager | None): Optional executor-level hook manager. If
runner_hooks (HookManager | None): Optional executor-level hook manager. If
omitted, a new `HookManager` is created.
console (Console | None): Optional Rich console for output. If omitted,
the default Falyx console is used.
"""
self.command = command
self.options = options or OptionsManager()
self.hooks = hooks or HookManager()
self.console = console or falyx_console
self.options = self._get_options(options)
self.runner_hooks = self._get_hooks(runner_hooks)
self.console = self._get_console(console)
self.command.options_manager = self.options
if isinstance(self.command.arg_parser, CommandArgumentParser):
self.command.arg_parser.set_options_manager(self.options)
self.executor = CommandExecutor(
options=self.options,
hooks=self.hooks,
hooks=self.runner_hooks,
console=self.console,
)
self.options.from_mapping(values={}, namespace_name="execution")
def _get_console(self, console) -> Console:
if console is None:
return falyx_console
elif isinstance(console, Console):
return console
else:
raise NotAFalyxError("console must be an instance of rich.Console or None.")
def _get_options(self, options) -> OptionsManager:
if options is None:
return OptionsManager()
elif isinstance(options, OptionsManager):
return options
else:
raise NotAFalyxError("options must be an instance of OptionsManager or None.")
def _get_hooks(self, hooks) -> HookManager:
if hooks is None:
return HookManager()
elif isinstance(hooks, HookManager):
return hooks
else:
raise NotAFalyxError("hooks must be an instance of HookManager or None.")
async def run(
self,
argv: list[str] | None = None,
argv: list[str] | str | None = None,
raise_on_error: bool = True,
wrap_errors: bool = False,
summary_last_result: bool = False,
@@ -145,8 +169,9 @@ class CommandRunner:
then delegates execution to the internal `CommandExecutor`.
Args:
argv (list[str] | None): Optional argv-style argument tokens. If
omitted, `sys.argv[1:]` is used.
argv (list[str] | str | None): Optional argv-style argument tokens or
string (uses `shlex.split()` if a string is provided). If omitted,
`sys.argv[1:]` is used.
Returns:
Any: The result returned by the bound command.
@@ -176,7 +201,7 @@ class CommandRunner:
async def cli(
self,
argv: list[str] | None = None,
argv: list[str] | str | None = None,
summary_last_result: bool = False,
) -> Any:
"""Run the bound command as a shell-oriented CLI entrypoint.
@@ -197,8 +222,9 @@ class CommandRunner:
- Exits with status code `130` for quit/interrupt-style termination
Args:
argv (list[str] | None): Optional argv-style argument tokens. If omitted,
`sys.argv[1:]` is used by `run()`.
argv (list[str] | str | None): Optional argv-style argument tokens or string
(uses `shlex.split()` if a string is provided). If omitted, `sys.argv[1:]`
is used by `run()`.
summary_last_result (bool): Whether summary output should include the last
recorded result when summary reporting is enabled.
@@ -274,12 +300,14 @@ class CommandRunner:
NotAFalyxError: If `runner_hooks` is provided but is not a
`HookManager` instance.
"""
if not isinstance(command, Command):
raise NotAFalyxError("command must be an instance of Command.")
if runner_hooks and not isinstance(runner_hooks, HookManager):
raise NotAFalyxError("runner_hooks must be an instance of HookManager.")
return cls(
command=command,
options=options,
hooks=runner_hooks,
runner_hooks=runner_hooks,
console=console,
)
@@ -462,6 +490,6 @@ class CommandRunner:
return cls(
command=command,
options=options,
hooks=runner_hooks,
runner_hooks=runner_hooks,
console=console,
)

19
falyx/completer.py
View File

@@ -27,6 +27,8 @@ from typing import TYPE_CHECKING, Iterable
from prompt_toolkit.completion import Completer, Completion
from prompt_toolkit.document import Document
from falyx.namespace import FalyxNamespace
if TYPE_CHECKING:
from falyx import Falyx
@@ -35,7 +37,7 @@ class FalyxCompleter(Completer):
"""Prompt Toolkit completer for Falyx CLI command input.
This completer provides real-time, context-aware suggestions for:
- Command keys and aliases (resolved via Falyx._name_map)
- Command keys and aliases (resolved via Falyx._entry_map)
- CLI argument flags and values for each command
- Suggestions and choices defined in the associated CommandArgumentParser
@@ -89,14 +91,14 @@ class FalyxCompleter(Completer):
def _resolve_command_for_completion(self, token: str):
normalized = token.upper().strip()
name_map = self.falyx._name_map
entry_map = self.falyx._entry_map
if normalized in name_map:
return name_map[normalized]
if normalized in entry_map:
return entry_map[normalized]
matches = []
seen = set()
for key, command in name_map.items():
for key, command in entry_map.items():
if key.startswith(normalized) and id(command) not in seen:
matches.append(command)
seen.add(id(command))
@@ -146,6 +148,13 @@ class FalyxCompleter(Completer):
# Identify command
command_key = tokens[0].upper()
command = self._resolve_command_for_completion(command_key)
if isinstance(command, FalyxNamespace):
completer = command.namespace._get_completer()
for completion in completer.get_completions(
Document(" ".join(tokens[1:])), complete_event
):
yield completion
return
if not command or not command.arg_parser:
return

3
falyx/config.py
View File

@@ -1,6 +1,5 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""
Configuration loader and schema definitions for the Falyx CLI framework.
"""Configuration loader and schema definitions for the Falyx CLI framework.
This module supports config-driven initialization of CLI commands and submenus
from YAML or TOML files. It enables declarative command definitions, auto-imports

28
falyx/context.py
View File

@@ -1,6 +1,5 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""
Execution context management for Falyx CLI actions.
"""Context management for Falyx CLI.
This module defines `ExecutionContext` and `SharedContext`, which are responsible for
capturing per-action and cross-action metadata during CLI workflow execution. These
@@ -26,6 +25,7 @@ from pydantic import BaseModel, ConfigDict, Field
from rich.console import Console
from falyx.console import console
from falyx.mode import FalyxMode
class ExecutionContext(BaseModel):
@@ -285,6 +285,30 @@ class SharedContext(BaseModel):
)
class InvocationContext(BaseModel):
program: str = ""
typed_path: list[str] = Field(default_factory=list)
mode: FalyxMode = FalyxMode.MENU
is_preview: bool = False
@property
def is_cli_mode(self) -> bool:
return self.mode != FalyxMode.MENU
def child(self, token: str) -> InvocationContext:
return InvocationContext(
program=self.program,
typed_path=[*self.typed_path, token],
mode=self.mode,
is_preview=self.is_preview,
)
def display_path(self) -> str:
if self.is_cli_mode:
return " ".join([self.program, *self.typed_path]).strip()
return " ".join(self.typed_path).strip()
if __name__ == "__main__":
import asyncio

13
falyx/exceptions.py
View File

@@ -1,6 +1,5 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""
Defines all custom exception classes used in the Falyx CLI framework.
"""Defines all custom exception classes used in the Falyx CLI framework.
These exceptions provide structured error handling for common failure cases,
including command conflicts, invalid actions or hooks, parser errors, and execution guards
@@ -26,11 +25,11 @@ developer-facing problems that should be caught and reported.
class FalyxError(Exception):
"""Custom exception for the Menu class."""
"""Custom exception for the Falyx class."""
class CommandAlreadyExistsError(FalyxError):
"""Exception raised when an command with the same key already exists in the menu."""
"""Exception raised when an command with the same key already exists in the Falyx instance."""
class InvalidHookError(FalyxError):
@@ -42,7 +41,7 @@ class InvalidActionError(FalyxError):
class NotAFalyxError(FalyxError):
"""Exception raised when the provided submenu is not an instance of Menu."""
"""Exception raised when the provided object is not an instance of a Falyx class."""
class CircuitBreakerOpen(FalyxError):
@@ -54,11 +53,11 @@ class EmptyChainError(FalyxError):
class EmptyGroupError(FalyxError):
"""Exception raised when the chain is empty."""
"""Exception raised when the group is empty."""
class EmptyPoolError(FalyxError):
"""Exception raised when the chain is empty."""
"""Exception raised when the pool is empty."""
class CommandArgumentError(FalyxError):

723
falyx/falyx.py
View File

@@ -57,11 +57,13 @@ from rich.text import Text
from falyx.action.action import Action
from falyx.action.base_action import BaseAction
from falyx.action.signal_action import SignalAction
from falyx.bottom_bar import BottomBar
from falyx.command import Command
from falyx.command_executor import CommandExecutor
from falyx.completer import FalyxCompleter
from falyx.console import console
from falyx.context import InvocationContext
from falyx.debug import log_after, log_before, log_error, log_success
from falyx.exceptions import (
CommandAlreadyExistsError,
@@ -75,14 +77,16 @@ from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import Hook, HookManager, HookType
from falyx.logger import logger
from falyx.mode import FalyxMode
from falyx.namespace import FalyxNamespace
from falyx.options_manager import OptionsManager
from falyx.parser import CommandArgumentParser, FalyxParser, ParseResult
from falyx.parser import CommandArgumentParser, FalyxParser, RootParseResult
from falyx.prompt_utils import rich_text_to_prompt_text
from falyx.protocols import ArgParserProtocol
from falyx.retry import RetryPolicy
from falyx.routing import RouteKind, RouteResult
from falyx.signals import BackSignal, CancelSignal, HelpSignal, QuitSignal
from falyx.themes import OneColors
from falyx.utils import CaseInsensitiveDict, _noop, chunks, ensure_async
from falyx.utils import CaseInsensitiveDict, chunks, ensure_async
from falyx.validators import CommandValidator
from falyx.version import __version__
@@ -209,11 +213,11 @@ class Falyx:
self.columns: int = columns
self.commands: dict[str, Command] = CaseInsensitiveDict()
self.builtins: dict[str, Command] = CaseInsensitiveDict()
self.namespaces: dict[str, FalyxNamespace] = CaseInsensitiveDict()
self.console: Console = console
self.welcome_message: str | Markdown | dict[str, Any] = welcome_message
self.exit_message: str | Markdown | dict[str, Any] = exit_message
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] | None = bottom_bar
self._never_prompt: bool = never_prompt
@@ -247,6 +251,21 @@ class Falyx:
console=self.console,
)
def get_current_invocation_context(self) -> InvocationContext:
"""Returns the current invocation context."""
return InvocationContext(
program=self.program,
typed_path=[],
mode=self.options.get("mode"),
)
def format_invocation_path(
self, program: str, typed_path: list[str], *, cli_mode: bool
) -> str:
if cli_mode:
return " ".join([program, *typed_path]).strip()
return " ".join(typed_path).strip()
@property
def is_cli_mode(self) -> bool:
"""Checks if the current mode is a CLI mode."""
@@ -280,27 +299,30 @@ class Falyx:
if not self.options.get("program_style"):
self.options.set("program_style", self.program_style)
if not self.options.get("invocation_path"):
self.options.set("invocation_path", self.program)
@property
def _name_map(self) -> dict[str, Command]:
def _entry_map(self) -> dict[str, Command | FalyxNamespace]:
"""Builds a mapping of all valid input names to Command objects.
If a collision occurs, logs a warning and keeps the first
registered command.
"""
mapping: dict[str, Command] = {}
mapping: dict[str, Command | FalyxNamespace] = {}
def register(name: str, command: Command):
def register(name: str, entry: Command | FalyxNamespace):
norm = name.upper().strip()
if norm in mapping:
existing = mapping[norm]
if existing is not command:
if existing is not entry:
raise CommandAlreadyExistsError(
f"Identifier '{norm}' is already registered.\n"
f"Existing command: {mapping[norm].key}\n"
f"New command: {command.key}"
f"Existing entry: {mapping[norm].key}\n"
f"New entry: {entry.key}"
)
else:
mapping[norm] = command
mapping[norm] = entry
for special in [self.exit_command, self.history_command]:
if special:
@@ -320,6 +342,11 @@ class Falyx:
for alias in command.aliases:
register(alias, command)
register(command.description, command)
for namespace in self.namespaces.values():
register(namespace.key, namespace)
for alias in namespace.aliases:
register(alias, namespace)
return mapping
def get_title(self) -> str:
@@ -335,7 +362,7 @@ class Falyx:
exit_command = Command(
key="X",
description="Exit",
action=Action("Exit", action=_noop),
action=SignalAction("Exit", QuitSignal()),
aliases=["EXIT", "QUIT"],
style=OneColors.DARK_RED,
simple_help_signature=True,
@@ -454,42 +481,37 @@ class Falyx:
)
return choice(tips)
async def _render_command_tldr(self, key: str | None = None) -> None:
async def _render_command_tldr(self, command: Command) -> None:
"""Renders the TLDR examples for a command, if available."""
if not key and self.help_command:
key = "H"
if not key:
self.console.print("[bold]No command specified for TLDR examples.[/bold]")
if not isinstance(command, Command):
self.console.print(
f"Entry '{command.key}' is not a command.", style=OneColors.DARK_RED
)
return None
_, command, args, kwargs, execution_args = await self.get_command(
key, from_help=True
)
if command and command.arg_parser:
command.arg_parser.render_tldr()
if command.render_tldr():
if self.enable_help_tips:
self.console.print(f"[bold]tip:[/bold] {self.get_tip()}")
elif command and not command.arg_parser:
else:
self.console.print(
f"[bold]No TLDR examples available for '{command.description}'.[/bold]"
)
else:
self.console.print(f"[bold]No command found for '{key}'.[/bold]")
async def _render_command_help(self, key: str) -> None:
async def _render_command_help(self, command: Command, tldr: bool = False) -> None:
"""Renders the detailed help for a command, if available."""
_, command, args, kwargs, execution_args = await self.get_command(
key, from_help=True
)
if command and command.arg_parser:
command.arg_parser.render_help()
if not isinstance(command, Command):
self.console.print(
f"Entry '{command.key}' is not a command.", style=OneColors.DARK_RED
)
return None
if tldr:
await self._render_command_tldr(command)
elif command.render_help():
if self.enable_help_tips:
self.console.print(f"\n[bold]tip:[/bold] {self.get_tip()}")
elif command and not command.arg_parser:
else:
self.console.print(
f"[bold]No detailed help available for '{command.description}'.[/bold]"
)
else:
self.console.print(f"[bold]No command found for '{key}'.[/bold]")
async def _render_tag_help(self, tag: str) -> None:
"""Renders a list of commands matching a specific tag."""
@@ -557,6 +579,30 @@ class Falyx:
if self.enable_help_tips:
self.console.print(f"[bold]tip:[/bold] {self.get_tip()}")
async def _render_unknown_route(self, route: RouteResult) -> None:
context = route.context
typed_key = context.typed_path[0].upper()
await route.namespace.render_namespace_help(context)
self.console.print(
f"[{OneColors.DARK_RED}]❌ Unknown Command or FalyxNamespace [{typed_key}]"
)
return None
async def render_namespace_help(
self, context: InvocationContext, tldr: bool = False
) -> None:
if context.mode is FalyxMode.MENU:
await self._render_menu_help()
else:
print(
self.format_invocation_path(
context.program,
context.typed_path,
cli_mode=True,
)
)
await self._render_cli_help()
async def _render_cli_help(self) -> None:
"""Renders the CLI help menu with all available commands and options."""
usage = self.usage or "[GLOBAL OPTIONS] [COMMAND] [OPTIONS]"
@@ -607,26 +653,36 @@ class Falyx:
if self.enable_help_tips:
self.console.print(f"\n[bold]tip:[/bold] {self.get_tip()}")
async def _render_help(
async def render_help(
self,
tag: str = "",
key: str | None = None,
tldr: bool = False,
) -> None:
"""Renders the help menu with command details, usage examples, and tips."""
if tldr:
await self._render_command_tldr(key)
return None
if key:
await self._render_command_help(key)
return None
if tag:
entry, suggestions = self.resolve_entry(key)
if suggestions:
self.console.print(
f"[{OneColors.LIGHT_YELLOW}]⚠️ Unknown entry '{key}'. Did you mean:[/]"
f"{', '.join(suggestions)[:10]}"
)
elif isinstance(entry, Command):
await self._render_command_help(entry, tldr)
elif isinstance(entry, FalyxNamespace):
await entry.namespace.render_namespace_help(
self.get_current_invocation_context(), tldr
)
else:
self.console.print(
f"[{OneColors.DARK_RED}]❌ No entry found for '{key}'.[/]"
)
elif tag:
await self._render_tag_help(tag)
return None
if self.options.get("mode") == FalyxMode.MENU:
elif self.options.get("mode") == FalyxMode.MENU:
await self._render_menu_help()
return None
await self._render_cli_help()
else:
await self._render_cli_help()
def _get_help_command(self) -> Command:
"""Returns the help command for the menu."""
@@ -667,7 +723,7 @@ class Falyx:
aliases=["HELP", "?"],
description="Help",
help_text="Show this help menu.",
action=Action("Help", self._render_help),
action=Action("Help", self.render_help),
style=OneColors.LIGHT_YELLOW,
arg_parser=parser,
ignore_in_history=True,
@@ -675,15 +731,11 @@ class Falyx:
program=self.program,
)
async def _preview(self, command_key: str) -> None:
async def _preview(self, key: str) -> None:
"""Previews the execution of a command without actually running it."""
_, command, args, kwargs, execution_args = await self.get_command(
command_key, from_help=True
)
command = await self.resolve_command(key)
if not command:
self.console.print(
f"[{OneColors.DARK_RED}]❌ Command '{command_key}' not found."
)
self.console.print(f"[{OneColors.DARK_RED}]❌ Command '{key}' not found.")
return None
self.console.print(f"Preview of command '{command.key}': {command.description}")
await command.preview()
@@ -699,7 +751,7 @@ class Falyx:
help_text="Preview the execution of a command without running it.",
)
preview_parser.add_argument(
"command_key",
"key",
help="The key or alias of the command to preview.",
)
preview_parser.add_tldr_examples(
@@ -747,7 +799,7 @@ class Falyx:
"""Adds a built-in command to Falyx."""
self._validate_command_aliases(command.key, command.aliases)
self.builtins[command.key.upper()] = command
_ = self._name_map
_ = self._entry_map
def _register_default_builtins(self) -> None:
"""Registers the default built-in commands for Falyx."""
@@ -761,19 +813,15 @@ class Falyx:
def _get_validator_error_message(self) -> str:
"""Validator to check if the input is a valid command."""
keys = {self.exit_command.key.upper()}
keys.update({alias.upper() for alias in self.exit_command.aliases})
if self.history_command:
keys.add(self.history_command.key.upper())
keys.update({alias.upper() for alias in self.history_command.aliases})
for command in self.builtins.values():
keys.add(command.key.upper())
keys.update({alias.upper() for alias in command.aliases})
for command in self.commands.values():
keys.add(command.key.upper())
keys.update({alias.upper() for alias in command.aliases})
visible = self.iter_visible_entries(
include_help=True,
include_history=True,
include_exit=True,
)
keys = {entry.key.upper() for entry in visible}
for entry in visible:
for alias in entry.aliases:
keys.add(alias.upper())
commands_str = ", ".join(sorted(keys))
@@ -799,9 +847,7 @@ class Falyx:
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 | str | Callable[[], Any] = BottomBar(
self.columns, self.key_bindings
)
self._bottom_bar = BottomBar(self.columns, self.key_bindings)
elif isinstance(bottom_bar, BottomBar):
bottom_bar.key_bindings = self.key_bindings
self._bottom_bar = bottom_bar
@@ -815,7 +861,7 @@ class Falyx:
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._named_items:
if isinstance(self.bottom_bar, BottomBar) and self.bottom_bar.has_items:
return self.bottom_bar.render
elif callable(self.bottom_bar):
return self.bottom_bar
@@ -916,7 +962,7 @@ class Falyx:
) -> None:
"""Updates the back command of the menu."""
self._validate_command_aliases(key, aliases)
action = action or Action(description, action=_noop)
action = action or SignalAction(description, QuitSignal())
if not callable(action):
raise InvalidActionError("Action must be a callable.")
self.exit_command = Command(
@@ -936,15 +982,32 @@ class Falyx:
self.exit_command.arg_parser.add_tldr_examples([("", help_text)])
def add_submenu(
self, key: str, description: str, submenu: Falyx, *, style: str = OneColors.CYAN
self,
key: str,
description: str,
submenu: Falyx,
*,
style: str = OneColors.CYAN,
aliases: list[str] | None = None,
help_text: str = "",
) -> None:
"""Adds a submenu to the menu."""
if not isinstance(submenu, Falyx):
raise NotAFalyxError("submenu must be an instance of Falyx.")
self._validate_command_aliases(key, [])
self.add_command(
key, description, submenu.menu, style=style, simple_help_signature=True
self._validate_command_aliases(key, aliases)
entry = FalyxNamespace(
key=key,
description=description,
namespace=submenu,
aliases=aliases or [],
help_text=help_text or f"Open the {description} namespace.",
style=style,
)
self.namespaces[key] = entry
if submenu.exit_command.key == "X":
submenu.update_exit_command(
key="B",
@@ -971,7 +1034,7 @@ class Falyx:
raise FalyxError("command must be an instance of Command.")
self._validate_command_aliases(command.key, command.aliases)
self.commands[command.key] = command
_ = self._name_map
_ = self._entry_map
def add_command(
self,
@@ -1064,7 +1127,7 @@ class Falyx:
)
self.commands[key] = command
_ = self._name_map
_ = self._entry_map
return command
def get_bottom_row(self) -> list[str]:
@@ -1086,18 +1149,39 @@ class Falyx:
)
return bottom_row
def iter_visible_entries(
self,
*,
include_builtins: bool = False,
include_help: bool = False,
include_history: bool = False,
include_exit: bool = False,
) -> list[Command | FalyxNamespace]:
visible: list[Command | FalyxNamespace] = []
visible.extend([cmd for cmd in self.commands.values() if not cmd.hidden])
visible.extend([ns for ns in self.namespaces.values() if not ns.hidden])
if include_builtins:
visible.extend([cmd for cmd in self.builtins.values() if not cmd.hidden])
if include_help:
visible.append(self.help_command)
if include_history and self.history_command:
visible.append(self.history_command)
if include_exit:
visible.append(self.exit_command)
return visible
def build_default_table(self) -> Table:
"""Build the standard table layout.
Developers can subclass or call this in custom tables.
"""
table = Table(title=self.title, show_header=False, box=box.SIMPLE) # type: ignore[arg-type]
visible_commands = [item for item in self.commands.items() if not item[1].hidden]
for chunk in chunks(visible_commands, self.columns):
visible = self.iter_visible_entries()
for chunk in chunks(visible, self.columns):
row = []
for key, command in chunk:
escaped_key = escape(f"[{key}]")
row.append(f"{escaped_key} [{command.style}]{command.description}")
for entry in chunk:
escaped_key = escape(f"[{entry.key}]")
row.append(f"{escaped_key} [{entry.style}]{entry.description}")
table.add_row(*row)
bottom_row = self.get_bottom_row()
for row in chunks(bottom_row, self.columns):
@@ -1139,105 +1223,176 @@ class Falyx:
return True, input_str[1:].strip()
return False, input_str.strip()
async def get_command(
self, raw_choices: str, from_validate=False, from_help=False
) -> tuple[bool, Command | None, tuple, dict[str, Any], dict[str, Any]]:
"""Returns the selected command based on user input.
def resolve_entry(
self,
token: str,
) -> tuple[Command | FalyxNamespace | None, list[str]]:
normalized = token.upper().strip()
Supports keys, aliases, and abbreviations.
"""
args = ()
# exact match
if normalized in self._entry_map:
return self._entry_map[normalized], []
# unique prefix match
prefix_matches = []
seen = set()
for key, entry in self._entry_map.items():
if key.startswith(normalized) and id(entry) not in seen:
prefix_matches.append(entry)
seen.add(id(entry))
if len(prefix_matches) == 1:
return prefix_matches[0], []
suggestions = get_close_matches(
normalized, list(self._entry_map.keys()), n=3, cutoff=0.7
)
return None, suggestions
async def prepare_route(
self,
raw_arguments: list[str] | str,
*,
mode: FalyxMode | None = None,
from_validate: bool = False,
) -> tuple[RouteResult | None, tuple, dict[str, Any], dict[str, Any]]:
args: tuple = ()
kwargs: dict[str, Any] = {}
execution_args: dict[str, Any] = {}
try:
choice, *input_args = shlex.split(raw_choices)
except ValueError:
return False, None, args, kwargs, execution_args
is_preview, choice = self.parse_preview_command(choice)
if is_preview and not choice and self.help_command:
is_preview = False
choice = "?"
elif is_preview and not choice:
# No help (list) command enabled
if not from_validate:
self.console.print(
f"[{OneColors.DARK_RED}]❌ You must enter a command for preview mode."
)
return is_preview, None, args, kwargs, execution_args
choice = choice.upper()
name_map = self._name_map
run_command = None
if name_map.get(choice):
run_command = name_map[choice]
else:
prefix_matches = [
command for key, command in name_map.items() if key.startswith(choice)
]
if len(prefix_matches) == 1:
run_command = prefix_matches[0]
if run_command:
if not from_validate:
logger.info("Command '%s' selected.", run_command.key)
if is_preview:
return True, run_command, args, kwargs, execution_args
elif self.is_cli_mode or from_help:
return False, run_command, args, kwargs, execution_args
if isinstance(raw_arguments, str):
try:
args, kwargs, execution_args = await run_command.resolve_args(
input_args, from_validate
)
except (CommandArgumentError, Exception) as error:
if not from_validate:
run_command.render_help()
self.console.print(
f"[{OneColors.DARK_RED}]❌ [{run_command.key}]: {error}"
)
else:
tokens = shlex.split(raw_arguments)
except ValueError as error:
if from_validate:
raise ValidationError(
message=str(error), cursor_position=len(raw_choices)
)
return is_preview, None, args, kwargs, execution_args
except HelpSignal:
return True, None, args, kwargs, execution_args
return is_preview, run_command, args, kwargs, execution_args
fuzzy_matches = get_close_matches(choice, list(name_map.keys()), n=3, cutoff=0.7)
if fuzzy_matches:
if not from_validate:
cursor_position=len(raw_arguments), message=str(error)
) from error
self.console.print(
f"[{OneColors.LIGHT_YELLOW}]⚠️ Unknown command '{choice}'. "
"Did you mean:"
)
for match in fuzzy_matches:
command = name_map[match]
self.console.print(f" • [bold]{match}[/] → {command.description}")
else:
raise ValidationError(
message=f"Unknown command '{choice}'. Did you mean: "
f"{', '.join(fuzzy_matches)}?",
cursor_position=len(raw_choices),
f"Parse error: {error}",
style=OneColors.DARK_RED,
)
return None, args, kwargs, execution_args
elif isinstance(raw_arguments, list):
tokens = raw_arguments
else:
if not from_validate:
self.console.print(
f"[{OneColors.LIGHT_YELLOW}]⚠️ Unknown command '{choice}'[/]"
)
else:
if from_validate:
raise ValidationError(
message=f"Unknown command '{choice}'.",
cursor_position=len(raw_choices),
cursor_position=len(raw_arguments),
message="TypeError",
)
return is_preview, None, args, kwargs, execution_args
return None, args, kwargs, execution_args
is_preview = False
if tokens and tokens[0].startswith("?"):
is_preview = True
tokens[0] = tokens[0][1:]
context = InvocationContext(
program=self.program,
typed_path=[],
mode=mode or self.options.get("mode"),
is_preview=is_preview,
)
route = await self.resolve_route(tokens, context=context)
if is_preview:
route.is_preview = True
return route, args, kwargs, execution_args
if route.kind is RouteKind.COMMAND:
assert route.command is not None
try:
args, kwargs, execution_args = await route.command.resolve_args(
route.leaf_argv, from_validate=from_validate
)
except CommandArgumentError as error:
if from_validate:
raise ValidationError(
cursor_position=len(raw_arguments), message=str(error)
) from error
else:
route.command.render_help()
self.console.print(
f"[{OneColors.DARK_RED}]❌ [{route.command.key}]: {error}"
)
raise error
except HelpSignal:
if not from_validate:
raise
return route, args, kwargs, execution_args
return route, args, kwargs, execution_args
async def _dispatch_route(
self,
route: RouteResult,
*,
args: tuple = (),
kwargs: dict[str, Any] | None = None,
execution_args: dict[str, Any] | None = None,
raise_on_error: bool = False,
wrap_errors: bool = True,
summary_last_result: bool = False,
) -> Any | None:
if route.kind is RouteKind.NAMESPACE_MENU:
await route.namespace.menu()
return None
if route.kind is RouteKind.NAMESPACE_HELP:
await route.namespace.render_namespace_help(route.context)
return None
if route.kind is RouteKind.NAMESPACE_TLDR:
await route.namespace.render_namespace_help(route.context, tldr=True)
return None
if route.kind is RouteKind.UNKNOWN:
await self._render_unknown_route(route)
return None
if route.kind is RouteKind.COMMAND:
if not route.command:
self.console.print(
f"[{OneColors.DARK_RED}]Error: No command specified for execution mode.[/]"
)
if wrap_errors or raise_on_error:
raise FalyxError
return None
command = route.command
if route.is_preview:
logger.info("Preview command '%s' selected.", command.key)
await command.preview()
return None
logger.debug(
"Executing command '%s' with args=%s, kwargs=%s, execution_args=%s",
route.command.description,
args,
kwargs,
execution_args,
)
return await self._executor.execute(
command=route.command,
args=args,
kwargs=kwargs or {},
execution_args=execution_args or {},
raise_on_error=raise_on_error,
wrap_errors=wrap_errors,
summary_last_result=summary_last_result,
)
async def execute_command(
self,
raw_arguments: str,
raw_arguments: list[str] | str,
*,
raise_on_error: bool = False,
wrap_errors: bool = True,
summary_last_result: bool = False,
mode: FalyxMode = FalyxMode.MENU,
) -> Any | None:
"""Execute a command from a raw CLI-style input string.
@@ -1247,10 +1402,9 @@ class Falyx:
Behavior:
- Resolves the command and its parsed `args`, `kwargs`, and
`execution_args` via `get_command()`.
`execution_args` via `prepare_route()`.
- Returns `None` when help output is triggered, argument parsing fails,
the command cannot be found, or preview mode is requested.
- Updates `last_run_command` when a valid command is resolved.
- Raises `QuitSignal` if the resolved command is the configured exit
command.
- For normal execution, forwards the resolved command and execution
@@ -1281,53 +1435,86 @@ class Falyx:
command from a raw input string outside the interactive menu loop.
"""
try:
is_preview, command, args, kwargs, execution_args = await self.get_command(
raw_arguments
route, args, kwargs, execution_args = await self.prepare_route(
raw_arguments, mode=mode
)
except (CommandArgumentError, Exception):
return None
except HelpSignal:
return None
except CommandArgumentError as error:
logger.error(
"Argument parsing error for input '%s': %s", raw_arguments, error
)
self.console.print(f"[{OneColors.DARK_RED}]❌ ['{raw_arguments}'] {error}[/]")
if route is None:
return None
if not command:
logger.error("Command not found for input '%s'", raw_arguments)
self.console.print(
f"[{OneColors.DARK_RED}]❌ ['{raw_arguments}'] Command not found.[/]"
)
return None
self.last_run_command = command
if is_preview:
logger.info("Preview command '%s' selected.", command.key)
await command.preview()
return None
if command == self.exit_command:
logger.info("Back selected: exiting %s", self.get_title())
raise QuitSignal()
logger.debug(
"Executing command '%s' with args=%s, kwargs=%s, execution_args=%s",
command.description,
args,
kwargs,
execution_args,
)
return await self._executor.execute(
command=command,
return await self._dispatch_route(
route=route,
args=args,
kwargs=kwargs or {},
execution_args=execution_args or {},
kwargs=kwargs,
execution_args=execution_args,
raise_on_error=raise_on_error,
wrap_errors=wrap_errors,
summary_last_result=summary_last_result,
)
async def resolve_route(
self,
tokens: list[str],
*,
context: InvocationContext,
) -> RouteResult:
# 1. No more tokens -> this namespace itself was targeted
if not tokens:
return RouteResult(
kind=RouteKind.NAMESPACE_MENU,
namespace=self,
context=context,
)
head, *tail = tokens
# 2. Namespace-level help/tldr belongs to the current namespace
if head in {"-h", "--help"}:
return RouteResult(
kind=RouteKind.NAMESPACE_HELP,
namespace=self,
context=context,
)
if head in {"-T", "--tldr"}:
return RouteResult(
kind=RouteKind.NAMESPACE_TLDR,
namespace=self,
context=context,
)
# 3. Resolve the next entry in this namespace
entry, suggestions = self.resolve_entry(head)
if entry is None:
return RouteResult(
kind=RouteKind.UNKNOWN,
namespace=self,
context=context,
suggestions=suggestions,
)
child_context = context.child(head)
# 4. Namespace entry -> recurse with remaining tokens
if isinstance(entry, FalyxNamespace):
return await entry.namespace.resolve_route(
tail,
context=child_context,
)
# 5. Leaf command -> stop routing; leave tail untouched for leaf parser
return RouteResult(
kind=RouteKind.COMMAND,
namespace=self,
context=child_context,
command=entry,
leaf_argv=tail,
)
async def process_command(self) -> None:
"""Processes the action of the selected command."""
app = get_app()
@@ -1388,7 +1575,7 @@ class Falyx:
if self.exit_message:
self.print_message(self.exit_message)
def _apply_parse_result(self, result: ParseResult) -> None:
def _apply_parse_result(self, result: RootParseResult) -> None:
"""Applies the parsed CLI arguments to the menu options."""
self.options.set("mode", result.mode)
@@ -1419,22 +1606,6 @@ class Falyx:
CLI arguments, configures runtime state, and dispatches execution based
on the resolved mode.
Execution Pipeline:
1. Parse CLI input via `FalyxParser` into a `ParseResult`
2. Optionally invoke a user-provided callback with the parse result
3. Apply root-level options (e.g. verbose, debug hooks, prompt behavior)
4. Dispatch based on `ParseResult.mode`:
- HELP: Render help output and exit
- COMMAND: Execute a resolved command
- MENU: Launch interactive menu loop
- ERROR: Render error and exit
Command Execution:
- Arguments are parsed via `CommandArgumentParser`
- Execution options (e.g. retries, confirmation flags) are separated
- Execution-scoped overrides are applied using `OptionsManager`
- Commands are executed via `CommandExecutor.execute()` with full lifecycle hooks
Callback Behavior:
- If provided, `callback` is executed after parsing but before dispatch
- Supports both sync and async callables
@@ -1466,9 +1637,7 @@ class Falyx:
>>> asyncio.run(flx.run())
```
"""
falyx_parser = FalyxParser(self)
parse_result = falyx_parser.parse(sys.argv[1:])
parse_result = FalyxParser.parse(sys.argv[1:])
if callback:
if not callable(callback):
@@ -1478,74 +1647,54 @@ class Falyx:
self._apply_parse_result(parse_result)
if parse_result.mode == FalyxMode.ERROR:
await self._render_help()
self.console.print(f"[{OneColors.DARK_RED}]Error: {parse_result.error}[/]")
sys.exit(1)
if parse_result.mode == FalyxMode.HELP:
await self._render_help()
await self.render_help()
sys.exit(0)
if parse_result.mode == FalyxMode.COMMAND:
if not parse_result.command:
self.console.print(
f"[{OneColors.DARK_RED}]Error: No command specified for execution mode.[/]"
)
sys.exit(1)
command = parse_result.command
try:
route, args, kwargs, execution_args = await self.prepare_route(
raw_arguments=parse_result.remaining_argv,
)
except CommandArgumentError:
sys.exit(2)
except HelpSignal:
sys.exit(0)
if parse_result.is_preview:
if command is None:
sys.exit(1)
logger.info("Preview command '%s' selected.", command.key)
await command.preview()
sys.exit(0)
if not command:
sys.exit(1)
try:
args, kwargs, execution_args = await command.resolve_args(
parse_result.command_argv
)
except HelpSignal:
sys.exit(0)
except CommandArgumentError as error:
command.render_help()
self.console.print(f"[{OneColors.DARK_RED}]❌ ['{command.key}'] {error}")
sys.exit(2)
try:
logger.debug(
"Executing command '%s' with args=%s, kwargs=%s, execution_args=%s",
command.description,
args,
kwargs,
execution_args,
)
await self._executor.execute(
command=command,
args=args,
kwargs=kwargs,
execution_args=execution_args,
raise_on_error=False,
wrap_errors=True,
)
except FalyxError as error:
self.console.print(f"[{OneColors.DARK_RED}]❌ Error: {error}[/]")
sys.exit(1)
except QuitSignal:
logger.info("[QuitSignal]. <- Exiting run.")
sys.exit(130)
except BackSignal:
logger.info("[BackSignal]. <- Exiting run.")
sys.exit(1)
except CancelSignal:
logger.info("[CancelSignal]. <- Exiting run.")
sys.exit(1)
except asyncio.CancelledError:
logger.info("[asyncio.CancelledError]. <- Exiting run.")
sys.exit(1)
if not route:
await self.render_help()
self.console.print(
f"[{OneColors.DARK_RED}]❌ Error unable to parse: {parse_result.raw_argv}"
)
sys.exit(2)
if not always_start_menu:
sys.exit(0)
try:
await self._dispatch_route(
route=route,
args=args,
kwargs=kwargs,
execution_args=execution_args,
raise_on_error=False,
wrap_errors=True,
)
except FalyxError as error:
self.console.print(f"[{OneColors.DARK_RED}]❌ Error: {error}[/]")
sys.exit(1)
except Exception:
sys.exit(1)
except QuitSignal:
logger.info("[QuitSignal]. <- Exiting run.")
sys.exit(130)
except BackSignal:
logger.info("[BackSignal]. <- Exiting run.")
sys.exit(1)
except CancelSignal:
logger.info("[CancelSignal]. <- Exiting run.")
sys.exit(1)
except asyncio.CancelledError:
logger.info("[asyncio.CancelledError]. <- Exiting run.")
sys.exit(1)
if route.kind is RouteKind.NAMESPACE_MENU or not always_start_menu:
sys.exit(0)
await self.menu()

20
falyx/namespace.py Normal file
View File

@@ -0,0 +1,20 @@
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
from falyx.themes import OneColors
if TYPE_CHECKING:
from falyx.falyx import Falyx
@dataclass
class FalyxNamespace:
key: str
description: str
namespace: Falyx
aliases: list[str] = field(default_factory=list)
help_text: str = ""
style: str = OneColors.CYAN
hidden: bool = False

2
falyx/parser/__init__.py
View File

@@ -9,7 +9,7 @@ from .argument import Argument
from .argument_action import ArgumentAction
from .command_argument_parser import CommandArgumentParser
from .falyx_parser import FalyxParser
from .parse_result import ParseResult
from .parse_result import RootParseResult
__all__ = [
"Argument",

4
falyx/parser/command_argument_parser.py
View File

@@ -58,7 +58,7 @@ from rich.panel import Panel
from falyx.action.base_action import BaseAction
from falyx.console import console
from falyx.exceptions import CommandArgumentError
from falyx.exceptions import CommandArgumentError, NotAFalyxError
from falyx.execution_option import ExecutionOption
from falyx.mode import FalyxMode
from falyx.options_manager import OptionsManager
@@ -170,7 +170,7 @@ class CommandArgumentParser:
def set_options_manager(self, options_manager: OptionsManager) -> None:
"""Set the options manager for the parser."""
if not isinstance(options_manager, OptionsManager):
raise ValueError("options_manager must be an instance of OptionsManager")
raise NotAFalyxError("options_manager must be an instance of OptionsManager")
self.options_manager = options_manager
def enable_execution_options(

110
falyx/parser/falyx_parser.py
View File

@@ -2,15 +2,10 @@
from __future__ import annotations
from dataclasses import dataclass
from difflib import get_close_matches
from typing import TYPE_CHECKING
from falyx.mode import FalyxMode
from falyx.parser.parse_result import ParseResult
if TYPE_CHECKING:
from falyx.command import Command
from falyx.falyx import Falyx
from falyx.parser.parse_result import RootParseResult
@dataclass(slots=True)
@@ -18,7 +13,6 @@ class RootOptions:
verbose: bool = False
debug_hooks: bool = False
never_prompt: bool = False
version: bool = False
help: bool = False
@@ -42,11 +36,9 @@ class FalyxParser:
"--help": "help",
}
def __init__(self, falyx: Falyx) -> None:
self.falyx = falyx
@classmethod
def _parse_root_options(
self,
cls,
argv: list[str],
) -> tuple[RootOptions, list[str]]:
"""Parse only root/session flags from the start of argv.
@@ -69,7 +61,7 @@ class FalyxParser:
remaining_start = index + 1
break
attr = self.ROOT_FLAG_ALIASES.get(token)
attr = cls.ROOT_FLAG_ALIASES.get(token)
if attr is None:
remaining_start = index
break
@@ -81,79 +73,13 @@ class FalyxParser:
remaining = argv[remaining_start:]
return options, remaining
def resolve_command(self, token: str) -> tuple[Command | None, list[str]]:
"""Resolve a command by key, alias, or unique prefix.
Returns:
(command, suggestions)
"""
normalized = token.upper().strip()
name_map = self.falyx._name_map
if normalized in name_map:
return name_map[normalized], []
prefix_matches = []
seen = set()
for key, command in name_map.items():
if key.startswith(normalized) and id(command) not in seen:
prefix_matches.append(command)
seen.add(id(command))
if len(prefix_matches) == 1:
return prefix_matches[0], []
suggestions = get_close_matches(
normalized, list(name_map.keys()), n=3, cutoff=0.7
)
return None, suggestions
def _parse_command(
self,
argv: list[str],
root: RootOptions,
remaining: list[str],
) -> ParseResult:
raw_name = remaining[0]
is_preview = raw_name.startswith("?")
command_name = raw_name[1:] if is_preview else raw_name
command, suggestions = self.resolve_command(command_name)
if not command:
sugguestions_text = (
f" Did you mean: {', '.join(suggestions)}?" if suggestions else ""
)
return ParseResult(
mode=FalyxMode.ERROR,
raw_argv=argv,
command_name=command_name,
command_argv=remaining[1:],
verbose=root.verbose,
debug_hooks=root.debug_hooks,
never_prompt=root.never_prompt,
error=f"Unknown command '{command_name}'.{sugguestions_text}",
)
command_argv = remaining[1:]
return ParseResult(
mode=FalyxMode.COMMAND,
raw_argv=argv,
command_name=command_name,
command=command,
command_argv=command_argv,
is_preview=is_preview,
verbose=root.verbose,
debug_hooks=root.debug_hooks,
never_prompt=root.never_prompt,
)
def parse(self, argv: list[str] | None = None) -> ParseResult:
@classmethod
def parse(cls, argv: list[str] | None = None) -> RootParseResult:
argv = argv or []
root, remaining = self._parse_root_options(argv)
root, remaining = cls._parse_root_options(argv)
if root.help:
return ParseResult(
return RootParseResult(
mode=FalyxMode.HELP,
raw_argv=argv,
never_prompt=root.never_prompt,
@@ -161,15 +87,11 @@ class FalyxParser:
debug_hooks=root.debug_hooks,
)
if not remaining:
return ParseResult(
mode=FalyxMode.MENU,
raw_argv=argv,
verbose=root.verbose,
debug_hooks=root.debug_hooks,
never_prompt=root.never_prompt,
)
head, *tail = remaining
return self._parse_command(argv, root, remaining)
return RootParseResult(
mode=FalyxMode.COMMAND,
raw_argv=argv,
verbose=root.verbose,
debug_hooks=root.debug_hooks,
never_prompt=root.never_prompt,
remaining_argv=remaining,
)

14
falyx/parser/parse_result.py
View File

@@ -1,24 +1,14 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING
from falyx.mode import FalyxMode
if TYPE_CHECKING:
from falyx.command import Command
@dataclass(slots=True)
class ParseResult:
class RootParseResult:
mode: FalyxMode
raw_argv: list[str] = field(default_factory=list)
verbose: bool = False
debug_hooks: bool = False
never_prompt: bool = False
command_name: str = ""
command: Command | None = None
command_argv: list[str] = field(default_factory=list)
is_preview: bool = False
error: str | None = None
remaining_argv: list[str] = field(default_factory=list)

4
falyx/protocols.py
View File

@@ -29,4 +29,6 @@ class ActionFactoryProtocol(Protocol):
@runtime_checkable
class ArgParserProtocol(Protocol):
def __call__(self, args: list[str]) -> tuple[tuple, dict, dict]: ...
def __call__(
self, args: list[str]
) -> tuple[tuple, dict[str, Any], dict[str, Any]]: ...

33
falyx/routing.py Normal file
View File

@@ -0,0 +1,33 @@
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import TYPE_CHECKING
from falyx.context import InvocationContext
from falyx.namespace import FalyxNamespace
if TYPE_CHECKING:
from falyx.command import Command
from falyx.falyx import Falyx
class RouteKind(Enum):
COMMAND = "command"
NAMESPACE_MENU = "namespace_menu"
NAMESPACE_HELP = "namespace_help"
NAMESPACE_TLDR = "namespace_tldr"
UNKNOWN = "unknown"
@dataclass(slots=True)
class RouteResult:
kind: RouteKind
namespace: "Falyx"
context: InvocationContext
command: "Command | None" = None
namespace_entry: FalyxNamespace | None = None
leaf_argv: list[str] = field(default_factory=list)
typed_path: list[str] = field(default_factory=list)
suggestions: list[str] = field(default_factory=list)
is_preview: bool = False

28
falyx/validators.py
View File

@@ -22,6 +22,8 @@ from typing import TYPE_CHECKING, KeysView, Sequence
from prompt_toolkit.validation import ValidationError, Validator
from falyx.routing import RouteKind
if TYPE_CHECKING:
from falyx.falyx import Falyx
@@ -48,12 +50,28 @@ class CommandValidator(Validator):
message=self.error_message,
cursor_position=len(text),
)
is_preview, choice, _, __, ___ = await self.falyx.get_command(
text, from_validate=True
)
if is_preview:
route, _, __, ___ = await self.falyx.prepare_route(text, from_validate=True)
if not route:
raise ValidationError(
message=self.error_message,
cursor_position=len(text),
)
if route.is_preview:
return None
if not choice:
if route.kind in {
RouteKind.NAMESPACE_MENU,
RouteKind.NAMESPACE_HELP,
RouteKind.NAMESPACE_TLDR,
}:
return None
if route.kind is RouteKind.COMMAND and route.command is None:
raise ValidationError(
message=self.error_message,
cursor_position=len(text),
)
elif route.kind is RouteKind.COMMAND:
return None
if route.kind is RouteKind.UNKNOWN:
raise ValidationError(
message=self.error_message,
cursor_position=len(text),

11
tests/test_command_argument_parser.py
View File

@@ -1,5 +1,7 @@
import pytest
from rich.text import Text
from falyx.console import console as falyx_console
from falyx.exceptions import CommandArgumentError
from falyx.parser import ArgumentAction, CommandArgumentParser
from falyx.signals import HelpSignal
@@ -825,4 +827,11 @@ async def test_render_help():
parser.add_argument("--foo", type=str, help="Foo help")
parser.add_argument("--bar", action=ArgumentAction.APPEND, type=str, help="Bar help")
assert parser.render_help() is None
with falyx_console.capture() as capture:
parser.render_help()
output = Text.from_ansi(capture.get()).plain
assert "usage:" in output
assert "--foo" in output
assert "Foo help" in output
assert "--bar" in output
assert "Bar help" in output

2
tests/test_completer/test_lcp_completions.py
View File

@@ -17,7 +17,7 @@ def fake_falyx():
help_command=SimpleNamespace(key="H", aliases=["HELP"]),
history_command=SimpleNamespace(key="Y", aliases=["HISTORY"]),
commands={"R": fake_command},
_name_map={"R": fake_command, "RUN": fake_command, "X": fake_command},
_entry_map={"R": fake_command, "RUN": fake_command, "X": fake_command},
)

30
tests/test_execution_option.py Normal file
View File

@@ -0,0 +1,30 @@
import pytest
from falyx.execution_option import ExecutionOption
def test_execution_option_accepts_valid_string_values():
assert ExecutionOption("summary") == ExecutionOption.SUMMARY
assert ExecutionOption("retry") == ExecutionOption.RETRY
assert ExecutionOption("confirm") == ExecutionOption.CONFIRM
def test_execution_option_rejects_invalid_string():
with pytest.raises(ValueError, match="Invalid ExecutionOption: 'invalid'"):
ExecutionOption("invalid")
def test_execution_option_normalizes_case_and_whitespace():
assert ExecutionOption(" SUMMARY ") == ExecutionOption.SUMMARY
assert ExecutionOption("ReTrY") == ExecutionOption.RETRY
assert ExecutionOption("\tconfirm\n") == ExecutionOption.CONFIRM
def test_execution_option_rejects_non_string():
with pytest.raises(ValueError, match="Invalid ExecutionOption: 123"):
ExecutionOption(123)
def test_execution_option_error_lists_valid_values():
with pytest.raises(ValueError, match="Must be one of: summary, retry, confirm"):
ExecutionOption("invalid")

3
tests/test_falyx/test_help.py
View File

@@ -51,7 +51,7 @@ async def test_render_help(capsys):
aliases=["SC"],
help_text="This is a sample command.",
)
await flx._render_help()
await flx.render_help()
captured = capsys.readouterr()
assert "This is a sample command." in captured.out
@@ -75,7 +75,6 @@ async def test_help_command_by_tag(capsys):
await flx.execute_command("H -t tag1")
captured = capsys.readouterr()
print(captured.out)
text = Text.from_ansi(captured.out)
assert "tag1" in text.plain
assert "This command is tagged." in text.plain

4
tests/test_falyx_parser/test_root_options.py
View File

@@ -1,10 +1,8 @@
from falyx import Falyx
from falyx.parser.falyx_parser import FalyxParser, RootOptions
def get_falyx_parser():
falyx = Falyx()
return FalyxParser(falyx=falyx)
return FalyxParser()
def test_parse_root_options_empty():

143
tests/test_parsers/test_execution_option_registration.py Normal file
View File

@@ -0,0 +1,143 @@
import pytest
from falyx.exceptions import CommandArgumentError
from falyx.execution_option import ExecutionOption
from falyx.parser import CommandArgumentParser
def test_enable_execution_options_registers_summary_flag():
parser = CommandArgumentParser()
parser.enable_execution_options(frozenset({ExecutionOption.SUMMARY}))
assert "--summary" in parser._flag_map
assert "--summary" in parser._keyword
assert "--summary" in parser._flag_map
assert "summary" in parser._execution_dests
def test_enable_execution_options_registers_retry_flags():
parser = CommandArgumentParser()
parser.enable_execution_options(frozenset({ExecutionOption.RETRY}))
assert "--retries" in parser._flag_map
assert "--retries" in parser._keyword
assert "--retries" in parser._flag_map
assert "retries" in parser._execution_dests
assert "--retry-delay" in parser._flag_map
assert "--retry-delay" in parser._keyword
assert "--retry-delay" in parser._flag_map
assert "retry_delay" in parser._execution_dests
assert "--retry-backoff" in parser._flag_map
assert "--retry-backoff" in parser._keyword
assert "--retry-backoff" in parser._flag_map
assert "retry_backoff" in parser._execution_dests
def test_enable_execution_options_registers_confirm_flags():
parser = CommandArgumentParser()
parser.enable_execution_options(frozenset({ExecutionOption.CONFIRM}))
assert "--confirm" in parser._flag_map
assert "--confirm" in parser._keyword
assert "--confirm" in parser._flag_map
assert "force_confirm" in parser._execution_dests
assert "--skip-confirm" in parser._flag_map
assert "--skip-confirm" in parser._keyword
assert "--skip-confirm" in parser._flag_map
assert "skip_confirm" in parser._execution_dests
def test_register_execution_dest_rejects_duplicates():
parser = CommandArgumentParser()
parser.enable_execution_options(frozenset({ExecutionOption.SUMMARY}))
with pytest.raises(
CommandArgumentError, match="Destination 'summary' is already defined"
):
parser.add_argument("--summary", action="store_true")
with pytest.raises(
CommandArgumentError, match="Destination 'summary' is already defined"
):
parser.enable_execution_options(frozenset({ExecutionOption.SUMMARY}))
@pytest.mark.asyncio
async def test_parse_args_split_with_execution_options_returns_correct_execution_args():
parser = CommandArgumentParser()
parser.add_argument("foo", type=int, help="A business argument.")
parser.add_argument("--bar", type=int, help="A business argument.")
parser.enable_execution_options(
frozenset({ExecutionOption.SUMMARY, ExecutionOption.RETRY})
)
args, kwargs, execution_args = await parser.parse_args_split(
["50", "--bar", "42", "--summary", "--retries", "3"]
)
assert args == (50,)
assert kwargs == {"bar": 42}
assert execution_args == {
"summary": True,
"retries": 3,
"retry_delay": 0.0,
"retry_backoff": 0.0,
}
@pytest.mark.asyncio
async def test_parse_args_split_with_all_execution_options_returns_correct_execution_args():
parser = CommandArgumentParser()
parser.add_argument("foo", type=int, help="A business argument.")
parser.add_argument("--bar", type=int, help="A business argument.")
parser.enable_execution_options(
frozenset(
{
ExecutionOption.SUMMARY,
ExecutionOption.RETRY,
ExecutionOption.CONFIRM,
}
)
)
args, kwargs, execution_args = await parser.parse_args_split(
[
"50",
"--bar",
"42",
"--summary",
"--retries",
"3",
"--confirm",
]
)
assert args == (50,)
assert kwargs == {"bar": 42}
assert execution_args == {
"summary": True,
"retries": 3,
"retry_delay": 0.0,
"retry_backoff": 0.0,
"force_confirm": True,
"skip_confirm": False,
}
@pytest.mark.asyncio
async def test_parse_args_split_with_no_execution_options_returns_empty_execution_args():
parser = CommandArgumentParser()
parser.add_argument("foo", type=int, help="A business argument.")
parser.add_argument("--bar", type=int, help="A business argument.")
args, kwargs, execution_args = await parser.parse_args_split(["50", "--bar", "42"])
assert args == (50,)
assert kwargs == {"bar": 42}
assert execution_args == {}
@pytest.mark.asyncio
async def test_parse_args_split_with_conflicting_execution_option_raises():
parser = CommandArgumentParser()
parser.add_argument("--summary", action="store_true", help="A conflicting argument.")
with pytest.raises(
CommandArgumentError, match="Destination 'summary' is already defined"
):
parser.enable_execution_options(frozenset({ExecutionOption.SUMMARY}))

241
tests/test_parsers/test_resolve_args.py Normal file
View File

@@ -0,0 +1,241 @@
import pytest
from falyx.command import Command
from falyx.exceptions import CommandArgumentError, NotAFalyxError
from falyx.execution_option import ExecutionOption
@pytest.mark.asyncio
async def test_resolve_args_separates_business_and_execution_options():
command = Command.build(
key="T",
description="Test Command",
action=lambda: None,
execution_options=["summary", "retry"],
)
command.arg_parser.add_argument("--foo", type=int, help="A business argument.")
args, kwargs, execution_args = await command.resolve_args(
["--foo", "42", "--summary", "--retries", "3"]
)
assert args == ()
assert kwargs == {"foo": 42}
assert execution_args == {
"summary": True,
"retries": 3,
"retry_delay": 0.0,
"retry_backoff": 0.0,
}
args, kwargs, execution_args = await command.arg_parser.parse_args_split(
["--foo", "42", "--summary", "--retries", "3"]
)
assert args == ()
assert kwargs == {"foo": 42}
assert execution_args == {
"summary": True,
"retries": 3,
"retry_delay": 0.0,
"retry_backoff": 0.0,
}
@pytest.mark.asyncio
async def test_parse_args_split_with_no_execution_options_returns_empty_execution_args():
command = Command.build(
key="T",
description="Test Command",
action=lambda: None,
)
command.arg_parser.add_argument("--foo", type=int, help="A business argument.")
args, kwargs, execution_args = await command.arg_parser.parse_args_split(
["--foo", "42"]
)
assert args == ()
assert kwargs == {"foo": 42}
assert execution_args == {}
@pytest.mark.asyncio
async def test_resolve_args_raises_on_conflicting_execution_option():
command = Command.build(
key="T",
description="Test Command",
action=lambda: None,
execution_options=["summary"],
)
with pytest.raises(
CommandArgumentError, match="Destination 'summary' is already defined"
):
command.arg_parser.add_argument(
"--summary", action="store_true", help="A conflicting argument."
)
with pytest.raises(
CommandArgumentError, match="Destination 'summary' is already defined"
):
command.arg_parser.enable_execution_options(frozenset({ExecutionOption.SUMMARY}))
@pytest.mark.asyncio
async def test_resolve_args_mix_of_business_and_execution_options():
command = Command.build(
key="T",
description="Test Command",
action=lambda: None,
execution_options=["retry"],
)
command.arg_parser.add_argument("--summary", type=str, help="A business argument.")
args, kwargs, execution_args = await command.resolve_args(
["--summary", "test", "--retries", "5", "--retry-delay", "2"]
)
assert args == ()
assert kwargs == {"summary": "test"}
assert execution_args == {"retries": 5, "retry_delay": 2.0, "retry_backoff": 0.0}
@pytest.mark.asyncio
async def test_resolve_args_with_no_arguments():
command = Command.build(
key="T",
description="Test Command",
action=lambda: None,
execution_options=["summary"],
)
args, kwargs, execution_args = await command.resolve_args([])
assert args == ()
assert kwargs == {}
assert execution_args == {"summary": False}
@pytest.mark.asyncio
async def test_resolve_args_with_confirmation_options():
command = Command.build(
key="T",
description="Test Command",
action=lambda: None,
execution_options=["confirm"],
)
args, kwargs, execution_args = await command.resolve_args(["--confirm"])
assert args == ()
assert kwargs == {}
assert execution_args == {"force_confirm": True, "skip_confirm": False}
args, kwargs, execution_args = await command.resolve_args(["--skip-confirm"])
assert args == ()
assert kwargs == {}
assert execution_args == {"force_confirm": False, "skip_confirm": True}
@pytest.mark.asyncio
async def test_resolve_args_with_all_execution_options():
command = Command.build(
key="T",
description="Test Command",
action=lambda: None,
execution_options=["summary", "retry", "confirm"],
)
args, kwargs, execution_args = await command.resolve_args(
["--summary", "--retries", "3", "--confirm"]
)
assert args == ()
assert kwargs == {}
assert execution_args == {
"summary": True,
"retries": 3,
"retry_delay": 0.0,
"retry_backoff": 0.0,
"force_confirm": True,
"skip_confirm": False,
}
@pytest.mark.asyncio
async def test_resolve_args_with_raw_string_input():
command = Command.build(
key="T",
description="Test Command",
action=lambda: None,
execution_options=["summary"],
)
command.arg_parser.add_argument("--foo", type=int, help="A business argument.")
args, kwargs, execution_args = await command.resolve_args("--foo 42 --summary")
assert args == ()
assert kwargs == {"foo": 42}
assert execution_args == {"summary": True}
@pytest.mark.asyncio
async def test_resolve_args_with_no_arg_parser():
command = Command.build(
key="T",
description="Test Command",
action=lambda: None,
execution_options=["summary"],
)
command.arg_parser = None
with pytest.raises(
NotAFalyxError,
match="Command has no parser configured. Provide a custom_parser or CommandArgumentParser.",
):
await command.resolve_args("--summary")
@pytest.mark.asyncio
async def test_resolve_args_with_custom_parser():
def parse_args_split(arg_list):
return (arg_list,), {}, {"custom_execution_arg": True}
command = Command.build(
key="T",
description="Test Command",
action=lambda: None,
execution_options=["summary"],
)
command.custom_parser = parse_args_split
args, kwargs, execution_args = await command.resolve_args("--summary")
assert args == (["--summary"],)
assert kwargs == {}
assert execution_args == {"custom_execution_arg": True}
# TODO: is this the right behavior? Should we expect the custom parser to handle non string inputs as well? Does this actually happen?
args, kwargs, execution_args = await command.resolve_args(2235235)
assert args == (2235235,)
assert kwargs == {}
assert execution_args == {"custom_execution_arg": True}
with pytest.raises(CommandArgumentError, match="Failed to parse arguments:"):
args, kwargs, execution_args = await command.resolve_args("unbalanced 'quotes")
@pytest.mark.asyncio
async def test_resolve_args_str_unbalanced_quotes():
command = Command.build(
key="T",
description="Test Command",
action=lambda: None,
execution_options=["summary"],
)
command.arg_parser.add_argument("--foo", type=str, help="A business argument.")
with pytest.raises(CommandArgumentError, match="Failed to parse arguments:"):
await command.resolve_args("--foo 'unbalanced quotes")

516
tests/test_runner/test_command_runner.py Normal file
View File

@@ -0,0 +1,516 @@
import asyncio
import sys
import pytest
from rich.console import Console
from rich.text import Text
from falyx.action import Action
from falyx.command import Command
from falyx.command_runner import CommandRunner
from falyx.console import console as falyx_console
from falyx.exceptions import CommandArgumentError, FalyxError, NotAFalyxError
from falyx.hook_manager import HookManager, HookType
from falyx.options_manager import OptionsManager
from falyx.signals import BackSignal, CancelSignal, HelpSignal, QuitSignal
async def ok_action(*args, **kwargs):
falyx_console.print("Action executed with args:", args, "and kwargs:", kwargs)
return "ok"
async def failing_action(*args, **kwargs):
raise RuntimeError("boom")
async def throw_error_action(error: str):
if error == "QuitSignal":
raise QuitSignal("Quit signal triggered.")
elif error == "BackSignal":
raise BackSignal("Back signal triggered.")
elif error == "CancelSignal":
raise CancelSignal("Cancel signal triggered.")
elif error == "ValueError":
raise ValueError("This is a ValueError.")
elif error == "HelpSignal":
raise HelpSignal("Help signal triggered.")
elif error == "FalyxError":
raise FalyxError("This is a FalyxError.")
else:
raise asyncio.CancelledError("An error occurred in the action.")
@pytest.fixture
def command_throwing_error():
command = Command(
key="E",
description="Error Command",
action=Action("throw_error", throw_error_action),
execution_options=["retry"],
)
return command
@pytest.fixture
def command_with_parser():
command = Command(
key="T",
description="Test Command",
action=ok_action,
)
command.arg_parser.add_argument("--foo", type=int, help="A business argument.")
return command
@pytest.fixture
def command_with_no_parser():
command = Command(
key="T",
description="Test Command",
action=ok_action,
execution_options=["summary"],
)
command.arg_parser = None
return command
@pytest.fixture
def command_with_custom_parser():
def parse_args_split(arg_list):
return (arg_list,), {}, {"custom_execution_arg": True}
command = Command(
key="T",
description="Test Command",
action=ok_action,
execution_options=["summary"],
)
command.custom_parser = parse_args_split
return command
@pytest.fixture
def command_with_failing_action():
command = Command(
key="T",
description="Test Command",
action=failing_action,
execution_options=["summary", "retry"],
)
command.arg_parser.add_argument("--foo", type=int, help="A business argument.")
return command
@pytest.fixture
def command_build_with_all_execution_options():
return Command.build(
key="T",
description="Test Command",
action=ok_action,
execution_options=["summary", "retry", "confirm"],
)
@pytest.fixture
def console():
return Console(record=True)
@pytest.mark.asyncio
async def test_command_runner_initialization(
command_with_parser,
command_with_no_parser,
command_with_custom_parser,
):
runner = CommandRunner(command_with_parser)
assert runner.command == command_with_parser
assert isinstance(runner.options, OptionsManager)
assert isinstance(runner.runner_hooks, HookManager)
assert runner.console == falyx_console
assert runner.command.options_manager == runner.options
assert runner.command.arg_parser.options_manager == runner.options
assert runner.command.options_manager == runner.options
assert runner.executor.options == runner.options
assert runner.executor.hooks == runner.runner_hooks
assert runner.executor.console == runner.console
assert runner.options.get("summary", namespace_name="execution") is None
runner_no_parser = CommandRunner(command_with_no_parser)
assert runner_no_parser.command == command_with_no_parser
assert runner_no_parser.command.arg_parser is None
CommandRunner(command_with_no_parser)
with pytest.raises(
NotAFalyxError,
match="Command has no parser configured. Provide a custom_parser or CommandArgumentParser.",
):
await runner_no_parser.run("--summary")
runner_custom_parser = CommandRunner(command_with_custom_parser)
assert runner_custom_parser.command == command_with_custom_parser
assert runner_custom_parser.command.custom_parser is not None
def test_command_runner_initialization_with_custom_options(command_with_parser):
custom_options = OptionsManager([("default", {"summary": True})])
runner = CommandRunner(command_with_parser, options=custom_options)
assert runner.options == custom_options
assert runner.options.get("summary", namespace_name="default") is True
assert runner.command.options_manager == runner.options
assert runner.command.arg_parser.options_manager == runner.options
assert runner.command.options_manager == runner.options
def test_command_runner_initialization_with_custom_console(command_with_parser):
custom_console = Console()
runner = CommandRunner(command_with_parser, console=custom_console)
assert runner.console == custom_console
assert runner.executor.console == custom_console
def test_command_runner_initialization_with_custom_hooks(command_with_parser):
custom_hooks = HookManager()
custom_hooks.register("before", lambda context: print("Before hook"))
runner = CommandRunner(command_with_parser, runner_hooks=custom_hooks)
assert runner.runner_hooks == custom_hooks
assert runner.executor.hooks == custom_hooks
assert runner.runner_hooks._hooks[HookType.BEFORE]
def test_command_runner_initialization_with_all_bad_components(command_with_parser):
custom_options = "Not an OptionsManager"
custom_console = 23456
custom_hooks = "Not a HookManager"
with pytest.raises(
NotAFalyxError, match="options must be an instance of OptionsManager"
):
CommandRunner(
command_with_parser,
options=custom_options,
)
with pytest.raises(
NotAFalyxError, match="console must be an instance of rich.Console"
):
CommandRunner(
command_with_parser,
console=custom_console,
)
with pytest.raises(NotAFalyxError, match="hooks must be an instance of HookManager"):
CommandRunner(
command_with_parser,
runner_hooks=custom_hooks,
)
@pytest.mark.asyncio
async def test_command_runner_run(command_with_parser):
runner = CommandRunner(command_with_parser)
with falyx_console.capture() as capture:
result = await runner.run("--foo 42")
captured = Text.from_ansi(capture.get()).plain
assert result == "ok"
assert "Action executed with args:" in captured
assert "and kwargs:" in captured
assert "{'foo': 42}" in captured
falyx_console.clear()
with falyx_console.capture() as capture:
result = await runner.run(["--foo", "123"])
captured = Text.from_ansi(capture.get()).plain
assert result == "ok"
assert "Action executed with args:" in captured
assert "and kwargs:" in captured
assert "{'foo': 123}" in captured
@pytest.mark.asyncio
async def test_command_runner_run_with_failing_action(command_with_failing_action):
runner = CommandRunner(command_with_failing_action)
with pytest.raises(RuntimeError, match="boom"):
await runner.run("--foo 42")
with pytest.raises(FalyxError, match="boom"):
await runner.run("--foo 42", wrap_errors=True)
assert await runner.run("--foo 42", wrap_errors=False, raise_on_error=False) is None
@pytest.mark.asyncio
async def test_command_runner_debug_statement(command_with_parser, caplog):
caplog.set_level("DEBUG")
runner = CommandRunner(command_with_parser)
await runner.run("--foo 42")
assert (
"Executing command 'Test Command' with args=(), kwargs={'foo': 42}" in caplog.text
)
@pytest.mark.asyncio
async def test_command_runner_run_with_retries_non_action(
command_with_failing_action, caplog
):
runner = CommandRunner(command_with_failing_action)
with pytest.raises(RuntimeError, match="boom"):
await runner.run("--foo 42 --retries 2")
assert "Retry requested, but action is not an Action instance." in caplog.text
@pytest.mark.asyncio
async def test_command_runner_run_with_retries_with_action(
command_throwing_error, caplog
):
runner = CommandRunner(command_throwing_error)
with pytest.raises(asyncio.CancelledError, match="An error occurred in the action."):
await runner.run("Other")
with pytest.raises(ValueError, match="This is a ValueError."):
await runner.run("ValueError --retries 2")
assert "[throw_error] Retry attempt 1/2 failed due to 'ValueError'." in caplog.text
assert "[throw_error] Retry attempt 2/2 failed due to 'ValueError'." in caplog.text
assert "[throw_error] All 2 retries failed." in caplog.text
@pytest.mark.asyncio
async def test_command_runner_run_from_command_build_with_all_execution_options(
command_build_with_all_execution_options,
):
runner = CommandRunner.from_command(command_build_with_all_execution_options)
with falyx_console.capture() as capture:
result = await runner.run("--summary")
captured = Text.from_ansi(capture.get()).plain
assert result == "ok"
assert "Action executed with args:" in captured
assert "and kwargs:" in captured
assert "Execution History" in captured
with falyx_console.capture() as capture:
result = await runner.run("--summary", summary_last_result=True)
captured = Text.from_ansi(capture.get()).plain
assert result == "ok"
assert "Action executed with args:" in captured
assert "and kwargs:" in captured
assert "Command(key='T', description='Test Command' action=" in captured
assert "ok" in captured
with falyx_console.capture() as capture:
result = await runner.run("--summary", summary_last_result=False)
captured = Text.from_ansi(capture.get()).plain
assert result == "ok"
assert "Action executed with args:" in captured
assert "and kwargs:" in captured
assert "Execution History" in captured
@pytest.mark.asyncio
async def test_command_runner_from_command_bad_command():
with pytest.raises(NotAFalyxError, match="command must be an instance of Command"):
CommandRunner.from_command("Not a Command")
with pytest.raises(
NotAFalyxError, match="runner_hooks must be an instance of HookManager"
):
CommandRunner.from_command(
Command(
key="T",
description="Test Command",
action=ok_action,
),
runner_hooks="Not a HookManager",
)
@pytest.mark.asyncio
async def test_command_runner_build():
runner = CommandRunner.build(
key="T",
description="Test Command",
action=ok_action,
execution_options=["summary", "retry"],
)
assert isinstance(runner, CommandRunner)
with falyx_console.capture() as capture:
result = await runner.run("--summary --retries 2")
captured = Text.from_ansi(capture.get()).plain
assert result == "ok"
assert "Action executed with args:" in captured
assert "and kwargs:" in captured
assert "Execution History" in captured
@pytest.mark.asyncio
async def test_command_runner_build_with_bad_execution_options():
with pytest.raises(
ValueError,
match="Invalid ExecutionOption: 'invalid_option'. Must be one of:",
):
CommandRunner.build(
key="T",
description="Test Command",
action=ok_action,
execution_options=["summary", "invalid_option"],
)
@pytest.mark.asyncio
async def test_command_runner_build_with_bad_runner_hooks():
with pytest.raises(
NotAFalyxError, match="runner_hooks must be an instance of HookManager"
):
CommandRunner.build(
key="T",
description="Test Command",
action=ok_action,
runner_hooks="Not a HookManager",
)
@pytest.mark.asyncio
async def test_command_runner_uses_sys_argv(command_with_parser, monkeypatch):
runner = CommandRunner(command_with_parser)
test_args = ["program_name", "--foo", "42"]
monkeypatch.setattr(sys, "argv", test_args)
with falyx_console.capture() as capture:
result = await runner.run()
captured = Text.from_ansi(capture.get()).plain
assert result == "ok"
assert "Action executed with args:" in captured
assert "and kwargs:" in captured
assert "{'foo': 42}" in captured
@pytest.mark.asyncio
async def test_command_runner_cli(command_with_parser):
runner = CommandRunner(command_with_parser)
with falyx_console.capture() as capture:
await runner.cli("--foo 42")
captured = Text.from_ansi(capture.get()).plain
assert "Action executed with args:" in captured
assert "and kwargs:" in captured
assert "{'foo': 42}" in captured
@pytest.mark.asyncio
async def test_command_runnner_run_propogates_exeptions(command_throwing_error):
runner = CommandRunner(command_throwing_error)
with pytest.raises(QuitSignal, match="Quit signal triggered."):
await runner.run("QuitSignal")
with pytest.raises(BackSignal, match="Back signal triggered."):
await runner.run("BackSignal")
with pytest.raises(CancelSignal, match="Cancel signal triggered."):
await runner.run("CancelSignal")
with pytest.raises(ValueError, match="This is a ValueError."):
await runner.run("ValueError")
with pytest.raises(HelpSignal, match="Help signal triggered."):
await runner.run("HelpSignal")
with pytest.raises(asyncio.CancelledError, match="An error occurred in the action."):
await runner.run("Other")
with pytest.raises(
CommandArgumentError,
match=r"\[E\] Failed to parse arguments: No closing quotation",
):
await runner.run("Mismatched'")
@pytest.mark.asyncio
async def test_command_runner_cli_with_failing_action(command_with_failing_action):
runner = CommandRunner(command_with_failing_action)
with pytest.raises(SystemExit, match="1"):
await runner.cli("--foo 42")
with pytest.raises(SystemExit, match="2"):
await runner.cli("--foo 42 --bar 123")
with falyx_console.capture() as capture:
with pytest.raises(SystemExit, match="0"):
await runner.cli(["--help"])
captured = Text.from_ansi(capture.get()).plain
assert "usage: falyx T" in captured
assert "--foo" in captured
assert "summary" in captured
assert "retries" in captured
assert "A business argument." in captured
@pytest.mark.asyncio
async def test_command_runner_cli_exceptions(command_throwing_error):
runner = CommandRunner(command_throwing_error)
with falyx_console.capture() as capture:
with pytest.raises(SystemExit, match="0"):
await runner.cli(["--help"])
captured = Text.from_ansi(capture.get()).plain
assert "falyx E [--help]" in captured
assert "usage:" in captured
assert "positional:" in captured
assert "options:" in captured
assert "❌" not in captured
with falyx_console.capture() as capture:
with pytest.raises(SystemExit, match="2"):
await runner.cli(["--not-an-arg"])
captured = Text.from_ansi(capture.get()).plain
assert "falyx E [--help]" in captured
assert "usage:" in captured
assert "positional:" in captured
assert "options:" in captured
assert "❌" in captured
falyx_console.clear()
with falyx_console.capture() as capture:
with pytest.raises(SystemExit, match="1"):
await runner.cli(["FalyxError"])
captured = Text.from_ansi(capture.get()).plain
assert "This is a FalyxError." in captured
assert "❌ Error:" in captured
falyx_console.clear()
with falyx_console.capture() as capture:
with pytest.raises(SystemExit, match="130"):
await runner.cli(["QuitSignal"])
captured = Text.from_ansi(capture.get()).plain
assert "❌" not in captured
with falyx_console.capture() as capture:
with pytest.raises(SystemExit, match="1"):
await runner.cli(["BackSignal"])
captured = Text.from_ansi(capture.get()).plain
assert "❌" not in captured
with falyx_console.capture() as capture:
with pytest.raises(SystemExit, match="1"):
await runner.cli(["CancelSignal"])
captured = Text.from_ansi(capture.get()).plain
assert "❌" not in captured
with falyx_console.capture() as capture:
with pytest.raises(SystemExit, match="1"):
await runner.cli(["Other"])
captured = Text.from_ansi(capture.get()).plain
assert "❌" not in captured
@pytest.mark.asyncio
async def test_command_runner_cli_uses_sys_argv(command_with_parser, monkeypatch):
runner = CommandRunner(command_with_parser)
test_args = ["program_name", "--foo", "42"]
monkeypatch.setattr(sys, "argv", test_args)
with falyx_console.capture() as capture:
await runner.cli()
captured = Text.from_ansi(capture.get()).plain
assert "Action executed with args:" in captured
assert "and kwargs:" in captured
assert "{'foo': 42}" in captured

21
tests/test_validators/test_command_validator.py
View File

@@ -1,42 +1,49 @@
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from prompt_toolkit.document import Document
from prompt_toolkit.validation import ValidationError
from falyx.routing import RouteKind
from falyx.validators import CommandValidator
@pytest.mark.asyncio
async def test_command_validator_validates_command():
fake_falyx = AsyncMock()
fake_falyx.get_command.return_value = (False, object(), (), {}, {})
fake_route = SimpleNamespace()
fake_route.is_preview = False
fake_route.kind = RouteKind.NAMESPACE_HELP
fake_falyx.prepare_route.return_value = (fake_route, (), {}, {})
validator = CommandValidator(fake_falyx, "Invalid!")
await validator.validate_async(Document("valid"))
fake_falyx.get_command.assert_awaited_once()
fake_falyx.prepare_route.assert_awaited_once()
@pytest.mark.asyncio
async def test_command_validator_rejects_invalid_command():
fake_falyx = AsyncMock()
fake_falyx.get_command.return_value = (False, None, (), {}, {})
fake_falyx.prepare_route.return_value = (None, (), {}, {})
validator = CommandValidator(fake_falyx, "Invalid!")
with pytest.raises(ValidationError):
await validator.validate_async(Document("not_a_command"))
await validator.validate_async(Document(""))
with pytest.raises(ValidationError):
await validator.validate_async(Document(""))
await validator.validate_async(Document("not_a_command"))
@pytest.mark.asyncio
async def test_command_validator_is_preview():
fake_falyx = AsyncMock()
fake_falyx.get_command.return_value = (True, None, (), {}, {})
fake_route = SimpleNamespace()
fake_route.is_preview = True
fake_falyx.prepare_route.return_value = (fake_route, (), {}, {})
validator = CommandValidator(fake_falyx, "Invalid!")
await validator.validate_async(Document("?preview_command"))
fake_falyx.get_command.assert_awaited_once_with(
fake_falyx.prepare_route.assert_awaited_once_with(
"?preview_command", from_validate=True
)