feat(core): centralize command execution and add standalone command runner

- add CommandExecutor to unify shared command execution lifecycle
  across Falyx and standalone command execution
- add CommandRunner for running a single Command directly as a CLI
  or programmatic entrypoint
- add Command.build() factory and rename parse_args() to resolve_args()
  to clarify the parsing-to-execution boundary
- introduce ExecutionOption and wire execution-scoped flags into
  CommandArgumentParser and Command construction
- refactor Falyx to use FalyxParser/ParseResult and CommandExecutor
  instead of the older argparse-based flow and run_key path
- simplify __main__.py bootstrap by building a bootstrap Falyx instance
  directly and running flx.run()
- improve completer support for preview commands and unique-prefix
  command resolution
- default BottomBar toggle namespace to "default"
- expand module/class docstrings to reflect the new execution architecture
This commit is contained in:
2026-04-07 18:58:24 -04:00
parent 8ce0ffa18e
commit 5d8f3aa603
34 changed files with 3043 additions and 1419 deletions

View File

@@ -8,13 +8,12 @@ Licensed under the MIT License. See LICENSE file for details.
import asyncio
import os
import sys
from argparse import ArgumentParser, Namespace, _SubParsersAction
from pathlib import Path
from typing import Any
from falyx.config import loader
from falyx.falyx import Falyx
from falyx.parser import CommandArgumentParser, get_root_parser, get_subparsers
from falyx.parser import CommandArgumentParser
def find_falyx_config() -> Path | None:
@@ -49,71 +48,39 @@ def init_config(parser: CommandArgumentParser) -> None:
)
def init_callback(args: Namespace) -> None:
"""Callback for the init command."""
if args.command == "init":
from falyx.init import init_project
def build_bootstrap_falyx() -> Falyx:
from falyx.init import init_global, init_project
init_project(args.name)
elif args.command == "init_global":
from falyx.init import init_global
flx = Falyx()
init_global()
def get_parsers() -> tuple[ArgumentParser, _SubParsersAction]:
root_parser: ArgumentParser = get_root_parser()
subparsers = get_subparsers(root_parser)
init_parser = subparsers.add_parser(
"init",
help="Initialize a new Falyx project",
description="Create a new Falyx project with mock configuration files.",
epilog="If no name is provided, the current directory will be used.",
flx.add_command(
"I",
"Initialize a new Falyx project",
init_project,
aliases=["init"],
argument_config=init_config,
help_epilog="If no name is provided, the current directory will be used.",
)
init_parser.add_argument(
"name",
type=str,
help="Name of the new Falyx project",
default=".",
nargs="?",
flx.add_command(
"G",
"Initialize Falyx global configuration",
init_global,
aliases=["init-global"],
help_text="Create a global Falyx configuration at ~/.config/falyx/.",
)
subparsers.add_parser(
"init-global",
help="Initialize Falyx global configuration",
description="Create a global Falyx configuration at ~/.config/falyx/.",
)
return root_parser, subparsers
return flx
def build_falyx() -> Falyx:
bootstrap_path = bootstrap()
if bootstrap_path:
return loader(bootstrap_path)
return build_bootstrap_falyx()
def main() -> Any:
bootstrap_path = bootstrap()
if not bootstrap_path:
from falyx.init import init_global, init_project
flx: Falyx = Falyx()
flx.add_command(
"I",
"Initialize a new Falyx project",
init_project,
aliases=["init"],
argument_config=init_config,
help_epilog="If no name is provided, the current directory will be used.",
)
flx.add_command(
"G",
"Initialize Falyx global configuration",
init_global,
aliases=["init-global"],
help_text="Create a global Falyx configuration at ~/.config/falyx/.",
)
else:
flx = loader(bootstrap_path)
root_parser, subparsers = get_parsers()
return asyncio.run(
flx.run(root_parser=root_parser, subparsers=subparsers, callback=init_callback)
)
flx = build_falyx()
return asyncio.run(flx.run())
if __name__ == "__main__":

View File

@@ -202,7 +202,7 @@ class BottomBar:
label: str,
options: OptionsManager,
option_name: str,
namespace_name: str = "cli_args",
namespace_name: str = "default",
fg: str = OneColors.BLACK,
bg_on: str = OneColors.GREEN,
bg_off: str = OneColors.DARK_RED,

View File

@@ -1,19 +1,43 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""
Defines the Command class for Falyx CLI.
"""Command abstraction for the Falyx CLI framework.
Commands are callable units representing a menu option or CLI task,
wrapping either a BaseAction or a simple function. They provide:
This module defines the `Command` class, which represents a single executable
unit exposed to users via CLI or interactive menu interfaces.
- Hook lifecycle (before, on_success, on_error, after, on_teardown)
A `Command` acts as a bridge between:
- User input (parsed via CommandArgumentParser)
- Execution logic (encapsulated in Action / BaseAction)
- Runtime configuration (OptionsManager)
- Lifecycle hooks (HookManager)
Core Responsibilities:
- Define command identity (key, aliases, description)
- Bind an executable action or workflow
- Configure argument parsing via CommandArgumentParser
- Separate execution arguments (e.g. retries, confirm) from action arguments
- Manage lifecycle hooks for command-level execution
- Provide help, usage, and preview interfaces
- Execution timing and duration tracking
- Retry logic (single action or recursively through action trees)
- Confirmation prompts and spinner integration
- Result capturing and summary logging
- Rich-based preview for CLI display
Every Command is self-contained, configurable, and plays a critical role
in building robust interactive menus.
Execution Model:
1. CLI input is routed via FalyxParser into a resolved Command
2. Arguments are parsed via CommandArgumentParser
3. Parsed values are split into:
- positional args
- keyword args
- execution args (e.g. retries, summary)
4. Execution occurs via the bound Action with lifecycle hooks applied
5. Results and context are tracked via ExecutionContext / ExecutionRegistry
Key Concepts:
- Commands are *user-facing entrypoints*, not execution units themselves
- Execution is always delegated to an underlying Action or callable
- Argument parsing is declarative and optional
- Execution options are handled separately from business logic inputs
This module defines the primary abstraction used by Falyx to expose structured,
composable workflows as CLI commands.
"""
from __future__ import annotations
@@ -29,8 +53,11 @@ 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.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
@@ -46,67 +73,100 @@ from falyx.utils import ensure_async
class Command(BaseModel):
"""
Represents a selectable command in a Falyx menu system.
"""Represents a user-invokable command in Falyx.
A Command wraps an executable action (function, coroutine, or BaseAction)
and enhances it with:
A `Command` encapsulates all metadata, parsing logic, and execution behavior
required to expose a callable workflow through the Falyx CLI or interactive
menu system.
- Lifecycle hooks (before, success, error, after, teardown)
- Retry support (single action or recursive for chained/grouped actions)
- Confirmation prompts for safe execution
- Spinner visuals during execution
- Tagging for categorization and filtering
- Rich-based CLI previews
It is responsible for:
- Identifying the command via key and aliases
- Binding an executable Action or callable
- Parsing user-provided arguments
- Managing execution configuration (retries, confirmation, etc.)
- Integrating with lifecycle hooks and execution context
Architecture:
- Parsing is delegated to CommandArgumentParser
- Execution is delegated to BaseAction / Action
- Runtime configuration is managed via OptionsManager
- Lifecycle hooks are managed via HookManager
Argument Handling:
- Supports positional and keyword arguments via CommandArgumentParser
- Separates execution-specific options (e.g. retries, confirm flags)
from action arguments
- Returns structured `(args, kwargs, execution_args)` for execution
Execution Behavior:
- Callable via `await command(*args, **kwargs)`
- Applies lifecycle hooks:
before → on_success/on_error → after → on_teardown
- Supports preview mode for dry-run introspection
- Supports retry policies and confirmation flows
- Result tracking and summary reporting
Commands are built to be flexible yet robust, enabling dynamic CLI workflows
without sacrificing control or reliability.
Help & Introspection:
- Provides usage, help text, and TLDR examples
- Supports both CLI help and interactive menu rendering
- Can expose simplified or full help signatures
Attributes:
key (str): Primary trigger key for the command.
Args:
key (str): Primary identifier used to invoke the command.
description (str): Short description for the menu display.
hidden (bool): Toggles visibility in the menu.
aliases (list[str]): Alternate keys or phrases.
action (BaseAction | Callable): The executable logic.
args (tuple): Static positional arguments.
kwargs (dict): Static keyword arguments.
help_text (str): Additional help or guidance text.
style (str): Rich style for description.
confirm (bool): Whether to require confirmation before executing.
confirm_message (str): Custom confirmation prompt.
preview_before_confirm (bool): Whether to preview before confirming.
spinner (bool): Whether to show a spinner during execution.
spinner_message (str): Spinner text message.
spinner_type (str): Spinner style (e.g., dots, line, etc.).
spinner_style (str): Color or style of the spinner.
spinner_speed (float): Speed of the spinner animation.
hooks (HookManager): Hook manager for lifecycle events.
retry (bool): Enable retry on failure.
retry_all (bool): Enable retry across chained or grouped actions.
retry_policy (RetryPolicy): Retry behavior configuration.
tags (list[str]): Organizational tags for the command.
logging_hooks (bool): Whether to attach logging hooks automatically.
options_manager (OptionsManager): Manages global command-line options.
arg_parser (CommandArgumentParser): Parses command arguments.
arguments (list[dict[str, Any]]): Argument definitions for the command.
argument_config (Callable[[CommandArgumentParser], None] | None): Function to configure arguments
for the command parser.
custom_parser (ArgParserProtocol | None): Custom argument parser.
custom_help (Callable[[], str | None] | None): Custom help message generator.
auto_args (bool): Automatically infer arguments from the action.
arg_metadata (dict[str, str | dict[str, Any]]): Metadata for arguments,
such as help text or choices.
simple_help_signature (bool): Whether to use a simplified help signature.
ignore_in_history (bool): Whether to ignore this command in execution history last result.
program: (str | None): The parent program name.
action (BaseAction | Callable[..., Any]):
Execution logic for the command.
args (tuple, optional): Static positional arguments.
kwargs (dict[str, Any], optional): Static keyword arguments.
hidden (bool): Whether to hide the command from menus.
aliases (list[str], optional): Alternate names for invocation.
help_text (str): Help description shown in CLI/menu.
help_epilog (str): Additional help content.
style (str): Rich style used for rendering.
confirm (bool): Whether confirmation is required before execution.
confirm_message (str): Confirmation prompt text.
preview_before_confirm (bool): Whether to preview before confirmation.
spinner (bool): Enable spinner during execution.
spinner_message (str): Spinner message text.
spinner_type (str): Rich Spinner animation type (e.g., dots, line, etc.).
spinner_style (str): Rich style for the spinner.
spinner_speed (float): Spinner speed multiplier.
hooks (HookManager | None): Hook manager for lifecycle events.
tags (list[str], optional): Tags for grouping and filtering.
logging_hooks (bool): Enable debug logging hooks.
retry (bool): Enable retry behavior.
retry_all (bool): Apply retry to all nested actions.
retry_policy (RetryPolicy | None): Retry configuration.
arg_parser (CommandArgumentParser | None):
Custom argument parser instance.
execution_options (frozenset[ExecutionOption], optional):
Enabled execution-level options.
arguments (list[dict[str, Any]], optional):
Declarative argument definitions.
argument_config (Callable[[CommandArgumentParser], None] | None):
Callback to configure parser.
custom_parser (ArgParserProtocol | None):
Override parser logic entirely.
custom_help (Callable[[], str | None] | None):
Override help 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.
ignore_in_history (bool):
Ignore command for `last_result` in execution history.
options_manager (OptionsManager | None):
Shared options manager instance.
program (str | None): The parent program name.
Methods:
__call__(): Executes the command, respecting hooks and retries.
preview(): Rich tree preview of the command.
confirmation_prompt(): Formatted prompt for confirmation.
result: Property exposing the last result.
log_summary(): Summarizes execution details to the console.
Raises:
CommandArgumentError: If argument parsing fails.
InvalidActionError: If action is not callable or invalid.
FalyxError: If command configuration is invalid.
Notes:
- Commands are lightweight wrappers; execution logic belongs in Actions
- Argument parsing and execution are intentionally decoupled
- Commands are case-insensitive and support alias resolution
"""
key: str
@@ -135,6 +195,7 @@ class Command(BaseModel):
logging_hooks: bool = False
options_manager: OptionsManager = Field(default_factory=OptionsManager)
arg_parser: CommandArgumentParser | None = None
execution_options: frozenset[ExecutionOption] = Field(default_factory=frozenset)
arguments: list[dict[str, Any]] = Field(default_factory=list)
argument_config: Callable[[CommandArgumentParser], None] | None = None
custom_parser: ArgParserProtocol | None = None
@@ -149,9 +210,53 @@ class Command(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
async def parse_args(
async def resolve_args(
self, raw_args: list[str] | str, from_validate: bool = False
) -> tuple[tuple, dict]:
) -> tuple[tuple, dict, dict]:
"""Parse CLI arguments into execution-ready components.
This method delegates argument parsing to the configured
CommandArgumentParser (if present) and normalizes the result into three
distinct groups used during execution:
- positional arguments (`args`)
- keyword arguments (`kwargs`)
- execution arguments (`execution_args`)
Execution arguments represent runtime configuration (e.g. retries,
confirmation flags, summary output) and are handled separately from the
action's business logic inputs.
Behavior:
- If an argument parser is defined, uses `CommandArgumentParser.parse_args_split()`
to resolve and type-coerce all inputs.
- If no parser is defined, returns empty args and kwargs.
- Supports validation mode (`from_validate=True`) for interactive input,
deferring certain errors and resolver execution where applicable.
- Handles help/preview signals raised during parsing.
Args:
args (list[str] | None): CLI-style argument tokens.
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.
Returns:
tuple:
- tuple[Any, ...]: Positional arguments for execution.
- dict[str, Any]: Keyword arguments for execution.
- dict[str, Any]: Execution-specific arguments (e.g. retries,
confirm flags, summary).
Raises:
CommandArgumentError: If argument parsing or validation fails.
HelpSignal: If help or TLDR output is triggered during parsing.
Notes:
- Execution arguments are not passed to the underlying Action.
- This method is the canonical boundary between CLI parsing and
execution semantics.
"""
if callable(self.custom_parser):
if isinstance(raw_args, str):
try:
@@ -162,7 +267,7 @@ class Command(BaseModel):
self.key,
raw_args,
)
return ((), {})
return ((), {}, {})
return self.custom_parser(raw_args)
if isinstance(raw_args, str):
@@ -174,13 +279,13 @@ class Command(BaseModel):
self.key,
raw_args,
)
return ((), {})
return ((), {}, {})
if not isinstance(self.arg_parser, CommandArgumentParser):
logger.warning(
"[Command:%s] No argument parser configured, using default parsing.",
self.key,
)
return ((), {})
return ((), {}, {})
return await self.arg_parser.parse_args_split(
raw_args, from_validate=from_validate
)
@@ -249,6 +354,12 @@ class Command(BaseModel):
for arg_def in self.get_argument_definitions():
self.arg_parser.add_argument(*arg_def.pop("flags"), **arg_def)
if isinstance(self.arg_parser, CommandArgumentParser) and self.execution_options:
self.arg_parser.enable_execution_options(self.execution_options)
if isinstance(self.arg_parser, CommandArgumentParser):
self.arg_parser.set_options_manager(self.options_manager)
if self.ignore_in_history and isinstance(self.action, BaseAction):
self.action.ignore_in_history = True
@@ -258,9 +369,41 @@ class Command(BaseModel):
self.action.set_options_manager(self.options_manager)
async def __call__(self, *args, **kwargs) -> Any:
"""
Run the action with full hook lifecycle, timing, error handling,
confirmation prompts, preview, and spinner integration.
"""Execute the command's underlying action with lifecycle management.
This method invokes the bound action (BaseAction or callable) using the
provided arguments while applying the full Falyx execution lifecycle.
Execution Flow:
1. Create an ExecutionContext for tracking inputs, results, and timing
2. Trigger `before` hooks
3. Execute the underlying action
4. Trigger `on_success` or `on_error` hooks
5. Trigger `after` and `on_teardown` hooks
6. Record execution via ExecutionRegistry
Behavior:
- Supports both synchronous and asynchronous actions
- Applies retry policies if configured
- Integrates with confirmation and execution options via OptionsManager
- Propagates exceptions unless recovered by hooks (e.g. retry handlers)
Args:
*args (Any): Positional arguments passed to the action.
**kwargs (Any): Keyword arguments passed to the action.
Returns:
Any: Result returned by the underlying action.
Raises:
Exception: Propagates execution errors unless handled by hooks.
Notes:
- This method does not perform argument parsing; inputs are assumed
to be pre-processed via `resolve_args`.
- Execution options (e.g. retries, confirm) are applied externally
via Falyx in OptionsManager before invocation.
- Lifecycle hooks are always executed, even in failure cases.
"""
self._inject_options_manager()
combined_args = args + self.args
@@ -341,15 +484,38 @@ class Command(BaseModel):
@property
def help_signature(self) -> tuple[str, str, str]:
"""Generate a help signature for the command."""
is_cli_mode = self.options_manager.get("mode") in {
FalyxMode.RUN,
FalyxMode.PREVIEW,
FalyxMode.RUN_ALL,
FalyxMode.HELP,
}
"""Return a formatted help signature for display.
program = f"{self.program} run " if is_cli_mode else ""
This property provides the core information used to render command help
in both CLI and interactive menu modes.
The signature consists of:
- usage: A formatted usage string (including arguments if defined)
- description: A short description of the command
- tag: Optional tag or category label (if applicable)
Behavior:
- If a CommandArgumentParser is present, delegates usage generation to
the parser (`get_usage()`).
- Otherwise, constructs a minimal usage string from the command key.
- Honors `simple_help_signature` to produce a condensed representation
(e.g. omitting argument details).
- Applies styling appropriate for Rich rendering.
Returns:
tuple:
- str: Usage string (e.g. "falyx D | deploy [--help] region")
- str: Command description
- str | None: 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()}"
@@ -416,3 +582,219 @@ class Command(BaseModel):
f"Command(key='{self.key}', description='{self.description}' "
f"action='{self.action}')"
)
@classmethod
def build(
cls,
key: str,
description: str,
action: BaseAction | Callable[..., Any],
*,
args: tuple = (),
kwargs: dict[str, Any] | None = None,
hidden: bool = False,
aliases: list[str] | None = None,
help_text: str = "",
help_epilog: str = "",
style: str = OneColors.WHITE,
confirm: bool = False,
confirm_message: str = "Are you sure?",
preview_before_confirm: bool = True,
spinner: bool = False,
spinner_message: str = "Processing...",
spinner_type: str = "dots",
spinner_style: str = OneColors.CYAN,
spinner_speed: float = 1.0,
options_manager: OptionsManager | None = None,
hooks: HookManager | None = None,
before_hooks: list[Callable] | None = None,
success_hooks: list[Callable] | None = None,
error_hooks: list[Callable] | None = None,
after_hooks: list[Callable] | None = None,
teardown_hooks: list[Callable] | None = None,
tags: list[str] | None = None,
logging_hooks: bool = False,
retry: bool = False,
retry_all: bool = False,
retry_policy: RetryPolicy | None = None,
arg_parser: CommandArgumentParser | None = None,
arguments: list[dict[str, Any]] | None = None,
argument_config: Callable[[CommandArgumentParser], None] | None = None,
execution_options: list[ExecutionOption | str] | None = None,
custom_parser: ArgParserProtocol | None = None,
custom_help: Callable[[], str | None] | None = None,
auto_args: bool = True,
arg_metadata: dict[str, str | dict[str, Any]] | None = None,
simple_help_signature: bool = False,
ignore_in_history: bool = False,
program: str | None = None,
) -> Command:
"""Build and configure a `Command` instance from high-level constructor inputs.
This factory centralizes command construction so callers such as `Falyx` and
`CommandRunner` can create fully configured commands through one consistent
path. It normalizes optional inputs, validates selected objects, converts
execution options into their canonical internal form, and registers any
requested command-level hooks.
In addition to instantiating the `Command`, this method can:
- validate and attach an explicit `CommandArgumentParser`
- normalize execution options into a `frozenset[ExecutionOption]`
- ensure a shared `OptionsManager` is available
- attach a custom `HookManager`
- register lifecycle hooks for the command
- register spinner hooks when spinner support is enabled
Args:
key (str): Primary identifier used to invoke the command.
description (str): Short description of the command.
action (BaseAction | Callable[..., Any]): Underlying execution logic for
the command.
args (tuple): Static positional arguments applied to every execution.
kwargs (dict[str, Any] | None): Static keyword arguments applied to every
execution.
hidden (bool): Whether the command should be hidden from menu displays.
aliases (list[str] | None): Optional alternate names for invocation.
help_text (str): Help text shown in command help output.
help_epilog (str): Additional help text shown after the main help body.
style (str): Rich style used when rendering the command.
confirm (bool): Whether confirmation is required before execution.
confirm_message (str): Confirmation prompt text.
preview_before_confirm (bool): Whether to preview before confirmation.
spinner (bool): Whether to enable spinner lifecycle hooks.
spinner_message (str): Spinner message text.
spinner_type (str): Spinner animation type.
spinner_style (str): Spinner style.
spinner_speed (float): Spinner speed multiplier.
options_manager (OptionsManager | None): Shared options manager for the
command and its parser.
hooks (HookManager | None): Optional hook manager to assign directly to the
command.
before_hooks (list[Callable] | None): Hooks registered for the `BEFORE`
lifecycle stage.
success_hooks (list[Callable] | None): Hooks registered for the
`ON_SUCCESS` lifecycle stage.
error_hooks (list[Callable] | None): Hooks registered for the `ON_ERROR`
lifecycle stage.
after_hooks (list[Callable] | None): Hooks registered for the `AFTER`
lifecycle stage.
teardown_hooks (list[Callable] | None): Hooks registered for the
`ON_TEARDOWN` lifecycle stage.
tags (list[str] | None): Optional tags used for grouping and filtering.
logging_hooks (bool): Whether to enable debug hook logging.
retry (bool): Whether retry behavior is enabled.
retry_all (bool): Whether retry behavior should be applied recursively.
retry_policy (RetryPolicy | None): Retry configuration for the command.
arg_parser (CommandArgumentParser | None): Optional explicit argument
parser instance.
arguments (list[dict[str, Any]] | None): Declarative argument
definitions for the command parser.
argument_config (Callable[[CommandArgumentParser], None] | None): Callback
used to configure the argument parser.
execution_options (list[ExecutionOption | str] | None): Execution-level
options to enable for the command.
custom_parser (ArgParserProtocol | None): Optional custom parser
implementation that overrides normal parser behavior.
custom_help (Callable[[], str | None] | None): Optional custom help
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
used during argument inference.
simple_help_signature (bool): Whether to use a simplified help signature.
ignore_in_history (bool): Whether to exclude the command from execution
history tracking.
program (str | None): Parent program name used in help rendering.
Returns:
Command: A fully configured `Command` instance.
Raises:
NotAFalyxError: If `arg_parser` is provided but is not a
`CommandArgumentParser` instance, or if `hooks` is provided but is not
a `HookManager` instance.
Notes:
- Execution options supplied as strings are converted to
`ExecutionOption` enum values before the command is created.
- If no `options_manager` is provided, a new `OptionsManager` is created.
- Spinner hooks are registered at build time when `spinner=True`.
- 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 execution_options:
parsed_execution_options = frozenset(
ExecutionOption(option) if isinstance(option, str) else option
for option in execution_options
)
else:
parsed_execution_options = frozenset()
options_manager = options_manager or OptionsManager()
command = Command(
key=key,
description=description,
action=action,
args=args,
kwargs=kwargs if kwargs else {},
hidden=hidden,
aliases=aliases if aliases else [],
help_text=help_text,
help_epilog=help_epilog,
style=style,
confirm=confirm,
confirm_message=confirm_message,
preview_before_confirm=preview_before_confirm,
spinner=spinner,
spinner_message=spinner_message,
spinner_type=spinner_type,
spinner_style=spinner_style,
spinner_speed=spinner_speed,
tags=tags if tags else [],
logging_hooks=logging_hooks,
retry=retry,
retry_all=retry_all,
retry_policy=retry_policy or RetryPolicy(),
options_manager=options_manager,
arg_parser=arg_parser,
execution_options=parsed_execution_options,
arguments=arguments or [],
argument_config=argument_config,
custom_parser=custom_parser,
custom_help=custom_help,
auto_args=auto_args,
arg_metadata=arg_metadata or {},
simple_help_signature=simple_help_signature,
ignore_in_history=ignore_in_history,
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 []:
command.hooks.register(HookType.ON_SUCCESS, hook)
for hook in error_hooks or []:
command.hooks.register(HookType.ON_ERROR, hook)
for hook in after_hooks or []:
command.hooks.register(HookType.AFTER, hook)
for hook in teardown_hooks or []:
command.hooks.register(HookType.ON_TEARDOWN, hook)
if spinner:
command.hooks.register(HookType.BEFORE, spinner_before_hook)
command.hooks.register(HookType.ON_TEARDOWN, spinner_teardown_hook)
return command

334
falyx/command_executor.py Normal file
View File

@@ -0,0 +1,334 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""Shared command execution engine for the Falyx CLI framework.
This module defines `CommandExecutor`, a low-level execution service responsible
for running already-resolved `Command` objects with a consistent outer lifecycle.
`CommandExecutor` sits between higher-level orchestration layers (such as
`Falyx.execute_command()` or `CommandRunner.run()`) and the command itself.
It does not perform command lookup or argument parsing. Instead, it accepts a
resolved `Command` plus prepared `args`, `kwargs`, and `execution_args`, then
applies executor-level behavior around the command invocation.
Responsibilities:
- Apply execution-scoped runtime overrides such as confirmation flags
- Apply retry overrides from execution arguments
- Trigger executor-level lifecycle hooks
- Create and manage an outer `ExecutionContext`
- Delegate actual invocation to the resolved `Command`
- Handle interruption and failure policies
- Optionally print execution summaries via `ExecutionRegistry`
Execution Model:
1. A command is resolved and its arguments are prepared elsewhere.
2. Retry and execution-option overrides are derived from `execution_args`.
3. An outer `ExecutionContext` is created for executor-level tracking.
4. Executor hooks are triggered around the command invocation.
5. The command is executed inside an `OptionsManager.override_namespace()`
context for scoped runtime overrides.
6. Errors are either surfaced, wrapped, or rendered depending on the
configured execution policy.
7. Optional summary output is emitted after execution completes.
Design Notes:
- `CommandExecutor` is intentionally narrower in scope than `Falyx`.
It does not resolve commands, parse raw CLI text, or manage menu state.
- `Command` still owns command-local behavior such as confirmation,
command hooks, and delegation to the underlying `Action`.
- This module exists to centralize shared execution behavior and reduce
duplication across Falyx runtime entrypoints.
Typical Usage:
executor = CommandExecutor(options=options, hooks=hooks, console=console)
result = await executor.execute(
command=command,
args=args,
kwargs=kwargs,
execution_args=execution_args,
)
"""
from __future__ import annotations
from typing import Any
from rich.console import Console
from falyx.action import Action
from falyx.command import Command
from falyx.context import ExecutionContext
from falyx.exceptions import FalyxError
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import HookManager, HookType
from falyx.logger import logger
from falyx.options_manager import OptionsManager
from falyx.themes import OneColors
class CommandExecutor:
"""Execute resolved Falyx commands with shared outer lifecycle handling.
`CommandExecutor` provides a reusable execution service for running a
`Command` after command resolution and argument parsing have already been
completed.
This class is intended to be shared by higher-level entrypoints such as
`Falyx` and `CommandRunner`. It centralizes the outer execution flow so
command execution semantics remain consistent across menu-driven and
programmatic use cases.
Responsibilities:
- Apply retry overrides from execution arguments
- Apply scoped runtime overrides using `OptionsManager`
- Trigger executor-level hooks before and after command execution
- Create and manage an executor-level `ExecutionContext`
- Render execution errors to the configured console
- Control whether errors are raised, wrapped, or suppressed
- Emit optional execution summaries
Attributes:
options (OptionsManager): Shared options manager used to apply scoped
execution overrides.
hooks (HookManager): Hook manager for executor-level lifecycle hooks.
console (Console): Rich console used for user-facing error output.
"""
def __init__(
self,
*,
options: OptionsManager,
hooks: HookManager,
console: Console,
) -> None:
self.options = options
self.hooks = hooks
self.console = console
def _debug_hooks(self, command: Command) -> None:
"""Log executor-level and command-level hook registrations for debugging.
This helper is used to surface the currently registered hooks on both the
executor and the resolved command before execution begins.
Args:
command (Command): The command about to be executed.
"""
logger.debug("Executor hooks:\n%s", str(self.hooks))
logger.debug("['%s'] hooks:\n%s", command.key, str(command.hooks))
def _apply_retry_overrides(
self,
command: Command,
execution_args: dict[str, Any],
) -> None:
"""Apply retry-related execution overrides to the command.
This method inspects execution-level retry options and updates the
command's retry policy in place when overrides are provided. If the
command's underlying action is an `Action`, the updated retry policy is
propagated to that action as well.
Args:
command (Command): The command whose retry policy may be updated.
execution_args (dict[str, Any]): Execution-level arguments that may
contain retry overrides such as `retries`, `retry_delay`, and
`retry_backoff`.
Notes:
- If no retry-related overrides are provided, this method does nothing.
- If the command action is not an `Action`, a warning is logged and the
command-level retry policy is updated without propagating it further.
"""
retries = execution_args.get("retries", 0)
retry_delay = execution_args.get("retry_delay", 0.0)
retry_backoff = execution_args.get("retry_backoff", 0.0)
logger.debug(
"[_apply_retry_overrides]: retries=%s, retry_delay=%s, retry_backoff=%s",
retries,
retry_delay,
retry_backoff,
)
if not retries and not retry_delay and not retry_backoff:
return
command.retry_policy.enabled = True
if retries:
command.retry_policy.max_retries = retries
if retry_delay:
command.retry_policy.delay = retry_delay
if retry_backoff:
command.retry_policy.backoff = retry_backoff
if isinstance(command.action, Action):
command.action.set_retry_policy(command.retry_policy)
else:
logger.warning(
"[%s] Retry requested, but action is not an Action instance.",
command.description,
)
def _execution_option_overrides(
self,
execution_args: dict[str, Any],
) -> dict[str, Any]:
"""Build scoped option overrides from execution arguments.
This method extracts execution-only runtime flags that should be applied
through the `OptionsManager` during command execution.
Args:
execution_args (dict[str, Any]): Execution-level arguments returned
from command argument resolution.
Returns:
dict[str, Any]: Mapping of option names to temporary execution-scoped
override values.
"""
return {
"force_confirm": execution_args.get("force_confirm", False),
"skip_confirm": execution_args.get("skip_confirm", False),
}
async def _handle_action_error(
self, selected_command: Command, error: Exception
) -> None:
"""Render and log a command execution error.
This helper logs the full exception details for debugging and prints a
user-facing error message to the configured console.
Args:
selected_command (Command): The command that failed.
error (Exception): The exception raised during command execution.
"""
logger.debug(
"[%s] '%s' failed with error: %s",
selected_command.key,
selected_command.description,
error,
exc_info=True,
)
self.console.print(
f"[{OneColors.DARK_RED}]An error occurred while executing "
f"{selected_command.description}:[/] {error}"
)
async def execute(
self,
*,
command: Command,
args: tuple,
kwargs: dict[str, Any],
execution_args: dict[str, Any],
raise_on_error: bool = True,
wrap_errors: bool = False,
summary_last_result: bool = False,
) -> Any:
"""Execute a resolved command with executor-level lifecycle management.
This method is the primary entrypoint of `CommandExecutor`. It accepts an
already-resolved `Command` and its prepared execution inputs, then applies
shared outer execution behavior around the command invocation.
Execution Flow:
1. Log currently registered hooks for debugging.
2. Apply retry overrides from `execution_args`.
3. Derive scoped runtime overrides for the execution namespace.
4. Create and start an outer `ExecutionContext`.
5. Trigger executor-level `BEFORE` hooks.
6. Execute the command inside an execution-scoped options override
context.
7. Trigger executor-level `SUCCESS` or `ERROR` hooks.
8. Trigger `AFTER` and `ON_TEARDOWN` hooks.
9. Optionally print an execution summary.
Args:
command (Command): The resolved command to execute.
args (tuple): Positional arguments to pass to the command.
kwargs (dict[str, Any]): Keyword arguments to pass to the command.
execution_args (dict[str, Any]): Execution-only arguments that affect
runtime behavior, such as retry or confirmation overrides.
raise_on_error (bool): Whether execution errors should be re-raised
after handling.
wrap_errors (bool): Whether handled errors should be wrapped in a
`FalyxError` before being raised.
summary_last_result (bool): Whether summary output should only have the
last recorded result when summary reporting is enabled.
Returns:
Any: The result returned by the command, or any recovered result
attached to the execution context.
Raises:
KeyboardInterrupt: If execution is interrupted by the user and
`raise_on_error` is True and `wrap_errors` is False.
EOFError: If execution receives EOF interruption and `raise_on_error`
is True and `wrap_errors` is False.
FalyxError: If `wrap_errors` is True and execution is interrupted or
fails.
Exception: Re-raises the underlying execution error when
`raise_on_error` is True and `wrap_errors` is False.
Notes:
- This method assumes the command has already been resolved and its
arguments have already been parsed.
- Command-local behavior, such as confirmation prompts and command hook
execution, remains the responsibility of `Command.__call__()`.
- Summary output is only emitted when the `summary` execution option is
present in `execution_args`.
"""
self._debug_hooks(command)
self._apply_retry_overrides(command, execution_args)
overrides = self._execution_option_overrides(execution_args)
context = ExecutionContext(
name=command.description,
args=args,
kwargs=kwargs,
action=command,
)
logger.info(
"[execute] Starting execution of '%s' with args: %s, kwargs: %s",
command.description,
args,
kwargs,
)
context.start_timer()
try:
await self.hooks.trigger(HookType.BEFORE, context)
with self.options.override_namespace(
overrides=overrides,
namespace_name="execution",
):
result = await command(*args, **kwargs)
context.result = result
await self.hooks.trigger(HookType.ON_SUCCESS, context)
except (KeyboardInterrupt, EOFError) as error:
logger.info(
"[execute] '%s' interrupted by user.",
command.description,
)
if wrap_errors:
raise FalyxError(
f"[execute] ⚠️ '{command.description}' interrupted by user."
) from error
if raise_on_error:
raise error
except Exception as error:
context.exception = error
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
if raise_on_error:
raise error
finally:
context.stop_timer()
await self.hooks.trigger(HookType.AFTER, context)
await self.hooks.trigger(HookType.ON_TEARDOWN, context)
if execution_args.get("summary") and summary_last_result:
er.summary(last_result=True)
elif execution_args.get("summary"):
er.summary()
return context.result

467
falyx/command_runner.py Normal file
View File

@@ -0,0 +1,467 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""Standalone command runner for the Falyx CLI framework.
This module defines `CommandRunner`, a developer-facing convenience wrapper for
executing a single `Command` outside the full `Falyx` runtime.
`CommandRunner` is designed for programmatic and standalone command execution
where command lookup, menu interaction, and root CLI parsing are not needed.
It provides a small, focused API that:
- owns a single `Command`
- ensures the command and parser share a consistent `OptionsManager`
- delegates shared execution behavior to `CommandExecutor`
- supports both wrapping an existing `Command` and building one from raw
constructor-style arguments
Responsibilities:
- Hold a single resolved `Command` for repeated execution
- Normalize runtime dependencies such as `OptionsManager`, `HookManager`,
and `Console`
- Resolve command arguments from raw argv-style input
- Delegate execution to `CommandExecutor` for shared outer lifecycle
handling
Design Notes:
- `CommandRunner` is intentionally narrower than `Falyx`.
It does not resolve commands by name, render menus, or manage built-ins.
- `CommandExecutor` remains the shared execution core.
`CommandRunner` exists as a convenience layer for developer-facing and
standalone use cases.
- `Command` still owns command-local behavior such as confirmation,
command hook execution, and delegation to the underlying `Action`.
Typical Usage:
runner = CommandRunner.from_command(existing_command)
result = await runner.run(["--region", "us-east"])
#!/usr/bin/env python
import asyncio
runner = CommandRunner.build(
key="D",
description="Deploy",
action=deploy,
)
result = asyncio.run(runner.cli())
$ ./deploy.py --region us-east
"""
from __future__ import annotations
import asyncio
import sys
from typing import Any, Callable
from rich.console import Console
from falyx.action import BaseAction
from falyx.command import Command
from falyx.command_executor import CommandExecutor
from falyx.console import console as falyx_console
from falyx.exceptions import CommandArgumentError, FalyxError, NotAFalyxError
from falyx.execution_option import ExecutionOption
from falyx.hook_manager import HookManager
from falyx.logger import logger
from falyx.options_manager import OptionsManager
from falyx.parser.command_argument_parser import CommandArgumentParser
from falyx.protocols import ArgParserProtocol
from falyx.retry import RetryPolicy
from falyx.signals import BackSignal, CancelSignal, HelpSignal, QuitSignal
from falyx.themes import OneColors
class CommandRunner:
"""Run a single Falyx command outside the full Falyx application runtime.
`CommandRunner` is a lightweight wrapper around a single `Command` plus a
`CommandExecutor`. It is intended for standalone execution, testing, and
developer-facing programmatic usage where command resolution has already
happened or is unnecessary.
This class is responsible for:
- storing the bound `Command`
- providing a shared `OptionsManager` to the command and its parser
- exposing a simple `run()` method that accepts argv-style input
- delegating shared execution behavior to `CommandExecutor`
Attributes:
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.
console (Console): Rich console used for user-facing output.
executor (CommandExecutor): Shared execution engine used to run the
bound command.
"""
def __init__(
self,
command: Command,
*,
options: OptionsManager | None = None,
hooks: HookManager | None = None,
console: Console | None = None,
) -> None:
"""Initialize a `CommandRunner` for a single command.
The runner ensures that the bound command, its argument parser, and the
internal `CommandExecutor` all share the same `OptionsManager` and runtime
dependencies.
Args:
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
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.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,
console=self.console,
)
self.options.from_mapping(values={}, namespace_name="execution")
async def run(
self,
argv: list[str] | None = None,
raise_on_error: bool = True,
wrap_errors: bool = False,
summary_last_result: bool = False,
) -> Any:
"""Resolve arguments and execute the bound command.
This method is the primary execution entrypoint for `CommandRunner`. It
accepts raw argv-style tokens, resolves them into positional arguments,
keyword arguments, and execution arguments via `Command.resolve_args()`,
then delegates execution to the internal `CommandExecutor`.
Args:
argv (list[str] | None): Optional argv-style argument tokens. If
omitted, `sys.argv[1:]` is used.
Returns:
Any: The result returned by the bound command.
Raises:
Exception: Propagates any execution error surfaced by the underlying
`CommandExecutor` or command execution path.
"""
argv = sys.argv[1:] if argv is None else argv
args, kwargs, execution_args = await self.command.resolve_args(argv)
logger.debug(
"Executing command '%s' with args=%s, kwargs=%s, execution_args=%s",
self.command.description,
args,
kwargs,
execution_args,
)
return await self.executor.execute(
command=self.command,
args=args,
kwargs=kwargs,
execution_args=execution_args,
raise_on_error=raise_on_error,
wrap_errors=wrap_errors,
summary_last_result=summary_last_result,
)
async def cli(
self,
argv: list[str] | None = None,
summary_last_result: bool = False,
) -> Any:
"""Run the bound command as a shell-oriented CLI entrypoint.
This method wraps `run()` with command-line specific behavior. It executes the
bound command using raw argv-style input, then translates framework signals and
execution failures into user-facing console output and process exit codes.
Unlike `run()`, this method is intended for direct CLI usage rather than
programmatic integration. It may terminate the current process via `sys.exit()`.
Behavior:
- Delegates normal execution to `run()`
- Exits with status code `0` when help output is requested
- Exits with status code `2` for command argument or usage errors
- Exits with status code `1` for execution failures and non-success control
flow such as cancellation or back-navigation
- 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()`.
summary_last_result (bool): Whether summary output should include the last
recorded result when summary reporting is enabled.
Returns:
Any: The result returned by the bound command when execution completes
successfully.
Raises:
SystemExit: Always raised for handled CLI exit paths, including help,
argument errors, cancellations, and execution failures.
Notes:
- This method is intentionally shell-facing and should be used in
script entrypoints such as `asyncio.run(runner.cli())`.
- For programmatic use, prefer `run()`, which preserves normal Python
exception behavior and does not call `sys.exit()`.
"""
try:
return await self.run(
argv=argv,
raise_on_error=False,
wrap_errors=True,
summary_last_result=summary_last_result,
)
except HelpSignal:
sys.exit(0)
except CommandArgumentError as error:
self.command.render_help()
self.console.print(f"[{OneColors.DARK_RED}]❌ ['{self.command.key}'] {error}")
sys.exit(2)
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)
@classmethod
def from_command(
cls,
command: Command,
*,
runner_hooks: HookManager | None = None,
options: OptionsManager | None = None,
console: Console | None = None,
) -> CommandRunner:
"""Create a `CommandRunner` from an existing `Command` instance.
This factory is useful when a command has already been defined elsewhere
and should be exposed through the standalone runner interface without
rebuilding it.
Args:
command (Command): Existing command instance to wrap.
runner_hooks (HookManager | None): Optional executor-level hook manager
for the runner.
options (OptionsManager | None): Optional shared options manager.
console (Console | None): Optional Rich console for output.
Returns:
CommandRunner: A runner bound to the provided command.
Raises:
NotAFalyxError: If `runner_hooks` is provided but is not a
`HookManager` instance.
"""
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,
console=console,
)
@classmethod
def build(
cls,
key: str,
description: str,
action: BaseAction | Callable[..., Any],
*,
runner_hooks: HookManager | None = None,
args: tuple = (),
kwargs: dict[str, Any] | None = None,
hidden: bool = False,
aliases: list[str] | None = None,
help_text: str = "",
help_epilog: str = "",
style: str = OneColors.WHITE,
confirm: bool = False,
confirm_message: str = "Are you sure?",
preview_before_confirm: bool = True,
spinner: bool = False,
spinner_message: str = "Processing...",
spinner_type: str = "dots",
spinner_style: str = OneColors.CYAN,
spinner_speed: float = 1.0,
options: OptionsManager | None = None,
command_hooks: HookManager | None = None,
before_hooks: list[Callable] | None = None,
success_hooks: list[Callable] | None = None,
error_hooks: list[Callable] | None = None,
after_hooks: list[Callable] | None = None,
teardown_hooks: list[Callable] | None = None,
tags: list[str] | None = None,
logging_hooks: bool = False,
retry: bool = False,
retry_all: bool = False,
retry_policy: RetryPolicy | None = None,
arg_parser: CommandArgumentParser | None = None,
arguments: list[dict[str, Any]] | None = None,
argument_config: Callable[[CommandArgumentParser], None] | None = None,
execution_options: list[ExecutionOption | str] | None = None,
custom_parser: ArgParserProtocol | None = None,
custom_help: Callable[[], str | None] | None = None,
auto_args: bool = True,
arg_metadata: dict[str, str | dict[str, Any]] | None = None,
simple_help_signature: bool = False,
ignore_in_history: bool = False,
console: Console | None = None,
) -> CommandRunner:
"""Build a `Command` and wrap it in a `CommandRunner`.
This factory is a convenience constructor for standalone usage. It mirrors
the high-level command-building API by creating a configured `Command`
through `Command.build()` and then returning a `CommandRunner` bound to it.
Args:
key (str): Primary key used to invoke the command.
description (str): Short description of the command.
action (BaseAction | Callable[..., Any]): Underlying execution logic for
the command.
runner_hooks (HookManager | None): Optional executor-level hooks for the
runner.
args (tuple): Static positional arguments applied to the command.
kwargs (dict[str, Any] | None): Static keyword arguments applied to the
command.
hidden (bool): Whether the command should be hidden from menu displays.
aliases (list[str] | None): Optional alternate invocation names.
help_text (str): Help text shown in command help output.
help_epilog (str): Additional help text shown after the main help body.
style (str): Rich style used for rendering the command.
confirm (bool): Whether confirmation is required before execution.
confirm_message (str): Confirmation prompt text.
preview_before_confirm (bool): Whether to preview before confirmation.
spinner (bool): Whether to enable spinner integration.
spinner_message (str): Spinner message text.
spinner_type (str): Spinner animation type.
spinner_style (str): Spinner style.
spinner_speed (float): Spinner speed multiplier.
options (OptionsManager | None): Shared options manager for the command
and runner.
command_hooks (HookManager | None): Optional hook manager for the built
command itself.
before_hooks (list[Callable] | None): Command hooks registered for the
`BEFORE` lifecycle stage.
success_hooks (list[Callable] | None): Command hooks registered for the
`ON_SUCCESS` lifecycle stage.
error_hooks (list[Callable] | None): Command hooks registered for the
`ON_ERROR` lifecycle stage.
after_hooks (list[Callable] | None): Command hooks registered for the
`AFTER` lifecycle stage.
teardown_hooks (list[Callable] | None): Command hooks registered for the
`ON_TEARDOWN` lifecycle stage.
tags (list[str] | None): Optional tags used for grouping and filtering.
logging_hooks (bool): Whether to enable debug hook logging.
retry (bool): Whether retry behavior is enabled.
retry_all (bool): Whether retry behavior should be applied recursively.
retry_policy (RetryPolicy | None): Retry configuration for the command.
arg_parser (CommandArgumentParser | None): Optional explicit argument
parser instance.
arguments (list[dict[str, Any]] | None): Declarative argument
definitions.
argument_config (Callable[[CommandArgumentParser], None] | None):
Callback used to configure the argument parser.
execution_options (list[ExecutionOption | str] | None): Execution-level
options to enable for the command.
custom_parser (ArgParserProtocol | None): Optional custom parser
implementation.
custom_help (Callable[[], str | None] | None): Optional custom help
renderer.
auto_args (bool): Whether to infer arguments automatically from the
action signature.
arg_metadata (dict[str, str | dict[str, Any]] | None): Optional
metadata used during argument inference.
simple_help_signature (bool): Whether to use a simplified help
signature.
ignore_in_history (bool): Whether to exclude the command from execution
history tracking.
console (Console | None): Optional Rich console for output.
Returns:
CommandRunner: A runner wrapping the newly built command.
Raises:
NotAFalyxError: If `runner_hooks` is provided but is not a
`HookManager` instance.
Notes:
- This method is intended as a standalone convenience factory.
- Command construction is delegated to `Command.build()` so command
configuration remains centralized.
"""
options = options or OptionsManager()
command = Command.build(
key=key,
description=description,
action=action,
args=args,
kwargs=kwargs,
hidden=hidden,
aliases=aliases,
help_text=help_text,
help_epilog=help_epilog,
style=style,
confirm=confirm,
confirm_message=confirm_message,
preview_before_confirm=preview_before_confirm,
spinner=spinner,
spinner_message=spinner_message,
spinner_type=spinner_type,
spinner_style=spinner_style,
spinner_speed=spinner_speed,
tags=tags,
logging_hooks=logging_hooks,
retry=retry,
retry_all=retry_all,
retry_policy=retry_policy,
options_manager=options,
hooks=command_hooks,
before_hooks=before_hooks,
success_hooks=success_hooks,
error_hooks=error_hooks,
after_hooks=after_hooks,
teardown_hooks=teardown_hooks,
arg_parser=arg_parser,
execution_options=execution_options,
arguments=arguments,
argument_config=argument_config,
custom_parser=custom_parser,
custom_help=custom_help,
auto_args=auto_args,
arg_metadata=arg_metadata,
simple_help_signature=simple_help_signature,
ignore_in_history=ignore_in_history,
)
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,
console=console,
)

View File

@@ -1,6 +1,5 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""
Provides `FalyxCompleter`, an intelligent autocompletion engine for Falyx CLI
"""Provides `FalyxCompleter`, an intelligent autocompletion engine for Falyx CLI
menus using Prompt Toolkit.
This completer supports:
@@ -33,8 +32,7 @@ if TYPE_CHECKING:
class FalyxCompleter(Completer):
"""
Prompt Toolkit completer for Falyx CLI command input.
"""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)
@@ -57,9 +55,58 @@ class FalyxCompleter(Completer):
def __init__(self, falyx: "Falyx"):
self.falyx = falyx
@property
def _command_names(self) -> list[str]:
names: list[str] = []
seen: set[str] = set()
def add(name: str):
normalized = name.upper()
if normalized not in seen:
seen.add(normalized)
names.append(name)
for command in self.falyx.commands.values():
add(command.key)
for alias in command.aliases:
add(alias)
for command in self.falyx.builtins.values():
add(command.key)
for alias in command.aliases:
add(alias)
if self.falyx.history_command:
add(self.falyx.history_command.key)
for alias in self.falyx.history_command.aliases:
add(alias)
add(self.falyx.exit_command.key)
for alias in self.falyx.exit_command.aliases:
add(alias)
return names
def _resolve_command_for_completion(self, token: str):
normalized = token.upper().strip()
name_map = self.falyx._name_map
if normalized in name_map:
return name_map[normalized]
matches = []
seen = set()
for key, command in name_map.items():
if key.startswith(normalized) and id(command) not in seen:
matches.append(command)
seen.add(id(command))
if len(matches) == 1:
return matches[0]
return None
def get_completions(self, document: Document, complete_event) -> Iterable[Completion]:
"""
Compute completions for the current user input.
"""Compute completions for the current user input.
Analyzes the input buffer, determines whether the user is typing:
• A command key/alias
@@ -82,6 +129,13 @@ class FalyxCompleter(Completer):
except ValueError:
return
if tokens and not cursor_at_end_of_token and tokens[0].startswith("?"):
stub = tokens[0][1:]
suggestions = [c.text for c in self._suggest_commands(stub)]
prefixed = [f"?{s}" for s in suggestions]
yield from self._yield_lcp_completions(prefixed, tokens[0])
return
if not tokens or (len(tokens) == 1 and not cursor_at_end_of_token):
# Suggest command keys and aliases
stub = tokens[0] if tokens else ""
@@ -91,7 +145,7 @@ class FalyxCompleter(Completer):
# Identify command
command_key = tokens[0].upper()
command = self.falyx._name_map.get(command_key)
command = self._resolve_command_for_completion(command_key)
if not command or not command.arg_parser:
return
@@ -108,8 +162,7 @@ class FalyxCompleter(Completer):
return
def _suggest_commands(self, prefix: str) -> Iterable[Completion]:
"""
Suggest top-level command keys and aliases based on the given prefix.
"""Suggest top-level command keys and aliases based on the given prefix.
Filters all known commands (and `exit`, `help`, `history` built-ins)
to only those starting with the given prefix.
@@ -120,26 +173,13 @@ class FalyxCompleter(Completer):
Yields:
Completion: Matching keys or aliases from all registered commands.
"""
keys = [self.falyx.exit_command.key]
keys.extend(self.falyx.exit_command.aliases)
if self.falyx.history_command:
keys.append(self.falyx.history_command.key)
keys.extend(self.falyx.history_command.aliases)
if self.falyx.help_command:
keys.append(self.falyx.help_command.key)
keys.extend(self.falyx.help_command.aliases)
for cmd in self.falyx.commands.values():
keys.append(cmd.key)
keys.extend(cmd.aliases)
for key in keys:
if key.upper().startswith(prefix):
yield Completion(key.upper(), start_position=-len(prefix))
elif key.lower().startswith(prefix):
yield Completion(key.lower(), start_position=-len(prefix))
for name in self._command_names:
if name.upper().startswith(prefix.upper()):
text = name.lower() if prefix.islower() else name
yield Completion(text, start_position=-len(prefix), display=text)
def _ensure_quote(self, text: str) -> str:
"""
Ensure that a suggestion is shell-safe by quoting if needed.
"""Ensure that a suggestion is shell-safe by quoting if needed.
Adds quotes around completions containing whitespace so they can
be inserted into the CLI without breaking tokenization.
@@ -155,8 +195,7 @@ class FalyxCompleter(Completer):
return text
def _yield_lcp_completions(self, suggestions, stub):
"""
Yield completions for the current stub using longest-common-prefix logic.
"""Yield completions for the current stub using longest-common-prefix logic.
Behavior:
- If only one match → yield it fully.

21
falyx/execution_option.py Normal file
View File

@@ -0,0 +1,21 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
from __future__ import annotations
from enum import Enum
class ExecutionOption(Enum):
SUMMARY = "summary"
RETRY = "retry"
CONFIRM = "confirm"
@classmethod
def _missing_(cls, value: object) -> ExecutionOption:
if not isinstance(value, str):
raise ValueError(f"Invalid {cls.__name__}: {value!r}")
normalized = value.strip().lower()
for member in cls:
if member.value == normalized:
return member
valid = ", ".join(member.value for member in cls)
raise ValueError(f"Invalid {cls.__name__}: '{value}'. Must be one of: {valid}")

File diff suppressed because it is too large Load Diff

View File

@@ -38,34 +38,34 @@ from falyx.themes import OneColors
async def spinner_before_hook(context: ExecutionContext):
"""Adds a spinner before the action starts."""
cmd = context.action
if cmd.options_manager is None:
command = context.action
if command.options_manager is None:
return
sm = context.action.options_manager.spinners
if hasattr(cmd, "name"):
cmd_name = cmd.name
if hasattr(command, "name"):
command_name = command.name
else:
cmd_name = cmd.key
command_name = command.key
await sm.add(
cmd_name,
cmd.spinner_message,
cmd.spinner_type,
cmd.spinner_style,
cmd.spinner_speed,
command_name,
command.spinner_message,
command.spinner_type,
command.spinner_style,
command.spinner_speed,
)
async def spinner_teardown_hook(context: ExecutionContext):
"""Removes the spinner after the action finishes (success or failure)."""
cmd = context.action
if cmd.options_manager is None:
command = context.action
if command.options_manager is None:
return
if hasattr(cmd, "name"):
cmd_name = cmd.name
if hasattr(command, "name"):
command_name = command.name
else:
cmd_name = cmd.key
command_name = command.key
sm = context.action.options_manager.spinners
await sm.remove(cmd_name)
await sm.remove(command_name)
class ResultReporter:

View File

@@ -101,12 +101,16 @@ class MenuOptionMap(CaseInsensitiveDict):
self,
options: dict[str, MenuOption] | None = None,
allow_reserved: bool = False,
disable_reserved: bool = False,
):
super().__init__()
self.allow_reserved = allow_reserved
if options:
self.update(options)
self._inject_reserved_defaults()
if not disable_reserved:
self._inject_reserved_defaults()
else:
self.allow_reserved = True
def _inject_reserved_defaults(self):
from falyx.action import SignalAction

View File

@@ -1,13 +1,11 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""
Defines `FalyxMode`, an enum representing the different modes of operation for Falyx.
"""
"""Defines `FalyxMode`, an enum representing the different modes of operation for Falyx."""
from enum import Enum
class FalyxMode(Enum):
MENU = "menu"
RUN = "run"
COMMAND = "command"
PREVIEW = "preview"
RUN_ALL = "run-all"
HELP = "help"
ERROR = "error"

View File

@@ -1,12 +1,11 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""
Manages global or scoped CLI options across namespaces for Falyx commands.
"""Manages global or scoped CLI options across namespaces for Falyx commands.
The `OptionsManager` provides a centralized interface for retrieving, setting, toggling,
and introspecting options defined in `argparse.Namespace` objects. It is used internally
by Falyx to pass and resolve runtime flags like `--verbose`, `--force-confirm`, etc.
Each option is stored under a namespace key (e.g., "cli_args", "user_config") to
Each option is stored under a namespace key (e.g., "default", "user_config") to
support multiple sources of configuration.
Key Features:
@@ -17,7 +16,7 @@ Key Features:
Typical Usage:
options = OptionsManager()
options.from_namespace(args, namespace_name="cli_args")
options.from_namespace(args, namespace_name="default")
if options.get("verbose"):
...
options.toggle("force_confirm")
@@ -29,51 +28,71 @@ Used by:
- Bottom bar toggles
- Dynamic flag injection into commands and actions
"""
from argparse import Namespace
from collections import defaultdict
from typing import Any, Callable
from contextlib import contextmanager
from typing import Any, Callable, Iterator, Mapping
from falyx.logger import logger
from falyx.spinner_manager import SpinnerManager
class OptionsManager:
"""
Manages CLI option state across multiple argparse namespaces.
"""Manages CLI option state across multiple argparse namespaces.
Allows dynamic retrieval, setting, toggling, and introspection of command-line
options. Supports named namespaces (e.g., "cli_args") and is used throughout
options. Supports named namespaces (e.g., "default") and is used throughout
Falyx for runtime configuration and bottom bar toggle integration.
"""
def __init__(self, namespaces: list[tuple[str, Namespace]] | None = None) -> None:
self.options: defaultdict = defaultdict(Namespace)
def __init__(
self,
namespaces: list[tuple[str, dict[str, Any]]] | None = None,
) -> None:
self.options: defaultdict = defaultdict(dict)
self.spinners = SpinnerManager()
if namespaces:
for namespace_name, namespace in namespaces:
self.from_namespace(namespace, namespace_name)
self.from_mapping(namespace, namespace_name)
def from_namespace(
self, namespace: Namespace, namespace_name: str = "cli_args"
def from_mapping(
self,
values: Mapping[str, Any],
namespace_name: str = "default",
) -> None:
self.options[namespace_name] = namespace
"""Load options from a mapping, optionally with a prefix for namespacing."""
self.options[namespace_name].update(dict(values))
def get(
self, option_name: str, default: Any = None, namespace_name: str = "cli_args"
self,
option_name: str,
default: Any = None,
namespace_name: str = "default",
) -> Any:
"""Get the value of an option."""
return getattr(self.options[namespace_name], option_name, default)
return self.options[namespace_name].get(option_name, default)
def set(self, option_name: str, value: Any, namespace_name: str = "cli_args") -> None:
def set(
self,
option_name: str,
value: Any,
namespace_name: str = "default",
) -> None:
"""Set the value of an option."""
setattr(self.options[namespace_name], option_name, value)
self.options[namespace_name][option_name] = value
def has_option(self, option_name: str, namespace_name: str = "cli_args") -> bool:
def has_option(
self,
option_name: str,
namespace_name: str = "default",
) -> bool:
"""Check if an option exists in the namespace."""
return hasattr(self.options[namespace_name], option_name)
return option_name in self.options[namespace_name]
def toggle(self, option_name: str, namespace_name: str = "cli_args") -> None:
def toggle(
self,
option_name: str,
namespace_name: str = "default",
) -> None:
"""Toggle a boolean option."""
current = self.get(option_name, namespace_name=namespace_name)
if not isinstance(current, bool):
@@ -86,7 +105,9 @@ class OptionsManager:
)
def get_value_getter(
self, option_name: str, namespace_name: str = "cli_args"
self,
option_name: str,
namespace_name: str = "default",
) -> Callable[[], Any]:
"""Get the value of an option as a getter function."""
@@ -96,7 +117,9 @@ class OptionsManager:
return _getter
def get_toggle_function(
self, option_name: str, namespace_name: str = "cli_args"
self,
option_name: str,
namespace_name: str = "default",
) -> Callable[[], None]:
"""Get the toggle function for a boolean option."""
@@ -105,8 +128,22 @@ class OptionsManager:
return _toggle
def get_namespace_dict(self, namespace_name: str) -> Namespace:
def get_namespace_dict(self, namespace_name: str) -> dict[str, Any]:
"""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])
return dict(self.options[namespace_name])
@contextmanager
def override_namespace(
self,
overrides: Mapping[str, Any],
namespace_name: str = "execution",
) -> Iterator[None]:
"""Temporarily override options in a namespace within a context."""
original = self.get_namespace_dict(namespace_name)
try:
self.from_mapping(values=overrides, namespace_name=namespace_name)
yield
finally:
self.options[namespace_name] = original

View File

@@ -8,14 +8,13 @@ Licensed under the MIT License. See LICENSE file for details.
from .argument import Argument
from .argument_action import ArgumentAction
from .command_argument_parser import CommandArgumentParser
from .parsers import FalyxParsers, get_arg_parsers, get_root_parser, get_subparsers
from .falyx_parser import FalyxParser
from .parse_result import ParseResult
__all__ = [
"Argument",
"ArgumentAction",
"CommandArgumentParser",
"get_arg_parsers",
"get_root_parser",
"get_subparsers",
"FalyxParsers",
"FalyxParser",
"ParseResult",
]

View File

@@ -60,6 +60,8 @@ class Argument:
An action object that resolves the argument, if applicable.
lazy_resolver (bool): True if the resolver should be called lazily, False otherwise
suggestions (list[str] | None): Optional completions for interactive shells
group (str | None): Optional name of the argument group this belongs to.
mutex_group (str | None): Optional name of the mutually exclusive group this belongs to.
"""
flags: tuple[str, ...]
@@ -75,6 +77,8 @@ class Argument:
resolver: BaseAction | None = None
lazy_resolver: bool = False
suggestions: list[str] | None = None
group: str | None = None
mutex_group: str | None = None
def get_positional_text(self) -> str:
"""Get the positional text for the argument."""
@@ -132,6 +136,8 @@ class Argument:
and self.positional == other.positional
and self.default == other.default
and self.help == other.help
and self.group == other.group
and self.mutex_group == other.mutex_group
)
def __hash__(self) -> int:
@@ -147,5 +153,7 @@ class Argument:
self.positional,
self.default,
self.help,
self.group,
self.mutex_group,
)
)

View File

@@ -1,56 +1,55 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""
This module implements `CommandArgumentParser`, a flexible, rich-aware alternative to
argparse tailored specifically for Falyx CLI workflows. It provides structured parsing,
type coercion, flag support, and usage/help rendering for CLI-defined commands.
"""CommandArgumentParser implementation for the Falyx CLI framework.
Unlike argparse, this parser is lightweight, introspectable, and designed to integrate
deeply with Falyx's Action system, including support for lazy execution and resolver
binding via `BaseAction`.
This module provides a structured, extensible argument parsing system designed
specifically for Falyx commands. It replaces traditional argparse usage with a
parser that is deeply integrated with Falyx's execution model, including support
for Actions, execution options, and interactive completion.
The parser is designed to:
- Define command arguments declaratively via `add_argument`
- Support both positional and keyword-style flags
- Perform type coercion and validation
- Separate execution-level options (e.g. retries, confirmation) from command inputs
- Integrate with Falyx lifecycle and Action-based execution
- Provide rich help rendering and interactive suggestions
Key Features:
- Declarative argument registration via `add_argument()`
- Support for positional and keyword flags, type coercion, default values
- Enum- and action-driven argument semantics via `ArgumentAction`
- Lazy evaluation of arguments using Falyx `Action` resolvers
- Optional value completion via suggestions and choices
- Rich-powered help rendering with grouped display
- Optional boolean flags via `--flag` / `--no-flag`
- POSIX-style bundling for single-character flags (`-abc`)
- Partial parsing for completions and validation via `suggest_next()`
- Positional and flagged argument support
- Type coercion via configurable `type` handlers
- Enum-driven behavior via `ArgumentAction`
- Lazy and eager resolution using BaseAction resolvers
- Execution option support (e.g. retries, summary, confirm flags)
- Mutually exclusive and grouped argument definitions
- POSIX-style short flag bundling (e.g. `-abc`)
- Interactive suggestions via `suggest_next`
- Rich-based help and TLDR rendering
Public Interface:
- `add_argument(...)`: Register a new argument with type, flags, and behavior.
- `parse_args(...)`: Parse CLI-style argument list into a `dict[str, Any]`.
- `parse_args_split(...)`: Return `(*args, **kwargs)` for Action invocation.
- `render_help()`: Render a rich-styled help panel.
- `render_tldr()`: Render quick usage examples.
- `suggest_next(...)`: Return suggested flags or values for completion.
Core Parsing APIs:
- `parse_args(...)`:
Parse arguments into a resolved dictionary of values
- `parse_args_split(...)`:
Split parsed results into `(args, kwargs, execution_args)` for execution
- `add_argument(...)`:
Register argument definitions declaratively
- `suggest_next(...)`:
Provide completion suggestions for interactive input
Example Usage:
parser = CommandArgumentParser(command_key="D")
parser.add_argument("--env", choices=["prod", "dev"], required=True)
parser.add_argument("path", type=Path)
Design Principles:
- Minimal surface area compared to argparse
- Strong integration with Falyx execution model
- Predictable and explicit parsing behavior
- Separation of parsing, execution, and runtime configuration
args = await parser.parse_args(["--env", "prod", "./config.yml"])
# args == {'env': 'prod', 'path': Path('./config.yml')}
parser.render_help() # Pretty Rich output
Design Notes:
This parser intentionally omits argparse-style groups, metavar support,
and complex multi-level conflict handling. Instead, it favors:
- Simplicity
- Completeness
- Falyx-specific integration (hooks, lifecycle, and error surfaces)
This parser is intended for use exclusively within Falyx and is not a
general-purpose argparse replacement.
"""
from __future__ import annotations
from collections import Counter, defaultdict
from copy import deepcopy
from pathlib import Path
from typing import Any, Iterable, Sequence
from typing import Any, Generator, Iterable, Sequence
from rich.console import Console
from rich.markup import escape
@@ -60,15 +59,49 @@ from rich.panel import Panel
from falyx.action.base_action import BaseAction
from falyx.console import console
from falyx.exceptions import CommandArgumentError
from falyx.execution_option import ExecutionOption
from falyx.mode import FalyxMode
from falyx.options_manager import OptionsManager
from falyx.parser.argument import Argument
from falyx.parser.argument_action import ArgumentAction
from falyx.parser.group import ArgumentGroup, MutuallyExclusiveGroup
from falyx.parser.parser_types import ArgumentState, TLDRExample, false_none, true_none
from falyx.parser.utils import coerce_value
from falyx.signals import HelpSignal
class _GroupBuilder:
"""Helper for assigning arguments to a named group or mutex group.
This lightweight wrapper preserves the normal `add_argument()` API while
injecting `group` or `mutex_group` metadata into each registered argument.
Args:
parser (CommandArgumentParser): Parser that owns the group definitions.
group_name (str | None): Name of the argument group to assign.
mutex_name (str | None): Name of the mutually exclusive group to assign.
"""
def __init__(
self,
parser: CommandArgumentParser,
*,
group_name: str | None = None,
mutex_name: str | None = None,
) -> None:
self.parser = parser
self.group_name = group_name
self.mutex_name = mutex_name
def add_argument(self, *flags, **kwargs) -> None:
self.parser.add_argument(
*flags,
group=self.group_name,
mutex_group=self.mutex_name,
**kwargs,
)
class CommandArgumentParser:
"""
Custom argument parser for Falyx Commands.
@@ -90,7 +123,7 @@ class CommandArgumentParser:
- Render Help using Rich library.
"""
RESERVED_DESTS = frozenset(("help", "tldr"))
RESERVED_DESTS = frozenset({"help", "tldr"})
def __init__(
self,
@@ -120,15 +153,89 @@ class CommandArgumentParser:
self._keyword_list: list[Argument] = []
self._flag_map: dict[str, Argument] = {}
self._dest_set: set[str] = set()
self._execution_dests: set[str] = set()
self._add_help()
self._last_positional_states: dict[str, ArgumentState] = {}
self._last_keyword_states: dict[str, ArgumentState] = {}
self._argument_groups: dict[str, ArgumentGroup] = {}
self._mutex_groups: dict[str, MutuallyExclusiveGroup] = {}
self._arg_group_by_dest: dict[str, str] = {}
self._mutex_group_by_dest: dict[str, str] = {}
self._tldr_examples: list[TLDRExample] = []
self._is_help_command: bool = _is_help_command
if tldr_examples:
self.add_tldr_examples(tldr_examples)
self.options_manager: OptionsManager = options_manager or OptionsManager()
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")
self.options_manager = options_manager
def enable_execution_options(
self,
execution_options: frozenset[ExecutionOption],
) -> None:
"""Enable support for execution options like retries, summary, etc."""
if ExecutionOption.SUMMARY in execution_options:
self.add_argument(
"--summary",
action=ArgumentAction.STORE_TRUE,
help="Print an execution summary after command completes",
)
self._register_execution_dest("summary")
if ExecutionOption.RETRY in execution_options:
self.add_argument(
"--retries",
type=int,
help="Number of retries on failure",
default=0,
)
self._register_execution_dest("retries")
self.add_argument(
"--retry-delay",
type=float,
default=0.0,
help="Initial delay between retries in seconds",
)
self._register_execution_dest("retry_delay")
self.add_argument(
"--retry-backoff",
type=float,
default=0.0,
help="Backoff multiplier for retries (e.g. 2.0 doubles the delay each retry)",
)
self._register_execution_dest("retry_backoff")
if ExecutionOption.CONFIRM in execution_options:
self.add_argument(
"--confirm",
dest="force_confirm",
action=ArgumentAction.STORE_TRUE,
help="Force confirmation prompts",
)
self._register_execution_dest("force_confirm")
self.add_argument(
"--skip-confirm",
action=ArgumentAction.STORE_TRUE,
help="Skip confirmation prompts",
)
self._register_execution_dest("skip_confirm")
def _register_execution_dest(self, dest: str) -> None:
"""Register a destination as an execution argument."""
if dest in self._execution_dests:
raise CommandArgumentError(
f"Destination '{dest}' is already registered as an execution argument"
)
self._execution_dests.add(dest)
def _is_execution_dest(self, dest: str) -> bool:
"""Check if a destination is registered as an execution argument."""
return dest in self._execution_dests
def _add_help(self):
"""Add help argument to the parser."""
help = Argument(
@@ -165,6 +272,32 @@ class CommandArgumentParser:
)
self._register_argument(tldr)
def add_argument_group(
self,
name: str,
description: str = "",
) -> _GroupBuilder:
if name in self._argument_groups:
raise CommandArgumentError(f"Argument group '{name}' already exists")
self._argument_groups[name] = ArgumentGroup(name=name, description=description)
return _GroupBuilder(self, group_name=name)
def add_mutually_exclusive_group(
self,
name: str,
*,
required: bool = False,
description: str = "",
) -> _GroupBuilder:
if name in self._mutex_groups:
raise CommandArgumentError(f"Mutex group '{name}' already exists")
self._mutex_groups[name] = MutuallyExclusiveGroup(
name=name,
required=required,
description=description,
)
return _GroupBuilder(self, mutex_name=name)
def _is_positional(self, flags: tuple[str, ...]) -> bool:
"""Check if the flags are positional."""
positional = False
@@ -175,6 +308,34 @@ class CommandArgumentParser:
raise CommandArgumentError("Positional arguments cannot have multiple flags")
return positional
def _validate_groups(
self,
group: str | None,
mutex_group: str | None,
positional: bool = False,
required: bool = False,
) -> None:
"""Validate that the specified groups exist and are compatible."""
if group is not None:
if group not in self._argument_groups:
raise CommandArgumentError(f"Argument group '{group}' does not exist")
if mutex_group is not None:
if mutex_group not in self._mutex_groups:
raise CommandArgumentError(
f"Mutually exclusive group '{mutex_group}' does not exist"
)
if positional and mutex_group is not None:
raise CommandArgumentError(
"Positional arguments cannot belong to a mutually exclusive group"
)
if required and mutex_group is not None:
raise CommandArgumentError(
"Arguments inside a mutually exclusive group should not be individually required; "
"make the group required instead."
)
def _get_dest_from_flags(self, flags: tuple[str, ...], dest: str | None) -> str:
"""Convert flags to a destination name."""
if dest:
@@ -444,6 +605,8 @@ class CommandArgumentParser:
flags: tuple[str, ...],
dest: str,
help: str,
group: str | None,
mutex_group: str | None,
) -> None:
"""Register a store_bool_optional action with the parser."""
if len(flags) != 1:
@@ -464,6 +627,8 @@ class CommandArgumentParser:
type=true_none,
default=None,
help=help,
group=group,
mutex_group=mutex_group,
)
negated_argument = Argument(
@@ -473,6 +638,8 @@ class CommandArgumentParser:
type=false_none,
default=None,
help=help,
group=group,
mutex_group=mutex_group,
)
self._register_argument(argument)
@@ -503,6 +670,14 @@ class CommandArgumentParser:
else:
self._keyword_list.append(argument)
if argument.group:
self._arg_group_by_dest[argument.dest] = argument.group
self._argument_groups[argument.group].dests.append(argument.dest)
if argument.mutex_group:
self._mutex_group_by_dest[argument.dest] = argument.mutex_group
self._mutex_groups[argument.mutex_group].dests.append(argument.dest)
def add_argument(
self,
*flags,
@@ -517,6 +692,8 @@ class CommandArgumentParser:
resolver: BaseAction | None = None,
lazy_resolver: bool = True,
suggestions: list[str] | None = None,
group: str | None = None,
mutex_group: str | None = None,
) -> None:
"""
Define a new argument for the parser.
@@ -537,6 +714,8 @@ class CommandArgumentParser:
resolver (BaseAction | None): If action="action", the BaseAction to call.
lazy_resolver (bool): If True, resolver defers until action is triggered.
suggestions (list[str] | None): Optional suggestions for interactive completion.
group (str | None): Optional argument group name for help organization.
mutex_group (str | None): Optional mutually exclusive group name.
"""
expected_type = type
self._validate_flags(flags)
@@ -552,6 +731,9 @@ class CommandArgumentParser:
raise CommandArgumentError(
f"Destination '{dest}' is reserved and cannot be used."
)
self._validate_groups(group, mutex_group, positional, required)
action = self._validate_action(action, positional)
resolver = self._validate_resolver(action, resolver)
@@ -587,7 +769,7 @@ class CommandArgumentParser:
f"lazy_resolver must be a boolean, got {type(lazy_resolver)}"
)
if action == ArgumentAction.STORE_BOOL_OPTIONAL:
self._register_store_bool_optional(flags, dest, help)
self._register_store_bool_optional(flags, dest, help, group, mutex_group)
else:
argument = Argument(
flags=flags,
@@ -603,6 +785,8 @@ class CommandArgumentParser:
resolver=resolver,
lazy_resolver=lazy_resolver,
suggestions=suggestions,
group=group,
mutex_group=mutex_group,
)
self._register_argument(argument)
@@ -641,6 +825,8 @@ class CommandArgumentParser:
"positional": arg.positional,
"default": arg.default,
"help": arg.help,
"group": arg.group,
"mutex_group": arg.mutex_group,
}
)
return defs
@@ -700,6 +886,10 @@ class CommandArgumentParser:
), f"Invalid nargs value: {spec.nargs}"
values = []
if isinstance(spec.nargs, int):
if index + spec.nargs > len(args):
raise CommandArgumentError(
f"Expected {spec.nargs} value(s) for '{spec.dest}' but got {len(args) - index}"
)
values = args[index : index + spec.nargs]
return values, index + spec.nargs
elif spec.nargs == "+":
@@ -744,7 +934,6 @@ class CommandArgumentParser:
if spec_index not in consumed_positional_indicies
]
index = 0
for spec_index, spec in remaining_positional_args:
# estimate how many args the remaining specs might need
is_last = spec_index == len(positional_args) - 1
@@ -779,7 +968,6 @@ class CommandArgumentParser:
)
values, new_index = self._consume_nargs(slice_args, 0, spec)
index += new_index
try:
typed = [coerce_value(value, spec.type) for value in values]
except Exception as error:
@@ -798,6 +986,14 @@ class CommandArgumentParser:
assert isinstance(
spec.resolver, BaseAction
), "resolver should be an instance of BaseAction"
if spec.nargs == "+" and len(typed) == 0:
raise CommandArgumentError(
f"Argument '{spec.dest}' requires at least one value"
)
if isinstance(spec.nargs, int) and len(typed) != spec.nargs:
raise CommandArgumentError(
f"Argument '{spec.dest}' requires exactly {spec.nargs} value(s)"
)
if not spec.lazy_resolver or not from_validate:
try:
result[spec.dest] = await spec.resolver(*typed)
@@ -831,7 +1027,6 @@ class CommandArgumentParser:
if spec.nargs not in ("*", "+"):
consumed_positional_indicies.add(spec_index)
if index < len(args):
if len(args[index:]) == 1 and args[index].startswith("-"):
token = args[index]
@@ -1103,18 +1298,90 @@ class CommandArgumentParser:
args[expand_index : expand_index + 1] = expand_token
expand_index += len(expand_token) if isinstance(expand_token, list) else 1
def _is_present(self, spec: Argument, value: Any) -> bool:
"""
Presence means 'user actually selected/provided this', not merely that
a default exists.
"""
if spec.action == ArgumentAction.STORE_TRUE:
return value is True
if spec.action == ArgumentAction.STORE_FALSE:
return value is False
if spec.action == ArgumentAction.STORE_BOOL_OPTIONAL:
return value is not None
if spec.action == ArgumentAction.COUNT:
return bool(value)
if spec.action in (ArgumentAction.APPEND, ArgumentAction.EXTEND):
return bool(value)
return value is not None
def _validate_mutex_groups(self, result: dict[str, Any]) -> None:
for group in self._mutex_groups.values():
present: list[str] = []
for dest in group.dests:
spec = self.get_argument(dest)
if spec is None:
continue
if self._is_present(spec, result.get(dest)):
present.append(dest)
if len(present) > 1:
raise CommandArgumentError(
f"Arguments in mutually exclusive group '{group.name}' "
f"cannot be used together: {', '.join(present)}"
)
if group.required and not present:
members = []
for dest in group.dests:
spec = self.get_argument(dest)
if spec:
members.append(spec.flags[0] if spec.flags else dest)
raise CommandArgumentError(
f"One of the following is required for group '{group.name}': "
f"{', '.join(members)}"
)
async def parse_args(
self, args: list[str] | None = None, from_validate: bool = False
) -> dict[str, Any]:
"""
Parse arguments into a dictionary of resolved values.
"""Parse CLI arguments into a resolved mapping of values.
This method parses the provided CLI-style tokens and returns a dictionary
mapping argument destinations to their resolved values. It performs full
validation, type coercion, default handling, and resolver execution.
Unlike `parse_args_split`, this method returns a unified mapping of all
parsed arguments, including both command arguments and execution options.
Behavior:
- Parses positional and keyword arguments based on registered definitions
- Applies type coercion via configured `type` handlers
- Resolves values using BaseAction resolvers (if defined)
- Validates required arguments, choices, and mutual exclusion constraints
- Applies default values for missing optional arguments
- Supports validation mode (`from_validate=True`) for interactive contexts
Args:
args (list[str]): The CLI-style argument list.
from_validate (bool): If True, enables relaxed resolution for validation mode.
args (list[str]): CLI-style argument tokens to parse.
from_validate (bool): Whether parsing is occurring in validation mode
(e.g. prompt_toolkit validator). When True, may defer certain
resolution steps or suppress eager failures.
Returns:
dict[str, Any]: Parsed argument result mapping.
dict[str, Any]: Mapping of argument destination names to resolved values.
Raises:
CommandArgumentError: If parsing, validation, or coercion fails.
HelpSignal: If help or TLDR output is triggered during parsing.
Notes:
- This method returns a flat mapping of all arguments.
- Use `parse_args_split` when separating execution options from
command arguments is required for execution.
- This is the primary parsing entrypoint used internally by
`parse_args_split`.
"""
if args is None:
args = []
@@ -1151,6 +1418,27 @@ class CommandArgumentParser:
from_validate=from_validate,
)
# Compare length of args with length of required positional arguments to catch missing required positionals
if len(args) < len(
[
arg
for arg in self._arguments
if (arg.positional and arg.required and not arg.default)
]
):
missing_positionals = [
arg.dest
for arg in self._arguments
if arg.positional
and arg.required
and arg.dest not in consumed_positional_indices
and not arg.default
]
if missing_positionals:
raise CommandArgumentError(
f"Missing positional argument(s): {', '.join(missing_positionals)}"
)
# Required validation
for spec in self._arguments:
if spec.dest == "help" or spec.dest == "tldr":
@@ -1203,6 +1491,21 @@ class CommandArgumentParser:
f"Invalid number of values for '{spec.dest}': expected {spec.nargs}, got {len(result[spec.dest])}"
)
if isinstance(spec.nargs, str) and spec.nargs == "+":
assert isinstance(
result.get(spec.dest), list
), f"Invalid value for '{spec.dest}': expected a list"
if not result[spec.dest] and not spec.required:
continue
help_text = f" help: {spec.help}" if spec.help else ""
if not result[spec.dest]:
arg_states[spec.dest].reset()
raise CommandArgumentError(
f"Argument '{spec.dest}' requires at least one value{help_text}"
)
self._validate_mutex_groups(result)
result.pop("help", None)
if not self._is_help_command:
result.pop("tldr", None)
@@ -1210,18 +1513,33 @@ class CommandArgumentParser:
async def parse_args_split(
self, args: list[str], from_validate: bool = False
) -> tuple[tuple[Any, ...], dict[str, Any]]:
"""
Parse arguments and return both positional and keyword mappings.
) -> tuple[tuple[Any, ...], dict[str, Any], dict[str, Any]]:
"""Parse arguments and split them into execution-ready components.
Useful for function-style calling with `*args, **kwargs`.
This method parses the provided CLI-style tokens and separates the resolved
values into three categories:
- positional arguments for `*args`
- keyword arguments for `**kwargs`
- execution arguments for Falyx runtime behavior
Execution arguments are options such as retries, confirmation flags, or
summary output that should not be passed to the underlying action.
Args:
args (list[str]): CLI-style argument tokens to parse.
from_validate (bool): Whether parsing is occurring in validation mode.
Returns:
tuple: (args tuple, kwargs dict)
tuple:
- tuple[Any, ...]: Positional arguments for execution.
- dict[str, Any]: Keyword arguments for execution.
- dict[str, Any]: Execution-specific arguments handled by Falyx.
"""
parsed = await self.parse_args(args, from_validate)
args_list = []
kwargs_dict = {}
execution_dict = {}
for arg in self._arguments:
if arg.dest == "help":
continue
@@ -1229,9 +1547,11 @@ class CommandArgumentParser:
continue
if arg.positional:
args_list.append(parsed[arg.dest])
elif self._is_execution_dest(arg.dest):
execution_dict[arg.dest] = parsed[arg.dest]
else:
kwargs_dict[arg.dest] = parsed[arg.dest]
return tuple(args_list), kwargs_dict
return tuple(args_list), kwargs_dict, execution_dict
def _suggest_paths(self, stub: str) -> list[str]:
"""Return filesystem path suggestions based on a stub."""
@@ -1320,20 +1640,57 @@ class CommandArgumentParser:
return self._suggest_paths(prefix if not cursor_at_end_of_token else ".")
return []
def _filter_mutex_flags(
self,
remaining_flags: list[str],
consumed_dests: list[str],
) -> list[str]:
active_mutex_groups = {
self._mutex_group_by_dest[dest]
for dest in consumed_dests
if dest in self._mutex_group_by_dest
}
if not active_mutex_groups:
return remaining_flags
filtered: list[str] = []
for flag in remaining_flags:
arg = self._keyword[flag]
mutex_name = self._mutex_group_by_dest.get(arg.dest)
if (
mutex_name
and mutex_name in active_mutex_groups
and arg.dest not in consumed_dests
):
continue
filtered.append(flag)
return filtered
def suggest_next(
self, args: list[str], cursor_at_end_of_token: bool = False
) -> list[str]:
"""
Suggest completions for the next argument based on current input.
"""Suggest valid completions for the current argument state.
This is used for interactive shell completion or prompt_toolkit integration.
This method analyzes the partially entered argument list and returns
context-aware suggestions for the next token. Suggestions may include:
- remaining flags
- valid choices for the current argument
- configured custom suggestions
- filesystem paths for `Path`-typed arguments
It supports positional arguments, flagged arguments, multi-value arguments,
POSIX short-flag bundling, and mutually exclusive group filtering.
Args:
args (list[str]): Current partial argument tokens.
cursor_at_end_of_token (bool): True if space at end of args
cursor_at_end_of_token (bool): Whether the cursor is positioned after a
completed token (for example, after a trailing space).
Returns:
list[str]: List of suggested completions.
list[str]: Sorted completion suggestions valid for the current parse state.
"""
self._resolve_posix_bundling(args)
last = args[-1] if args else ""
@@ -1406,6 +1763,7 @@ class CommandArgumentParser:
remaining_flags = [
flag for flag, arg in self._keyword.items() if arg.dest not in consumed_dests
]
remaining_flags = self._filter_mutex_flags(remaining_flags, consumed_dests)
last_keyword_state_in_args = None
last_keyword = None
@@ -1665,23 +2023,65 @@ class CommandArgumentParser:
command_keys = self.get_command_keys_text(plain_text)
options_text = self.get_options_text(plain_text)
if options_text:
return f"{command_keys} {options_text}"
if self.options_manager.get("mode") == FalyxMode.MENU:
return f"{command_keys} {options_text}"
else:
program = self.program or "falyx"
program_style = (
self.options_manager.get("program_style") or self.command_style
)
return f"[{program_style}]{program}[/{program_style}] {command_keys} {options_text}"
return command_keys
def render_help(self) -> None:
def _iter_keyword_help_sections(
self,
) -> Generator[tuple[str, str, list[Argument]], None, None]:
"""
Print formatted help text for this command using Rich output.
Yields (title, description, arguments)
"""
assigned = set()
Includes usage, description, argument groups, and optional epilog.
for group in self._argument_groups.values():
args = []
for dest in group.dests:
spec = self.get_argument(dest)
if spec and not spec.positional:
args.append(spec)
assigned.add(dest)
if args:
yield group.name, group.description, args
ungrouped = []
for arg in self._keyword_list:
if arg.dest not in assigned:
ungrouped.append(arg)
if ungrouped:
yield "options", "", ungrouped
def render_help(self) -> None:
"""Render full help output for the command.
This method displays a complete help view for the command, including
usage, description, argument definitions, execution options, and any
additional help text.
The output is formatted using Rich and is intended for both CLI and
interactive menu contexts.
Behavior:
- Renders a usage string derived from the parser configuration
- Displays command description, aliases, and optional epilog text
- Lists positional and keyword arguments with types, defaults, and help text
- Supports argument grouping and mutually exclusive groups
- Applies styling based on configured command style
"""
usage = self.get_usage()
self.console.print(f"[bold]usage: {usage}[/bold]\n")
# Description
if self.help_text:
self.console.print(self.help_text + "\n")
# Arguments
if self._arguments:
if self._positional:
self.console.print("[bold]positional:[/bold]")
@@ -1692,62 +2092,70 @@ class CommandArgumentParser:
if help_text and len(flags) > 30:
help_text = f"\n{'':<33}{help_text}"
self.console.print(f"{arg_line}{help_text}")
self.console.print("[bold]options:[/bold]")
arg_groups = defaultdict(list)
for arg in self._keyword_list:
arg_groups[arg.dest].append(arg)
for group in arg_groups.values():
if len(group) == 2 and all(
arg.action == ArgumentAction.STORE_BOOL_OPTIONAL for arg in group
):
# Merge --flag / --no-flag pair into single help line for STORE_BOOL_OPTIONAL
all_flags = tuple(
sorted(
(arg.flags[0] for arg in group),
key=lambda f: f.startswith("--no-"),
for title, description, args in self._iter_keyword_help_sections():
self.console.print(f"\n[bold]{title}:[/bold]")
if description:
self.console.print(f" [dim]{description}[/dim]")
arg_groups: defaultdict[str, list[Argument]] = defaultdict(list)
for arg in args:
arg_groups[arg.dest].append(arg)
for group in arg_groups.values():
if len(group) == 2 and all(
arg.action == ArgumentAction.STORE_BOOL_OPTIONAL for arg in group
):
# Merge --flag / --no-flag pair into single help line for STORE_BOOL_OPTIONAL
all_flags = tuple(
sorted(
(arg.flags[0] for arg in group),
key=lambda f: f.startswith("--no-"),
)
)
)
else:
all_flags = group[0].flags
else:
all_flags = group[0].flags
flags = ", ".join(all_flags)
flags_choice = f"{flags} {group[0].get_choice_text()}"
arg_line = f" {flags_choice:<30} "
help_text = group[0].help or ""
if help_text and len(flags_choice) > 30:
help_text = f"\n{'':<33}{help_text}"
self.console.print(f"{arg_line}{help_text}")
suffix = ""
mutex_name = group[0].mutex_group
if mutex_name:
suffix = f" [dim]({mutex_name})[/dim]"
flags = ", ".join(all_flags)
flags_choice = f"{flags} {group[0].get_choice_text()}"
arg_line = f" {flags_choice:<30} "
help_text = f"{group[0].help or ''}{suffix}"
if help_text and len(flags_choice) > 30:
help_text = f"\n{'':<33}{help_text}"
self.console.print(f"{arg_line}{help_text}")
# Epilog
if self.help_epilog:
self.console.print("\n" + self.help_epilog, style="dim")
def render_tldr(self) -> None:
"""
Print TLDR examples for this command using Rich output.
"""Render concise example usage (TLDR) for the command.
Displays brief usage examples with descriptions.
This method displays a minimal, example-driven view of how to invoke
the command. It is intended as a quick-start reference rather than a
complete specification.
Notes:
- TLDR output is designed for speed and clarity, not completeness.
- Typically invoked via `--tldr` or equivalent help flags.
- Complements `render_help`, which provides full documentation.
"""
if not self._tldr_examples:
self.console.print(
f"[bold]No TLDR examples available for {self.command_key}.[/bold]"
)
return
is_cli_mode = self.options_manager.get("mode") in {
FalyxMode.RUN,
FalyxMode.PREVIEW,
FalyxMode.RUN_ALL,
FalyxMode.HELP,
}
is_cli_mode = self.options_manager.get("mode") != FalyxMode.MENU
program = self.program or "falyx"
program_style = self.options_manager.get("program_style") or self.command_style
command = self.aliases[0] if self.aliases else self.command_key
if self._is_help_command and is_cli_mode:
command = f"[{self.command_style}]{program} help[/{self.command_style}]"
command = f"[{program_style}]{program}[/{program_style}] [{self.command_style}]help[/{self.command_style}]"
elif is_cli_mode:
command = (
f"[{self.command_style}]{program} run {command}[/{self.command_style}]"
)
command = f"[{program_style}]{program}[/{program_style}] [{self.command_style}]{command}[/{self.command_style}]"
else:
command = f"[{self.command_style}]{command}[/{self.command_style}]"

View File

@@ -0,0 +1,175 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
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
@dataclass(slots=True)
class RootOptions:
verbose: bool = False
debug_hooks: bool = False
never_prompt: bool = False
version: bool = False
help: bool = False
class FalyxParser:
"""Root parser and command router for Falyx.
Responsibilities:
- parse global/root flags
- resolve built-ins vs registered commands
- normalize CLI input into ParseResult
- delegate command-specific parsing to CommandArgumentParser
"""
ROOT_FLAG_ALIASES: dict[str, str] = {
"--never-prompt": "never_prompt",
"-v": "verbose",
"--verbose": "verbose",
"--debug-hooks": "debug_hooks",
"?": "help",
"-h": "help",
"--help": "help",
}
def __init__(self, falyx: Falyx) -> None:
self.falyx = falyx
def _parse_root_options(
self,
argv: list[str],
) -> tuple[RootOptions, list[str]]:
"""Parse only root/session flags from the start of argv.
Parsing stops at the first token that is not a recognized root flag.
Remaining tokens are returned untouched for later routing.
Examples:
["--verbose", "deploy", "--env", "prod"]
-> (RootOptions(verbose=True), ["deploy", "--env", "prod"])
["deploy", "--verbose"]
-> (RootOptions(), ["deploy", "--verbose"])
"""
options = RootOptions()
remaining_start = 0
for index, token in enumerate(argv):
if token == "--":
remaining_start = index + 1
break
attr = self.ROOT_FLAG_ALIASES.get(token)
if attr is None:
remaining_start = index
break
setattr(options, attr, True)
else:
remaining_start = len(argv)
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:
argv = argv or []
root, remaining = self._parse_root_options(argv)
if root.help:
return ParseResult(
mode=FalyxMode.HELP,
raw_argv=argv,
never_prompt=root.never_prompt,
verbose=root.verbose,
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)

19
falyx/parser/group.py Normal file
View File

@@ -0,0 +1,19 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
from __future__ import annotations
from dataclasses import dataclass, field
@dataclass(slots=True)
class ArgumentGroup:
name: str
description: str = ""
dests: list[str] = field(default_factory=list)
@dataclass(slots=True)
class MutuallyExclusiveGroup:
name: str
required: bool = False
description: str = ""
dests: list[str] = field(default_factory=list)

View File

@@ -0,0 +1,24 @@
# 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:
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

View File

@@ -1,408 +0,0 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""
Provides the argument parser infrastructure for the Falyx CLI.
This module defines the `FalyxParsers` dataclass and related utilities for building
structured CLI interfaces with argparse. It supports top-level CLI commands like
`run`, `run-all`, `preview`, `help`, and `version`, and integrates seamlessly with
registered `Command` objects for dynamic help, usage generation, and argument handling.
Key Components:
- `FalyxParsers`: Container for all CLI subparsers.
- `get_arg_parsers()`: Factory for generating full parser suite.
- `get_root_parser()`: Creates the root-level CLI parser with global options.
- `get_subparsers()`: Helper to attach subcommand parsers to the root parser.
Used internally by the Falyx CLI `run()` entry point to parse arguments and route
execution across commands and workflows.
"""
from argparse import (
REMAINDER,
ArgumentParser,
Namespace,
RawDescriptionHelpFormatter,
_SubParsersAction,
)
from dataclasses import asdict, dataclass
from typing import Any, Sequence
from falyx.command import Command
@dataclass
class FalyxParsers:
"""Defines the argument parsers for the Falyx CLI."""
root: ArgumentParser
subparsers: _SubParsersAction
run: ArgumentParser
run_all: ArgumentParser
preview: ArgumentParser
help: 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)
def get_parser(self, name: str) -> ArgumentParser | None:
"""Get the parser by name."""
return self.as_dict().get(name)
def get_root_parser(
prog: str | None = "falyx",
usage: str | None = None,
description: str | None = "Falyx CLI - Run structured async command workflows.",
epilog: str | None = "Tip: Use 'falyx help' to show available commands.",
parents: Sequence[ArgumentParser] | None = None,
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,
) -> ArgumentParser:
"""
Construct the root-level ArgumentParser for the Falyx CLI.
This parser handles global arguments shared across subcommands and can serve
as the base parser for the Falyx CLI or standalone applications. It includes
options for verbosity, debug logging, and version output.
Args:
prog (str | None): Name of the program (e.g., 'falyx').
usage (str | None): Optional custom usage string.
description (str | None): Description shown in the CLI help.
epilog (str | None): Message displayed at the end of help output.
parents (Sequence[ArgumentParser] | None): Optional parent parsers.
prefix_chars (str): Characters to denote optional arguments (default: "-").
fromfile_prefix_chars (str | None): Prefix to indicate argument file input.
argument_default (Any): Global default value for arguments.
conflict_handler (str): Strategy to resolve conflicting argument names.
add_help (bool): Whether to include help (`-h/--help`) in this parser.
allow_abbrev (bool): Allow abbreviated long options.
exit_on_error (bool): Exit immediately on error or raise an exception.
Returns:
ArgumentParser: The root parser with global options attached.
Notes:
```
Includes the following arguments:
--never-prompt : Run in non-interactive mode.
-v / --verbose : Enable debug logging.
--debug-hooks : Enable hook lifecycle debug logs.
--version : Print the Falyx version.
```
"""
parser = ArgumentParser(
prog=prog,
usage=usage,
description=description,
epilog=epilog,
parents=parents if parents else [],
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(
"--never-prompt",
action="store_true",
help="Run in non-interactive mode with all prompts bypassed.",
)
parser.add_argument(
"-v", "--verbose", action="store_true", help=f"Enable debug logging for {prog}."
)
parser.add_argument(
"--debug-hooks",
action="store_true",
help="Enable default lifecycle debug logging",
)
parser.add_argument("--version", action="store_true", help=f"Show {prog} version")
return parser
def get_subparsers(
parser: ArgumentParser,
title: str = "Falyx Commands",
description: str | None = "Available commands for the Falyx CLI.",
) -> _SubParsersAction:
"""
Create and return a subparsers object for registering Falyx CLI subcommands.
This function adds a `subparsers` block to the given root parser, enabling
structured subcommands such as `run`, `run-all`, `preview`, etc.
Args:
parser (ArgumentParser): The root parser to attach the subparsers to.
title (str): Title used in help output to group subcommands.
description (str | None): Optional text describing the group of subcommands.
Returns:
_SubParsersAction: The subparsers object that can be used to add new CLI subcommands.
Raises:
TypeError: If `parser` is not an instance of `ArgumentParser`.
Example:
```python
>>> parser = get_root_parser()
>>> subparsers = get_subparsers(parser, title="Available Commands")
>>> subparsers.add_parser("run", help="Run a Falyx command")
```
"""
if not isinstance(parser, ArgumentParser):
raise TypeError("parser must be an instance of ArgumentParser")
subparsers = parser.add_subparsers(
title=title,
description=description,
dest="command",
)
return subparsers
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
) = "Tip: Use 'falyx preview [COMMAND]' to preview any command from the CLI.",
parents: Sequence[ArgumentParser] | None = None,
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,
commands: dict[str, Command] | None = None,
root_parser: ArgumentParser | None = None,
subparsers: _SubParsersAction | None = None,
) -> FalyxParsers:
"""
Create and return the full suite of argument parsers used by the Falyx CLI.
This function builds the root parser and all subcommand parsers used for structured
CLI workflows in Falyx. It supports standard subcommands including `run`, `run-all`,
`preview`, `help`, and `version`, and integrates with registered `Command` objects
to populate dynamic help and usage documentation.
Args:
prog (str | None): Program name to display in help and usage messages.
usage (str | None): Optional usage message to override the default.
description (str | None): Description for the CLI root parser.
epilog (str | None): Epilog message shown after the help text.
parents (Sequence[ArgumentParser] | None): Optional parent parsers.
prefix_chars (str): Characters that prefix optional arguments.
fromfile_prefix_chars (str | None): Prefix character for reading args from file.
argument_default (Any): Default value for arguments if not specified.
conflict_handler (str): Strategy for resolving conflicting arguments.
add_help (bool): Whether to add the `-h/--help` option to the root parser.
allow_abbrev (bool): Whether to allow abbreviated long options.
exit_on_error (bool): Whether the parser exits on error or raises.
commands (dict[str, Command] | None): Optional dictionary of registered commands
to populate help and subcommand descriptions dynamically.
root_parser (ArgumentParser | None): Custom root parser to use instead of building one.
subparsers (_SubParsersAction | None): Optional existing subparser object to extend.
Returns:
FalyxParsers: A structured container of all parsers, including `run`, `run-all`,
`preview`, `help`, `version`, and the root parser.
Raises:
TypeError: If `root_parser` is not an instance of ArgumentParser or
`subparsers` is not an instance of _SubParsersAction.
Example:
```python
>>> parsers = get_arg_parsers(commands=my_command_dict)
>>> args = parsers.root.parse_args()
```
Notes:
- This function integrates dynamic command usage and descriptions if the
`commands` argument is provided.
- The `run` parser supports additional options for retry logic and confirmation
prompts.
- The `run-all` parser executes all commands matching a tag.
- Use `falyx run ?[COMMAND]` from the CLI to preview a command.
"""
if epilog is None:
epilog = f"Tip: Use '{prog} help' to show available commands."
if root_parser is None:
parser = get_root_parser(
prog=prog,
usage=usage,
description=description,
epilog=epilog,
parents=parents,
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,
)
else:
if not isinstance(root_parser, ArgumentParser):
raise TypeError("root_parser must be an instance of ArgumentParser")
parser = root_parser
if subparsers is None:
if prog == "falyx":
subparsers = get_subparsers(
parser,
title="Falyx Commands",
description="Available commands for the Falyx CLI.",
)
else:
subparsers = get_subparsers(parser, title="subcommands", description=None)
if not isinstance(subparsers, _SubParsersAction):
raise TypeError("subparsers must be an instance of _SubParsersAction")
run_description = ["Run a command by its key or alias.\n"]
run_description.append("commands:")
if isinstance(commands, dict):
for command in commands.values():
run_description.append(command.usage)
command_description = command.help_text or command.description
run_description.append(f"{' '*24}{command_description}")
run_epilog = (
f"Tip: Use '{prog} preview [COMMAND]' to preview commands by their key or alias."
)
run_parser = subparsers.add_parser(
"run",
help="Run a specific command",
description="\n".join(run_description),
epilog=run_epilog,
formatter_class=RawDescriptionHelpFormatter,
)
run_parser.add_argument(
"name", help="Run a command by its key or alias", metavar="COMMAND"
)
run_parser.add_argument(
"--summary",
action="store_true",
help="Print an execution summary after command completes",
)
run_parser.add_argument(
"--retries", type=int, help="Number of retries on failure", default=0
)
run_parser.add_argument(
"--retry-delay",
type=float,
help="Initial delay between retries in (seconds)",
default=0,
)
run_parser.add_argument(
"--retry-backoff", type=float, help="Backoff factor for retries", default=0
)
run_group = run_parser.add_mutually_exclusive_group(required=False)
run_group.add_argument(
"-c",
"--confirm",
dest="force_confirm",
action="store_true",
help="Force confirmation prompts",
)
run_group.add_argument(
"-s",
"--skip-confirm",
dest="skip_confirm",
action="store_true",
help="Skip confirmation prompts",
)
run_parser.add_argument(
"command_args",
nargs=REMAINDER,
help="Arguments to pass to the command (if applicable)",
metavar="ARGS",
)
run_all_parser = subparsers.add_parser(
"run-all", help="Run all commands with a given tag"
)
run_all_parser.add_argument("-t", "--tag", required=True, help="Tag to match")
run_all_parser.add_argument(
"--summary",
action="store_true",
help="Print a summary after all tagged commands run",
)
run_all_parser.add_argument(
"--retries", type=int, help="Number of retries on failure", default=0
)
run_all_parser.add_argument(
"--retry-delay",
type=float,
help="Initial delay between retries in (seconds)",
default=0,
)
run_all_parser.add_argument(
"--retry-backoff", type=float, help="Backoff factor for retries", default=0
)
run_all_group = run_all_parser.add_mutually_exclusive_group(required=False)
run_all_group.add_argument(
"-c",
"--confirm",
dest="force_confirm",
action="store_true",
help="Force confirmation prompts",
)
run_all_group.add_argument(
"-s",
"--skip-confirm",
dest="skip_confirm",
action="store_true",
help="Skip confirmation prompts",
)
preview_parser = subparsers.add_parser(
"preview", help="Preview a command without running it"
)
preview_parser.add_argument("name", help="Key, alias, or description of the command")
help_parser = subparsers.add_parser("help", help="List all available commands")
help_parser.add_argument(
"-k",
"--key",
help="Show help for a specific command by its key or alias",
default=None,
)
help_parser.add_argument(
"-T",
"--tldr",
action="store_true",
help="Show a simplified TLDR examples of a command if available",
)
help_parser.add_argument(
"-t", "--tag", help="Filter commands by tag (case-insensitive)", default=None
)
version_parser = subparsers.add_parser("version", help=f"Show {prog} version")
return FalyxParsers(
root=parser,
subparsers=subparsers,
run=run_parser,
run_all=run_all_parser,
preview=preview_parser,
help=help_parser,
version=version_parser,
)

View File

@@ -29,15 +29,27 @@ def should_prompt_user(
*,
confirm: bool,
options: OptionsManager,
namespace: str = "cli_args",
):
namespace: str = "default",
override_namespace: str = "execution",
) -> bool:
"""Determine whether to prompt the user for confirmation.
Checks the `confirm` flag and consults the `OptionsManager` for any relevant
flags that may override the need for confirmation, such as `--never-prompt`,
`--force-confirm`, or `--skip-confirm`. The `override_namespace` is checked
first for any explicit overrides, followed by the main `namespace` for defaults.
"""
Determine whether to prompt the user for confirmation based on command
and global options.
"""
never_prompt = options.get("never_prompt", False, namespace)
force_confirm = options.get("force_confirm", False, namespace)
skip_confirm = options.get("skip_confirm", False, namespace)
never_prompt = options.get("never_prompt", None, override_namespace)
if never_prompt is None:
never_prompt = options.get("never_prompt", False, namespace)
force_confirm = options.get("force_confirm", None, override_namespace)
if force_confirm is None:
force_confirm = options.get("force_confirm", False, namespace)
skip_confirm = options.get("skip_confirm", None, override_namespace)
if skip_confirm is None:
skip_confirm = options.get("skip_confirm", False, namespace)
if never_prompt or skip_confirm:
return False

View File

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

View File

@@ -25,15 +25,15 @@ def build_tagged_table(flx: Falyx) -> Table:
# Group commands by first tag
grouped: dict[str, list[Command]] = defaultdict(list)
for cmd in flx.commands.values():
first_tag = cmd.tags[0] if cmd.tags else "Other"
grouped[first_tag.capitalize()].append(cmd)
for command in flx.commands.values():
first_tag = command.tags[0] if command.tags else "Other"
grouped[first_tag.capitalize()].append(command)
# Add grouped commands to table
for group_name, commands in grouped.items():
table.add_row(f"[bold underline]{group_name} Commands[/]")
for cmd in commands:
table.add_row(f"[{cmd.key}] [{cmd.style}]{cmd.description}")
for command in commands:
table.add_row(f"[{command.key}] [{command.style}]{command.description}")
table.add_row("")
# Add bottom row

View File

@@ -48,7 +48,9 @@ class CommandValidator(Validator):
message=self.error_message,
cursor_position=len(text),
)
is_preview, choice, _, __ = await self.falyx.get_command(text, from_validate=True)
is_preview, choice, _, __, ___ = await self.falyx.get_command(
text, from_validate=True
)
if is_preview:
return None
if not choice:

View File

@@ -1 +1 @@
__version__ = "0.1.87"
__version__ = "0.2.0"

View File

@@ -1,6 +1,6 @@
[tool.poetry]
name = "falyx"
version = "0.1.87"
version = "0.2.0"
description = "Reliable and introspectable async CLI action framework."
authors = ["Roland Thomas Jr <roland@rtj.dev>"]
license = "MIT"

View File

@@ -431,7 +431,6 @@ async def test_parse_args_flagged_nargs_plus():
assert args["files"] == ["a", "b", "c"]
args = await parser.parse_args(["--files", "a"])
print(args)
assert args["files"] == ["a"]
args = await parser.parse_args([])
@@ -666,7 +665,7 @@ async def test_parse_args_split_order():
cap.add_argument("a")
cap.add_argument("--x")
cap.add_argument("b", nargs="*")
args, kwargs = await cap.parse_args_split(["1", "--x", "100", "2"])
args, kwargs, _ = await cap.parse_args_split(["1", "--x", "100", "2"])
assert args == ("1", ["2"])
assert kwargs == {"x": "100"}

View File

@@ -1,57 +1,65 @@
from types import SimpleNamespace
import pytest
from prompt_toolkit.completion import Completion
from prompt_toolkit.document import Document
from falyx import Falyx
from falyx.completer import FalyxCompleter
from falyx.parser import CommandArgumentParser
@pytest.fixture
def fake_falyx():
fake_arg_parser = SimpleNamespace(
suggest_next=lambda tokens, end: ["--tag", "--name", "value with space"]
def falyx():
flx = Falyx()
parser = CommandArgumentParser(
command_key="R",
command_description="Run Command",
)
fake_command = SimpleNamespace(key="R", aliases=["RUN"], arg_parser=fake_arg_parser)
return SimpleNamespace(
exit_command=SimpleNamespace(key="X", aliases=["EXIT"]),
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},
parser.add_argument(
"--tag",
)
parser.add_argument(
"--name",
)
flx.add_command(
"R",
"Run Command",
lambda x: None,
aliases=["RUN"],
arg_parser=parser,
)
return flx
def test_suggest_commands(fake_falyx):
completer = FalyxCompleter(fake_falyx)
def test_suggest_commands(falyx):
completer = FalyxCompleter(falyx)
completions = list(completer._suggest_commands("R"))
assert any(c.text == "R" for c in completions)
assert any(c.text == "RUN" for c in completions)
def test_suggest_commands_empty(fake_falyx):
completer = FalyxCompleter(fake_falyx)
def test_suggest_commands_empty(falyx):
completer = FalyxCompleter(falyx)
completions = list(completer._suggest_commands(""))
assert any(c.text == "X" for c in completions)
assert any(c.text == "H" for c in completions)
def test_suggest_commands_no_match(fake_falyx):
completer = FalyxCompleter(fake_falyx)
def test_suggest_commands_no_match(falyx):
completer = FalyxCompleter(falyx)
completions = list(completer._suggest_commands("Z"))
assert not completions
def test_get_completions_no_input(fake_falyx):
completer = FalyxCompleter(fake_falyx)
def test_get_completions_no_input(falyx):
completer = FalyxCompleter(falyx)
doc = Document("")
results = list(completer.get_completions(doc, None))
assert any(isinstance(c, Completion) for c in results)
assert any(c.text == "X" for c in results)
def test_get_completions_no_match(fake_falyx):
completer = FalyxCompleter(fake_falyx)
def test_get_completions_no_match(falyx):
completer = FalyxCompleter(falyx)
doc = Document("Z")
completions = list(completer.get_completions(doc, None))
assert not completions
@@ -60,38 +68,38 @@ def test_get_completions_no_match(fake_falyx):
assert not completions
def test_get_completions_partial_command(fake_falyx):
completer = FalyxCompleter(fake_falyx)
def test_get_completions_partial_command(falyx):
completer = FalyxCompleter(falyx)
doc = Document("R")
results = list(completer.get_completions(doc, None))
assert any(c.text in ("R", "RUN") for c in results)
def test_get_completions_with_flag(fake_falyx):
completer = FalyxCompleter(fake_falyx)
def test_get_completions_with_flag(falyx):
completer = FalyxCompleter(falyx)
doc = Document("R ")
results = list(completer.get_completions(doc, None))
assert "--tag" in [c.text for c in results]
def test_get_completions_partial_flag(fake_falyx):
completer = FalyxCompleter(fake_falyx)
def test_get_completions_partial_flag(falyx):
completer = FalyxCompleter(falyx)
doc = Document("R --t")
results = list(completer.get_completions(doc, None))
assert all(c.start_position <= 0 for c in results)
assert any(c.text.startswith("--t") or c.display == "--tag" for c in results)
def test_get_completions_bad_input(fake_falyx):
completer = FalyxCompleter(fake_falyx)
def test_get_completions_bad_input(falyx):
completer = FalyxCompleter(falyx)
doc = Document('R "unclosed quote')
results = list(completer.get_completions(doc, None))
assert results == []
def test_get_completions_exception_handling(fake_falyx):
completer = FalyxCompleter(fake_falyx)
fake_falyx.commands["R"].arg_parser.suggest_next = lambda *args: 1 / 0
def test_get_completions_exception_handling(falyx):
completer = FalyxCompleter(falyx)
falyx.commands["R"].arg_parser.suggest_next = lambda *args: 1 / 0
doc = Document("R --tag")
results = list(completer.get_completions(doc, None))
assert results == []

View File

@@ -5,7 +5,7 @@ from falyx.action import Action
@pytest.mark.asyncio
async def test_run_key():
async def test_execute_command():
"""Test if Falyx can run in run key mode."""
falyx = Falyx("Run Key Test")
@@ -17,12 +17,12 @@ async def test_run_key():
)
# Run the CLI
result = await falyx.run_key("T")
result = await falyx.execute_command("T")
assert result == "Hello, World!"
@pytest.mark.asyncio
async def test_run_key_recover():
async def test_execute_command_recover():
"""Test if Falyx can recover from a failure in run key mode."""
falyx = Falyx("Run Key Recovery Test")
@@ -42,5 +42,5 @@ async def test_run_key_recover():
retry=True,
)
result = await falyx.run_key("E")
result = await falyx.execute_command("E")
assert result == "ok"

View File

@@ -1,6 +1,8 @@
import pytest
from rich.text import Text
from falyx import Falyx
from falyx.console import console
@pytest.mark.asyncio
@@ -8,7 +10,7 @@ async def test_help_command(capsys):
flx = Falyx()
assert flx.help_command.arg_parser.aliases[0] == "HELP"
assert flx.help_command.arg_parser.command_key == "H"
await flx.run_key("H")
await flx.execute_command("H")
captured = capsys.readouterr()
assert "Show this help menu" in captured.out
@@ -28,7 +30,7 @@ async def test_help_command_with_new_command(capsys):
aliases=["TEST"],
help_text="This is a new command.",
)
await flx.run_key("H")
await flx.execute_command("H")
captured = capsys.readouterr()
assert "This is a new command." in captured.out
@@ -70,12 +72,14 @@ async def test_help_command_by_tag(capsys):
tags=["tag1"],
help_text="This command is tagged.",
)
await flx.run_key("H", args=("tag1",))
await flx.execute_command("H -t tag1")
captured = capsys.readouterr()
assert "tag1" in captured.out
assert "This command is tagged." in captured.out
assert "HELP" not in captured.out
print(captured.out)
text = Text.from_ansi(captured.out)
assert "tag1" in text.plain
assert "This command is tagged." in text.plain
assert "HELP" not in text.plain
@pytest.mark.asyncio
@@ -88,9 +92,8 @@ async def test_help_command_empty_tags(capsys):
flx.add_command(
"U", "Untagged Command", untagged_command, help_text="This command has no tags."
)
await flx.run_key("H", args=("nonexistent_tag",))
await flx.execute_command("H nonexistent_tag")
captured = capsys.readouterr()
print(captured.out)
assert "nonexistent_tag" in captured.out
assert "Nothing to show here" in captured.out
text = Text.from_ansi(captured.out)
assert "Unexpected positional argument: nonexistent_tag" in text.plain

View File

@@ -3,17 +3,14 @@ import sys
import pytest
from falyx import Falyx
from falyx.parser import get_arg_parsers
@pytest.mark.asyncio
async def test_run_basic(capsys):
sys.argv = ["falyx", "run", "-h"]
falyx_parsers = get_arg_parsers()
assert falyx_parsers is not None, "Falyx parsers should be initialized"
sys.argv = ["falyx", "-h"]
flx = Falyx()
with pytest.raises(SystemExit):
await flx.run(falyx_parsers)
await flx.run()
captured = capsys.readouterr()
assert "Run a command by its key or alias." in captured.out
assert "Show this help menu." in captured.out

View File

@@ -0,0 +1,47 @@
from falyx import Falyx
from falyx.parser.falyx_parser import FalyxParser, RootOptions
def get_falyx_parser():
falyx = Falyx()
return FalyxParser(falyx=falyx)
def test_parse_root_options_empty():
parser = get_falyx_parser()
opts, remaining = parser._parse_root_options([])
assert opts == RootOptions()
assert remaining == []
def test_parse_root_options_consumes_known_leading_flags():
parser = get_falyx_parser()
opts, remaining = parser._parse_root_options(
["--verbose", "--never-prompt", "deploy", "--env", "prod"]
)
assert opts.verbose is True
assert opts.never_prompt is True
assert remaining == ["deploy", "--env", "prod"]
def test_parse_root_options_stops_at_first_non_root_token():
parser = get_falyx_parser()
opts, remaining = parser._parse_root_options(["deploy", "--verbose"])
assert opts == RootOptions()
assert remaining == ["deploy", "--verbose"]
def test_parse_root_options_supports_help():
parser = get_falyx_parser()
opts, remaining = parser._parse_root_options(["--help"])
assert opts.help is True
assert remaining == []
def test_parse_root_options_supports_double_dash_separator():
parser = get_falyx_parser()
opts, remaining = parser._parse_root_options(
["--verbose", "--", "deploy", "--verbose"]
)
assert opts.verbose is True
assert remaining == ["deploy", "--verbose"]

View File

@@ -1,19 +1,11 @@
import shutil
import sys
import tempfile
from argparse import ArgumentParser, Namespace, _SubParsersAction
from pathlib import Path
import pytest
from falyx.__main__ import (
bootstrap,
find_falyx_config,
get_parsers,
init_callback,
init_config,
main,
)
from falyx.__main__ import bootstrap, find_falyx_config, init_config, main
from falyx.parser import CommandArgumentParser
@@ -94,38 +86,10 @@ async def test_init_config():
assert args["name"] == "."
def test_init_callback(tmp_path):
"""Test if the init_callback function works correctly."""
# Test project initialization
args = Namespace(command="init", name=str(tmp_path))
init_callback(args)
assert (tmp_path / "falyx.yaml").exists()
def test_init_global_callback():
# Test global initialization
args = Namespace(command="init_global")
init_callback(args)
assert (Path.home() / ".config" / "falyx" / "tasks.py").exists()
assert (Path.home() / ".config" / "falyx" / "falyx.yaml").exists()
def test_get_parsers():
"""Test if the get_parsers function returns the correct parsers."""
root_parser, subparsers = get_parsers()
assert isinstance(root_parser, ArgumentParser)
assert isinstance(subparsers, _SubParsersAction)
# Check if the 'init' command is available
init_parser = subparsers.choices.get("init")
assert init_parser is not None
assert "name" == init_parser._get_positional_actions()[0].dest
def test_main():
"""Test if the main function runs with the correct arguments."""
sys.argv = ["falyx", "run", "?"]
sys.argv = ["falyx", "?"]
with pytest.raises(SystemExit) as exc_info:
main()

View File

@@ -71,22 +71,28 @@ async def test_action_with_nargs_positional():
return int(a) * int(b)
action = Action("multiply", multiply)
parser.add_argument("mul", action=ArgumentAction.ACTION, resolver=action, nargs=2)
parser.add_argument(
"mul",
action=ArgumentAction.ACTION,
resolver=action,
nargs=2,
type=int,
)
args = await parser.parse_args(["3", "4"])
assert args["mul"] == 12
with pytest.raises(CommandArgumentError):
await parser.parse_args(["3"])
with pytest.raises(CommandArgumentError):
await parser.parse_args([])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["3", "4", "5"])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["--mul", "3", "4"])
with pytest.raises(CommandArgumentError):
await parser.parse_args([])
@pytest.mark.asyncio
async def test_action_with_nargs_positional_int():
@@ -102,6 +108,9 @@ async def test_action_with_nargs_positional_int():
args = await parser.parse_args(["3", "4"])
assert args["mul"] == 12
with pytest.raises(CommandArgumentError):
await parser.parse_args([])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["3"])
@@ -209,11 +218,19 @@ async def test_action_with_default_and_value_not():
@pytest.mark.asyncio
async def test_action_with_default_and_value_positional():
parser = CommandArgumentParser()
action = Action("default", lambda: "default_value")
parser.add_argument("default", action=ArgumentAction.ACTION, resolver=action)
action = Action("action", lambda x: x)
parser.add_argument(
"default",
action=ArgumentAction.ACTION,
resolver=action,
default="default_value",
)
args = await parser.parse_args([])
assert args["default"] == "default_value"
args = await parser.parse_args(["be"])
assert args["default"] == "be"
with pytest.raises(CommandArgumentError):
await parser.parse_args([])
with pytest.raises(CommandArgumentError):
await parser.parse_args(["be"])
await parser.parse_args(["one", "new_value"])

View File

@@ -10,7 +10,7 @@ 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_falyx.get_command.return_value = (False, object(), (), {}, {})
validator = CommandValidator(fake_falyx, "Invalid!")
await validator.validate_async(Document("valid"))
@@ -20,7 +20,7 @@ async def test_command_validator_validates_command():
@pytest.mark.asyncio
async def test_command_validator_rejects_invalid_command():
fake_falyx = AsyncMock()
fake_falyx.get_command.return_value = (False, None, (), {})
fake_falyx.get_command.return_value = (False, None, (), {}, {})
validator = CommandValidator(fake_falyx, "Invalid!")
with pytest.raises(ValidationError):
@@ -33,7 +33,7 @@ async def test_command_validator_rejects_invalid_command():
@pytest.mark.asyncio
async def test_command_validator_is_preview():
fake_falyx = AsyncMock()
fake_falyx.get_command.return_value = (True, None, (), {})
fake_falyx.get_command.return_value = (True, None, (), {}, {})
validator = CommandValidator(fake_falyx, "Invalid!")
await validator.validate_async(Document("?preview_command"))