Add help_text for commands to argparse run subcommand, change the way Falyx.run works and you can only pass FalyxParsers

This commit is contained in:
Roland Thomas Jr 2025-05-30 00:36:55 -04:00
parent 8a3c1d6cc8
commit c2eb854e5a
Signed by: roland
GPG Key ID: 7C3C2B085A4C2872
11 changed files with 168 additions and 119 deletions

View File

@ -3,7 +3,7 @@ commands:
description: Pipeline Demo
action: pipeline_demo.pipeline
tags: [pipeline, demo]
help_text: Run Demployment Pipeline with retries.
help_text: Run Deployment Pipeline with retries.
- key: G
description: Run HTTP Action Group

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 Namespace
from pathlib import Path
from typing import Any
from falyx.config import loader
from falyx.falyx import Falyx
from falyx.parsers import FalyxParsers, get_arg_parsers
from falyx.parsers import CommandArgumentParser
def find_falyx_config() -> Path | None:
@ -39,45 +38,40 @@ def bootstrap() -> Path | None:
return config_path
def get_falyx_parsers() -> FalyxParsers:
falyx_parsers: FalyxParsers = get_arg_parsers()
init_parser = falyx_parsers.subparsers.add_parser(
"init", help="Create a new Falyx CLI project"
def init_config(parser: CommandArgumentParser) -> None:
parser.add_argument(
"name",
type=str,
help="Name of the new Falyx project",
default=".",
nargs="?",
)
init_parser.add_argument("name", nargs="?", default=".", help="Project directory")
falyx_parsers.subparsers.add_parser(
"init-global", help="Set up ~/.config/falyx with example tasks"
)
return falyx_parsers
def run(args: Namespace) -> Any:
if args.command == "init":
from falyx.init import init_project
init_project(args.name)
return
if args.command == "init-global":
from falyx.init import init_global
init_global()
return
def main() -> Any:
bootstrap_path = bootstrap()
if not bootstrap_path:
print("No Falyx config file found. Exiting.")
return None
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,
)
flx.add_command(
"G",
"Initialize Falyx global configuration",
init_global,
aliases=["init-global"],
)
else:
flx = loader(bootstrap_path)
flx: Falyx = loader(bootstrap_path)
return asyncio.run(flx.run())
def main():
parsers = get_falyx_parsers()
args = parsers.parse_args()
run(args)
if __name__ == "__main__":
main()

View File

@ -307,6 +307,14 @@ class Command(BaseModel):
return FormattedText(prompt)
@property
def usage(self) -> str:
"""Generate a help string for the command arguments."""
if not self.arg_parser:
return "No arguments defined."
return self.arg_parser.get_usage(plain_text=True)
def log_summary(self) -> None:
if self._context:
self._context.log_summary()

View File

@ -18,6 +18,7 @@ from falyx.action.base import BaseAction
from falyx.command import Command
from falyx.falyx import Falyx
from falyx.logger import logger
from falyx.parsers import CommandArgumentParser
from falyx.retry import RetryPolicy
from falyx.themes import OneColors
@ -101,6 +102,7 @@ class RawCommand(BaseModel):
retry_policy: RetryPolicy = Field(default_factory=RetryPolicy)
hidden: bool = False
help_text: str = ""
help_epilogue: str = ""
@field_validator("retry_policy")
@classmethod
@ -116,6 +118,14 @@ def convert_commands(raw_commands: list[dict[str, Any]]) -> list[Command]:
commands = []
for entry in raw_commands:
raw_command = RawCommand(**entry)
parser = CommandArgumentParser(
command_key=raw_command.key,
command_description=raw_command.description,
command_style=raw_command.style,
help_text=raw_command.help_text,
help_epilogue=raw_command.help_epilogue,
aliases=raw_command.aliases,
)
commands.append(
Command.model_validate(
{
@ -123,9 +133,11 @@ def convert_commands(raw_commands: list[dict[str, Any]]) -> list[Command]:
"action": wrap_if_needed(
import_action(raw_command.action), name=raw_command.description
),
"arg_parser": parser,
}
)
)
return commands

View File

@ -59,7 +59,7 @@ from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import Hook, HookManager, HookType
from falyx.logger import logger
from falyx.options_manager import OptionsManager
from falyx.parsers import CommandArgumentParser, get_arg_parsers
from falyx.parsers import CommandArgumentParser, FalyxParsers, get_arg_parsers
from falyx.protocols import ArgParserProtocol
from falyx.retry import RetryPolicy
from falyx.signals import BackSignal, CancelSignal, HelpSignal, QuitSignal
@ -152,6 +152,11 @@ class Falyx:
self,
title: str | Markdown = "Menu",
*,
program: str | None = "falyx",
usage: str | None = None,
description: str | None = "Falyx CLI - Run structured async command workflows.",
epilog: str | None = None,
version: str = __version__,
prompt: str | AnyFormattedText = "> ",
columns: int = 3,
bottom_bar: BottomBar | str | Callable[[], Any] | None = None,
@ -170,6 +175,11 @@ class Falyx:
) -> None:
"""Initializes the Falyx object."""
self.title: str | Markdown = title
self.program: str | None = program
self.usage: str | None = usage
self.description: str | None = description
self.epilog: str | None = epilog
self.version: str = version
self.prompt: str | AnyFormattedText = prompt
self.columns: int = columns
self.commands: dict[str, Command] = CaseInsensitiveDict()
@ -1015,12 +1025,35 @@ class Falyx:
if self.exit_message:
self.print_message(self.exit_message)
async def run(self) -> None:
async def run(
self,
falyx_parsers: FalyxParsers | None = None,
callback: Callable[..., Any] | None = None,
) -> None:
"""Run Falyx CLI with structured subcommands."""
if not self.cli_args:
self.cli_args = get_arg_parsers().root.parse_args()
if self.cli_args:
raise FalyxError(
"Run is incompatible with CLI arguments. Use 'run_key' instead."
)
if falyx_parsers:
if not isinstance(falyx_parsers, FalyxParsers):
raise FalyxError("falyx_parsers must be an instance of FalyxParsers.")
else:
falyx_parsers = get_arg_parsers(
self.program,
self.usage,
self.description,
self.epilog,
commands=self.commands,
)
self.cli_args = falyx_parsers.parse_args()
self.options.from_namespace(self.cli_args, "cli_args")
if callback:
if not callable(callback):
raise FalyxError("Callback must be a callable function.")
callback(self.cli_args)
if not self.options.get("never_prompt"):
self.options.set("never_prompt", self._never_prompt)
@ -1075,11 +1108,24 @@ class Falyx:
args, kwargs = await command.parse_args(self.cli_args.command_args)
except HelpSignal:
sys.exit(0)
except CommandArgumentError as error:
self.console.print(f"[{OneColors.DARK_RED}]❌ ['{command.key}'] {error}")
command.show_help()
sys.exit(1)
try:
await self.run_key(self.cli_args.name, args=args, kwargs=kwargs)
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(0)
except BackSignal:
logger.info("[BackSignal]. <- Exiting run.")
sys.exit(0)
except CancelSignal:
logger.info("[CancelSignal]. <- Exiting run.")
sys.exit(0)
if self.cli_args.summary:
er.summary()

View File

@ -101,7 +101,7 @@ commands:
console = Console(color_system="auto")
def init_project(name: str = ".") -> None:
def init_project(name: str) -> None:
target = Path(name).resolve()
target.mkdir(parents=True, exist_ok=True)

View File

@ -168,6 +168,7 @@ class CommandArgumentParser:
self._arguments: list[Argument] = []
self._positional: dict[str, Argument] = {}
self._keyword: dict[str, Argument] = {}
self._keyword_list: list[Argument] = []
self._flag_map: dict[str, Argument] = {}
self._dest_set: set[str] = set()
self._add_help()
@ -488,6 +489,8 @@ class CommandArgumentParser:
self._arguments.append(argument)
if positional:
self._positional[dest] = argument
else:
self._keyword_list.append(argument)
def get_argument(self, dest: str) -> Argument | None:
return next((a for a in self._arguments if a.dest == dest), None)
@ -832,11 +835,11 @@ class CommandArgumentParser:
kwargs_dict[arg.dest] = parsed[arg.dest]
return tuple(args_list), kwargs_dict
def render_help(self) -> None:
def get_options_text(self, plain_text=False) -> str:
# Options
# Add all keyword arguments to the options list
options_list = []
for arg in self._keyword.values():
for arg in self._keyword_list:
choice_text = arg.get_choice_text()
if choice_text:
options_list.extend([f"[{arg.flags[0]} {choice_text}]"])
@ -848,19 +851,39 @@ class CommandArgumentParser:
choice_text = arg.get_choice_text()
if isinstance(arg.nargs, int):
choice_text = " ".join([choice_text] * arg.nargs)
options_list.append(escape(choice_text))
if plain_text:
options_list.append(choice_text)
else:
options_list.append(escape(choice_text))
options_text = " ".join(options_list)
command_keys = " | ".join(
[f"[{self.command_style}]{self.command_key}[/{self.command_style}]"]
+ [
f"[{self.command_style}]{alias}[/{self.command_style}]"
for alias in self.aliases
]
)
return " ".join(options_list)
usage = f"usage: {command_keys} {options_text}"
self.console.print(f"[bold]{usage}[/bold]\n")
def get_command_keys_text(self, plain_text=False) -> str:
if plain_text:
command_keys = " | ".join(
[f"{self.command_key}"] + [f"{alias}" for alias in self.aliases]
)
else:
command_keys = " | ".join(
[f"[{self.command_style}]{self.command_key}[/{self.command_style}]"]
+ [
f"[{self.command_style}]{alias}[/{self.command_style}]"
for alias in self.aliases
]
)
return command_keys
def get_usage(self, plain_text=False) -> str:
"""Get the usage text for the command."""
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}"
return command_keys
def render_help(self) -> None:
usage = self.get_usage()
self.console.print(f"[bold]usage: {usage}[/bold]\n")
# Description
if self.help_text:
@ -877,7 +900,7 @@ class CommandArgumentParser:
arg_line.append(help_text)
self.console.print(arg_line)
self.console.print("[bold]options:[/bold]")
for arg in self._keyword.values():
for arg in self._keyword_list:
flags = ", ".join(arg.flags)
flags_choice = f"{flags} {arg.get_choice_text()}"
arg_line = Text(f" {flags_choice:<30} ")

View File

@ -2,10 +2,18 @@
"""parsers.py
This module contains the argument parsers used for the Falyx CLI.
"""
from argparse import REMAINDER, ArgumentParser, Namespace, _SubParsersAction
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:
@ -47,6 +55,7 @@ def get_arg_parsers(
add_help: bool = True,
allow_abbrev: bool = True,
exit_on_error: bool = True,
commands: dict[str, Command] | None = None,
) -> FalyxParsers:
"""Returns the argument parser for the CLI."""
parser = ArgumentParser(
@ -79,8 +88,25 @@ def get_arg_parsers(
parser.add_argument("--version", action="store_true", help="Show Falyx version")
subparsers = parser.add_subparsers(dest="command")
run_parser = subparsers.add_parser("run", help="Run a specific command")
run_parser.add_argument("name", help="Key, alias, or description of the command")
run_description = "Run a command by its key or alias."
run_epilog = ["commands:"]
if isinstance(commands, dict):
for command in commands.values():
run_epilog.append(command.usage)
command_description = command.description or command.help_text
run_epilog.append(f" {command_description}")
run_epilog.append(" ")
run_epilog.append(
"Tip: Use 'falyx run ?[COMMAND]' to preview commands by their key or alias."
)
run_parser = subparsers.add_parser(
"run",
help="Run a specific command",
description=run_description,
epilog="\n".join(run_epilog),
formatter_class=RawDescriptionHelpFormatter,
)
run_parser.add_argument("name", help="Run a command by its key or alias")
run_parser.add_argument(
"--summary",
action="store_true",

View File

@ -1 +1 @@
__version__ = "0.1.40"
__version__ = "0.1.41"

View File

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

View File

@ -5,7 +5,7 @@ from pathlib import Path
import pytest
from falyx.__main__ import bootstrap, find_falyx_config, get_falyx_parsers, run
from falyx.__main__ import bootstrap, find_falyx_config, main
def test_find_falyx_config():
@ -50,63 +50,3 @@ def test_bootstrap_with_global_config():
assert str(config_file.parent) in sys.path
config_file.unlink()
sys.path = sys_path_before
def test_parse_args():
"""Test if the parse_args function works correctly."""
falyx_parsers = get_falyx_parsers()
args = falyx_parsers.parse_args(["init", "test_project"])
assert args.command == "init"
assert args.name == "test_project"
args = falyx_parsers.parse_args(["init-global"])
assert args.command == "init-global"
def test_run():
"""Test if the run function works correctly."""
falyx_parsers = get_falyx_parsers()
args = falyx_parsers.parse_args(["init", "test_project"])
run(args)
assert args.command == "init"
assert args.name == "test_project"
# Check if the project directory was created
assert Path("test_project").exists()
# Clean up
(Path("test_project") / "falyx.yaml").unlink()
(Path("test_project") / "tasks.py").unlink()
Path("test_project").rmdir()
# Test init-global
args = falyx_parsers.parse_args(["init-global"])
run(args)
# Check if the global config directory was created
assert (Path.home() / ".config" / "falyx" / "falyx.yaml").exists()
# Clean up
(Path.home() / ".config" / "falyx" / "falyx.yaml").unlink()
(Path.home() / ".config" / "falyx" / "tasks.py").unlink()
(Path.home() / ".config" / "falyx").rmdir()
def test_no_bootstrap():
"""Test if the main function works correctly when no config file is found."""
falyx_parsers = get_falyx_parsers()
args = falyx_parsers.parse_args(["list"])
assert run(args) is None
# Check if the task was run
assert args.command == "list"
def test_run_test_project():
"""Test if the main function works correctly with a test project."""
falyx_parsers = get_falyx_parsers()
args = falyx_parsers.parse_args(["init", "test_project"])
run(args)
args = falyx_parsers.parse_args(["run", "B"])
os.chdir("test_project")
with pytest.raises(SystemExit):
assert run(args) == "Build complete!"
os.chdir("..")
shutil.rmtree("test_project")
assert not Path("test_project").exists()