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

Add cancel for SelectionActions, Add args/kwargs to ActionFactoryAction, remove requires_input detection, add return types to SelectionAction, add option to hide_menu_table

This commit is contained in:
Roland Thomas Jr 2025-05-21 23:18:45 -04:00
parent 3c0a81359c
commit b51ba87999
Signed by: roland
GPG Key ID: 7C3C2B085A4C2872
24 changed files with 313 additions and 261 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
examples/http_demo.py
View File

@ -4,7 +4,6 @@ from rich.console import Console
from falyx import ActionGroup, Falyx
from falyx.action import HTTPAction
from falyx.hook_manager import HookType
from falyx.hooks import ResultReporter
console = Console()
@ -49,7 +48,7 @@ action_group = ActionGroup(
reporter = ResultReporter()
action_group.hooks.register(
HookType.ON_SUCCESS,
"on_success",
reporter.report,
)

3
examples/pipeline_demo.py
View File

@ -3,7 +3,6 @@ import asyncio
from falyx import Action, ActionGroup, ChainedAction
from falyx import ExecutionRegistry as er
from falyx import ProcessAction
from falyx.hook_manager import HookType
from falyx.retry import RetryHandler, RetryPolicy
@ -47,7 +46,7 @@ def build_pipeline():
checkout = Action("Checkout", checkout_code)
analysis = ProcessAction("Static Analysis", run_static_analysis)
tests = Action("Run Tests", flaky_tests)
tests.hooks.register(HookType.ON_ERROR, retry_handler.retry_on_error)
tests.hooks.register("on_error", retry_handler.retry_on_error)
# Parallel deploys
deploy_group = ActionGroup(

34
examples/selection_demo.py
View File

@ -1,22 +1,26 @@
import asyncio
from falyx.selection import (
SelectionOption,
prompt_for_selection,
render_selection_dict_table,
)
from falyx.action import SelectionAction
from falyx.selection import SelectionOption
menu = {
"A": SelectionOption("Run diagnostics", lambda: print("Running diagnostics...")),
"B": SelectionOption("Deploy to staging", lambda: print("Deploying...")),
selections = {
"1": SelectionOption(
description="Production", value="3bc2616e-3696-11f0-a139-089204eb86ac"
),
"2": SelectionOption(
description="Staging", value="42f2cd84-3696-11f0-a139-089204eb86ac"
),
}
table = render_selection_dict_table(
title="Main Menu",
selections=menu,
select = SelectionAction(
name="Select Deployment",
selections=selections,
title="Select a Deployment",
columns=2,
prompt_message="> ",
return_type="value",
show_table=True,
)
key = asyncio.run(prompt_for_selection(menu.keys(), table))
print(f"You selected: {key}")
menu[key.upper()].value()
print(asyncio.run(select()))

5
examples/shell_example.py
View File

@ -3,7 +3,6 @@ import asyncio
from falyx import Action, ChainedAction, Falyx
from falyx.action import ShellAction
from falyx.hook_manager import HookType
from falyx.hooks import ResultReporter
from falyx.utils import setup_logging
@ -42,12 +41,12 @@ reporter = ResultReporter()
a1 = Action("a1", a1, inject_last_result=True)
a1.hooks.register(
HookType.ON_SUCCESS,
"on_success",
reporter.report,
)
a2 = Action("a2", a2, inject_last_result=True)
a2.hooks.register(
HookType.ON_SUCCESS,
"on_success",
reporter.report,
)

1
falyx/__init__.py
View File

@ -12,7 +12,6 @@ from .command import Command
from .context import ExecutionContext, SharedContext
from .execution_registry import ExecutionRegistry
from .falyx import Falyx
from .hook_manager import HookType
logger = logging.getLogger("falyx")

26
falyx/action/action.py
View File

@ -62,8 +62,7 @@ class BaseAction(ABC):
inject_last_result (bool): Whether to inject the previous action's result
into kwargs.
inject_into (str): The name of the kwarg key to inject the result as
(default: 'last_result').
_requires_injection (bool): Whether the action requires input injection.
(default: 'last_result').
"""
def __init__(
@ -83,7 +82,6 @@ class BaseAction(ABC):
self.inject_last_result: bool = inject_last_result
self.inject_into: str = inject_into
self._never_prompt: bool = never_prompt
self._requires_injection: bool = False
self._skip_in_chain: bool = False
self.console = Console(color_system="auto")
self.options_manager: OptionsManager | None = None
@ -103,7 +101,7 @@ class BaseAction(ABC):
raise NotImplementedError("preview must be implemented by subclasses")
@abstractmethod
def get_infer_target(self) -> Callable[..., Any] | None:
def get_infer_target(self) -> tuple[Callable[..., Any] | None, dict[str, Any] | None]:
"""
Returns the callable to be used for argument inference.
By default, it returns None.
@ -163,10 +161,6 @@ class BaseAction(ABC):
async def _write_stdout(self, data: str) -> None:
"""Override in subclasses that produce terminal output."""
def requires_io_injection(self) -> bool:
"""Checks to see if the action requires input injection."""
return self._requires_injection
def __repr__(self) -> str:
return str(self)
@ -255,12 +249,12 @@ class Action(BaseAction):
if policy.enabled:
self.enable_retry()
def get_infer_target(self) -> Callable[..., Any]:
def get_infer_target(self) -> tuple[Callable[..., Any], None]:
"""
Returns the callable to be used for argument inference.
By default, it returns the action itself.
"""
return self.action
return self.action, None
async def _run(self, *args, **kwargs) -> Any:
combined_args = args + self.args
@ -493,10 +487,10 @@ class ChainedAction(BaseAction, ActionListMixin):
if hasattr(action, "register_teardown") and callable(action.register_teardown):
action.register_teardown(self.hooks)
def get_infer_target(self) -> Callable[..., Any] | None:
def get_infer_target(self) -> tuple[Callable[..., Any] | None, dict[str, Any] | None]:
if self.actions:
return self.actions[0].get_infer_target()
return None
return None, None
def _clear_args(self):
return (), {}
@ -690,7 +684,7 @@ class ActionGroup(BaseAction, ActionListMixin):
if hasattr(action, "register_teardown") and callable(action.register_teardown):
action.register_teardown(self.hooks)
def get_infer_target(self) -> Callable[..., Any] | None:
def get_infer_target(self) -> tuple[Callable[..., Any] | None, dict[str, Any] | None]:
arg_defs = same_argument_definitions(self.actions)
if arg_defs:
return self.actions[0].get_infer_target()
@ -698,7 +692,7 @@ class ActionGroup(BaseAction, ActionListMixin):
"[%s] auto_args disabled: mismatched ActionGroup arguments",
self.name,
)
return None
return None, None
async def _run(self, *args, **kwargs) -> list[tuple[str, Any]]:
shared_context = SharedContext(name=self.name, action=self, is_parallel=True)
@ -818,8 +812,8 @@ class ProcessAction(BaseAction):
self.executor = executor or ProcessPoolExecutor()
self.is_retryable = True
def get_infer_target(self) -> Callable[..., Any] | None:
return self.action
def get_infer_target(self) -> tuple[Callable[..., Any] | None, None]:
return self.action, None
async def _run(self, *args, **kwargs) -> Any:
if self.inject_last_result and self.shared_context:

8
falyx/action/action_factory.py
View File

@ -35,6 +35,8 @@ class ActionFactoryAction(BaseAction):
*,
inject_last_result: bool = False,
inject_into: str = "last_result",
args: tuple[Any, ...] = (),
kwargs: dict[str, Any] | None = None,
preview_args: tuple[Any, ...] = (),
preview_kwargs: dict[str, Any] | None = None,
):
@ -44,6 +46,8 @@ class ActionFactoryAction(BaseAction):
inject_into=inject_into,
)
self.factory = factory
self.args = args
self.kwargs = kwargs or {}
self.preview_args = preview_args
self.preview_kwargs = preview_kwargs or {}
@ -55,8 +59,8 @@ class ActionFactoryAction(BaseAction):
def factory(self, value: ActionFactoryProtocol):
self._factory = ensure_async(value)
def get_infer_target(self) -> Callable[..., Any]:
return self.factory
def get_infer_target(self) -> tuple[Callable[..., Any], None]:
return self.factory, None
async def _run(self, *args, **kwargs) -> Any:
updated_kwargs = self._maybe_inject_last_result(kwargs)

12
falyx/action/io_action.py
View File

@ -73,7 +73,6 @@ class BaseIOAction(BaseAction):
inject_last_result=inject_last_result,
)
self.mode = mode
self._requires_injection = True
def from_input(self, raw: str | bytes) -> Any:
raise NotImplementedError
@ -99,8 +98,8 @@ class BaseIOAction(BaseAction):
)
raise FalyxError("No input provided and no last result to inject.")
def get_infer_target(self) -> Callable[..., Any] | None:
return None
def get_infer_target(self) -> tuple[Callable[..., Any] | None, dict[str, Any] | None]:
return None, None
async def __call__(self, *args, **kwargs):
context = ExecutionContext(
@ -198,7 +197,6 @@ class ShellAction(BaseIOAction):
- Captures stdout and stderr from shell execution
- Raises on non-zero exit codes with stderr as the error
- Result is returned as trimmed stdout string
- Compatible with ChainedAction and Command.requires_input detection
Args:
name (str): Name of the action.
@ -223,10 +221,10 @@ class ShellAction(BaseIOAction):
)
return raw.strip() if isinstance(raw, str) else raw.decode("utf-8").strip()
def get_infer_target(self) -> Callable[..., Any] | None:
def get_infer_target(self) -> tuple[Callable[..., Any] | None, dict[str, Any] | None]:
if sys.stdin.isatty():
return self._run
return None
return self._run, {"parsed_input": {"help": self.command_template}}
return None, None
async def _run(self, parsed_input: str) -> str:
# Replace placeholder in template, or use raw input as full command

4
falyx/action/menu_action.py
View File

@ -73,8 +73,8 @@ class MenuAction(BaseAction):
table.add_row(*row)
return table
def get_infer_target(self) -> None:
return None
def get_infer_target(self) -> tuple[None, None]:
return None, None
async def _run(self, *args, **kwargs) -> Any:
kwargs = self._maybe_inject_last_result(kwargs)

40
falyx/action/select_file_action.py
View File

@ -25,6 +25,7 @@ from falyx.selection import (
prompt_for_selection,
render_selection_dict_table,
)
from falyx.signals import CancelSignal
from falyx.themes import OneColors
@ -121,8 +122,15 @@ class SelectFileAction(BaseAction):
logger.warning("[ERROR] Failed to parse %s: %s", file.name, error)
return options
def get_infer_target(self) -> None:
return None
def _find_cancel_key(self, options) -> str:
"""Return first numeric value not already used in the selection dict."""
for index in range(len(options)):
if str(index) not in options:
return str(index)
return str(len(options))
def get_infer_target(self) -> tuple[None, None]:
return None, None
async def _run(self, *args, **kwargs) -> Any:
context = ExecutionContext(name=self.name, args=args, kwargs=kwargs, action=self)
@ -131,28 +139,38 @@ class SelectFileAction(BaseAction):
await self.hooks.trigger(HookType.BEFORE, context)
files = [
f
for f in self.directory.iterdir()
if f.is_file()
and (self.suffix_filter is None or f.suffix == self.suffix_filter)
file
for file in self.directory.iterdir()
if file.is_file()
and (self.suffix_filter is None or file.suffix == self.suffix_filter)
]
if not files:
raise FileNotFoundError("No files found in directory.")
options = self.get_options(files)
cancel_key = self._find_cancel_key(options)
cancel_option = {
cancel_key: SelectionOption(
description="Cancel", value=CancelSignal(), style=OneColors.DARK_RED
)
}
table = render_selection_dict_table(
title=self.title, selections=options, columns=self.columns
title=self.title, selections=options | cancel_option, columns=self.columns
)
key = await prompt_for_selection(
options.keys(),
(options | cancel_option).keys(),
table,
console=self.console,
prompt_session=self.prompt_session,
prompt_message=self.prompt_message,
)
if key == cancel_key:
raise CancelSignal("User canceled the selection.")
result = options[key].value
context.result = result
await self.hooks.trigger(HookType.ON_SUCCESS, context)
@ -179,11 +197,11 @@ class SelectFileAction(BaseAction):
try:
files = list(self.directory.iterdir())
if self.suffix_filter:
files = [f for f in files if f.suffix == self.suffix_filter]
files = [file for file in files if file.suffix == self.suffix_filter]
sample = files[:10]
file_list = tree.add("[dim]Files:[/]")
for f in sample:
file_list.add(f"[dim]{f.name}[/]")
for file in sample:
file_list.add(f"[dim]{file.name}[/]")
if len(files) > 10:
file_list.add(f"[dim]... ({len(files) - 10} more)[/]")
except Exception as error:

103
falyx/action/selection_action.py
View File

@ -7,19 +7,21 @@ from rich.console import Console
from rich.tree import Tree
from falyx.action.action import BaseAction
from falyx.action.types import SelectionReturnType
from falyx.context import ExecutionContext
from falyx.execution_registry import ExecutionRegistry as er
from falyx.hook_manager import HookType
from falyx.logger import logger
from falyx.selection import (
SelectionOption,
SelectionOptionMap,
prompt_for_index,
prompt_for_selection,
render_selection_dict_table,
render_selection_indexed_table,
)
from falyx.signals import CancelSignal
from falyx.themes import OneColors
from falyx.utils import CaseInsensitiveDict
class SelectionAction(BaseAction):
@ -34,7 +36,13 @@ class SelectionAction(BaseAction):
def __init__(
self,
name: str,
selections: list[str] | set[str] | tuple[str, ...] | dict[str, SelectionOption],
selections: (
list[str]
| set[str]
| tuple[str, ...]
| dict[str, SelectionOption]
| dict[str, Any]
),
*,
title: str = "Select an option",
columns: int = 5,
@ -42,7 +50,7 @@ class SelectionAction(BaseAction):
default_selection: str = "",
inject_last_result: bool = False,
inject_into: str = "last_result",
return_key: bool = False,
return_type: SelectionReturnType | str = "value",
console: Console | None = None,
prompt_session: PromptSession | None = None,
never_prompt: bool = False,
@ -55,8 +63,8 @@ class SelectionAction(BaseAction):
never_prompt=never_prompt,
)
# Setter normalizes to correct type, mypy can't infer that
self.selections: list[str] | CaseInsensitiveDict = selections # type: ignore[assignment]
self.return_key = return_key
self.selections: list[str] | SelectionOptionMap = selections # type: ignore[assignment]
self.return_type: SelectionReturnType = self._coerce_return_type(return_type)
self.title = title
self.columns = columns
self.console = console or Console(color_system="auto")
@ -65,8 +73,15 @@ class SelectionAction(BaseAction):
self.prompt_message = prompt_message
self.show_table = show_table
def _coerce_return_type(
self, return_type: SelectionReturnType | str
) -> SelectionReturnType:
if isinstance(return_type, SelectionReturnType):
return return_type
return SelectionReturnType(return_type)
@property
def selections(self) -> list[str] | CaseInsensitiveDict:
def selections(self) -> list[str] | SelectionOptionMap:
return self._selections
@selections.setter
@ -74,19 +89,40 @@ class SelectionAction(BaseAction):
self, value: list[str] | set[str] | tuple[str, ...] | dict[str, SelectionOption]
):
if isinstance(value, (list, tuple, set)):
self._selections: list[str] | CaseInsensitiveDict = list(value)
self._selections: list[str] | SelectionOptionMap = list(value)
elif isinstance(value, dict):
cid = CaseInsensitiveDict()
cid.update(value)
self._selections = cid
som = SelectionOptionMap()
if all(isinstance(key, str) for key in value) and all(
not isinstance(value[key], SelectionOption) for key in value
):
som.update(
{
str(index): SelectionOption(key, option)
for index, (key, option) in enumerate(value.items())
}
)
elif all(isinstance(key, str) for key in value) and all(
isinstance(value[key], SelectionOption) for key in value
):
som.update(value)
else:
raise ValueError("Invalid dictionary format. Keys must be strings")
self._selections = som
else:
raise TypeError(
"'selections' must be a list[str] or dict[str, SelectionOption], "
f"got {type(value).__name__}"
)
def get_infer_target(self) -> None:
return None
def _find_cancel_key(self) -> str:
"""Return first numeric value not already used in the selection dict."""
for index in range(len(self.selections)):
if str(index) not in self.selections:
return str(index)
return str(len(self.selections))
def get_infer_target(self) -> tuple[None, None]:
return None, None
async def _run(self, *args, **kwargs) -> Any:
kwargs = self._maybe_inject_last_result(kwargs)
@ -128,16 +164,17 @@ class SelectionAction(BaseAction):
context.start_timer()
try:
cancel_key = self._find_cancel_key()
await self.hooks.trigger(HookType.BEFORE, context)
if isinstance(self.selections, list):
table = render_selection_indexed_table(
title=self.title,
selections=self.selections,
selections=self.selections + ["Cancel"],
columns=self.columns,
)
if not self.never_prompt:
index = await prompt_for_index(
len(self.selections) - 1,
len(self.selections),
table,
default_selection=effective_default,
console=self.console,
@ -147,14 +184,23 @@ class SelectionAction(BaseAction):
)
else:
index = effective_default
result = self.selections[int(index)]
if index == cancel_key:
raise CancelSignal("User cancelled the selection.")
result: Any = self.selections[int(index)]
elif isinstance(self.selections, dict):
cancel_option = {
cancel_key: SelectionOption(
description="Cancel", value=CancelSignal, style=OneColors.DARK_RED
)
}
table = render_selection_dict_table(
title=self.title, selections=self.selections, columns=self.columns
title=self.title,
selections=self.selections | cancel_option,
columns=self.columns,
)
if not self.never_prompt:
key = await prompt_for_selection(
self.selections.keys(),
(self.selections | cancel_option).keys(),
table,
default_selection=effective_default,
console=self.console,
@ -164,10 +210,25 @@ class SelectionAction(BaseAction):
)
else:
key = effective_default
result = key if self.return_key else self.selections[key].value
if key == cancel_key:
raise CancelSignal("User cancelled the selection.")
if self.return_type == SelectionReturnType.KEY:
result = key
elif self.return_type == SelectionReturnType.VALUE:
result = self.selections[key].value
elif self.return_type == SelectionReturnType.ITEMS:
result = {key: self.selections[key]}
elif self.return_type == SelectionReturnType.DESCRIPTION:
result = self.selections[key].description
elif self.return_type == SelectionReturnType.DESCRIPTION_VALUE:
result = {
self.selections[key].description: self.selections[key].value
}
else:
raise ValueError(f"Unsupported return type: {self.return_type}")
else:
raise TypeError(
"'selections' must be a list[str] or dict[str, tuple[str, Any]], "
"'selections' must be a list[str] or dict[str, Any], "
f"got {type(self.selections).__name__}"
)
context.result = result
@ -206,7 +267,7 @@ class SelectionAction(BaseAction):
return
tree.add(f"[dim]Default:[/] '{self.default_selection or self.last_result}'")
tree.add(f"[dim]Return:[/] {'Key' if self.return_key else 'Value'}")
tree.add(f"[dim]Return:[/] {self.return_type.name.capitalize()}")
tree.add(f"[dim]Prompt:[/] {'Disabled' if self.never_prompt else 'Enabled'}")
if not parent:
@ -221,6 +282,6 @@ class SelectionAction(BaseAction):
return (
f"SelectionAction(name={self.name!r}, type={selection_type}, "
f"default_selection={self.default_selection!r}, "
f"return_key={self.return_key}, "
f"return_type={self.return_type!r}, "
f"prompt={'off' if self.never_prompt else 'on'})"
)

15
falyx/action/types.py
View File

@ -35,3 +35,18 @@ class FileReturnType(Enum):
return member
valid = ", ".join(member.value for member in cls)
raise ValueError(f"Invalid FileReturnType: '{value}'. Must be one of: {valid}")
class SelectionReturnType(Enum):
"""Enum for dictionary return types."""
KEY = "key"
VALUE = "value"
DESCRIPTION = "description"
DESCRIPTION_VALUE = "description_value"
ITEMS = "items"
@classmethod
def _missing_(cls, value: object) -> SelectionReturnType:
valid = ", ".join(member.value for member in cls)
raise ValueError(f"Invalid DictReturnType: '{value}'. Must be one of: {valid}")

4
falyx/action/user_input_action.py
View File

@ -43,8 +43,8 @@ class UserInputAction(BaseAction):
self.console = console or Console(color_system="auto")
self.prompt_session = prompt_session or PromptSession()
def get_infer_target(self) -> None:
return None
def get_infer_target(self) -> tuple[None, None]:
return None, None
async def _run(self, *args, **kwargs) -> str:
context = ExecutionContext(

41
falyx/command.py
View File

@ -19,7 +19,6 @@ in building robust interactive menus.
from __future__ import annotations
import shlex
from functools import cached_property
from typing import Any, Callable
from prompt_toolkit.formatted_text import FormattedText
@ -27,8 +26,7 @@ from pydantic import BaseModel, ConfigDict, Field, PrivateAttr, field_validator
from rich.console import Console
from rich.tree import Tree
from falyx.action.action import Action, ActionGroup, BaseAction, ChainedAction
from falyx.action.io_action import BaseIOAction
from falyx.action.action import Action, BaseAction
from falyx.context import ExecutionContext
from falyx.debug import register_debug_hooks
from falyx.execution_registry import ExecutionRegistry as er
@ -90,7 +88,6 @@ class Command(BaseModel):
retry_policy (RetryPolicy): Retry behavior configuration.
tags (list[str]): Organizational tags for the command.
logging_hooks (bool): Whether to attach logging hooks automatically.
requires_input (bool | None): Indicates if the action needs input.
options_manager (OptionsManager): Manages global command-line options.
arg_parser (CommandArgumentParser): Parses command arguments.
custom_parser (ArgParserProtocol | None): Custom argument parser.
@ -129,7 +126,6 @@ class Command(BaseModel):
retry_policy: RetryPolicy = Field(default_factory=RetryPolicy)
tags: list[str] = Field(default_factory=list)
logging_hooks: bool = False
requires_input: bool | None = None
options_manager: OptionsManager = Field(default_factory=OptionsManager)
arg_parser: CommandArgumentParser = Field(default_factory=CommandArgumentParser)
arguments: list[dict[str, Any]] = Field(default_factory=list)
@ -146,7 +142,7 @@ class Command(BaseModel):
def parse_args(
self, raw_args: list[str] | str, from_validate: bool = False
) -> tuple[tuple, dict]:
if self.custom_parser:
if callable(self.custom_parser):
if isinstance(raw_args, str):
try:
raw_args = shlex.split(raw_args)
@ -183,13 +179,15 @@ class Command(BaseModel):
def get_argument_definitions(self) -> list[dict[str, Any]]:
if self.arguments:
return self.arguments
elif self.argument_config:
elif callable(self.argument_config):
self.argument_config(self.arg_parser)
elif self.auto_args:
if isinstance(self.action, BaseAction):
return infer_args_from_func(
self.action.get_infer_target(), self.arg_metadata
)
infer_target, maybe_metadata = self.action.get_infer_target()
# merge metadata with the action's metadata if not already in self.arg_metadata
if maybe_metadata:
self.arg_metadata = {**maybe_metadata, **self.arg_metadata}
return infer_args_from_func(infer_target, self.arg_metadata)
elif callable(self.action):
return infer_args_from_func(self.action, self.arg_metadata)
return []
@ -217,30 +215,9 @@ class Command(BaseModel):
if self.logging_hooks and isinstance(self.action, BaseAction):
register_debug_hooks(self.action.hooks)
if self.requires_input is None and self.detect_requires_input:
self.requires_input = True
self.hidden = True
elif self.requires_input is None:
self.requires_input = False
for arg_def in self.get_argument_definitions():
self.arg_parser.add_argument(*arg_def.pop("flags"), **arg_def)
@cached_property
def detect_requires_input(self) -> bool:
"""Detect if the action requires input based on its type."""
if isinstance(self.action, BaseIOAction):
return True
elif isinstance(self.action, ChainedAction):
return (
isinstance(self.action.actions[0], BaseIOAction)
if self.action.actions
else False
)
elif isinstance(self.action, ActionGroup):
return any(isinstance(action, BaseIOAction) for action in self.action.actions)
return False
def _inject_options_manager(self) -> None:
"""Inject the options manager into the action if applicable."""
if isinstance(self.action, BaseAction):
@ -333,7 +310,7 @@ class Command(BaseModel):
def show_help(self) -> bool:
"""Display the help message for the command."""
if self.custom_help:
if callable(self.custom_help):
output = self.custom_help()
if output:
console.print(output)

1
falyx/config.py
View File

@ -98,7 +98,6 @@ class RawCommand(BaseModel):
retry: bool = False
retry_all: bool = False
retry_policy: RetryPolicy = Field(default_factory=RetryPolicy)
requires_input: bool | None = None
hidden: bool = False
help_text: str = ""

59
falyx/falyx.py
View File

@ -61,7 +61,7 @@ from falyx.options_manager import OptionsManager
from falyx.parsers import CommandArgumentParser, get_arg_parsers
from falyx.protocols import ArgParserProtocol
from falyx.retry import RetryPolicy
from falyx.signals import BackSignal, CancelSignal, FlowSignal, HelpSignal, QuitSignal
from falyx.signals import BackSignal, CancelSignal, HelpSignal, QuitSignal
from falyx.themes import OneColors, get_nord_theme
from falyx.utils import CaseInsensitiveDict, _noop, chunks
from falyx.version import __version__
@ -90,7 +90,7 @@ class CommandValidator(Validator):
if not choice:
raise ValidationError(
message=self.error_message,
cursor_position=document.get_end_of_document_position(),
cursor_position=len(text),
)
@ -111,6 +111,8 @@ class Falyx:
- Submenu nesting and action chaining
- History tracking, help generation, and run key execution modes
- Seamless CLI argument parsing and integration via argparse
- Declarative option management with OptionsManager
- Command level argument parsing and validation
- Extensible with user-defined hooks, bottom bars, and custom layouts
Args:
@ -126,7 +128,7 @@ class Falyx:
never_prompt (bool): Seed default for `OptionsManager["never_prompt"]`
force_confirm (bool): Seed default for `OptionsManager["force_confirm"]`
cli_args (Namespace | None): Parsed CLI arguments, usually from argparse.
options (OptionsManager | None): Declarative option mappings.
options (OptionsManager | None): Declarative option mappings for global state.
custom_table (Callable[[Falyx], Table] | Table | None): Custom menu table
generator.
@ -160,6 +162,7 @@ class Falyx:
options: OptionsManager | None = None,
render_menu: Callable[[Falyx], None] | None = None,
custom_table: Callable[[Falyx], Table] | Table | None = None,
hide_menu_table: bool = False,
) -> None:
"""Initializes the Falyx object."""
self.title: str | Markdown = title
@ -185,6 +188,7 @@ class Falyx:
self.cli_args: Namespace | None = cli_args
self.render_menu: Callable[[Falyx], None] | None = render_menu
self.custom_table: Callable[[Falyx], Table] | Table | None = custom_table
self.hide_menu_table: bool = hide_menu_table
self.validate_options(cli_args, options)
self._prompt_session: PromptSession | None = None
self.mode = FalyxMode.MENU
@ -287,8 +291,6 @@ class Falyx:
for command in self.commands.values():
help_text = command.help_text or command.description
if command.requires_input:
help_text += " [dim](requires input)[/dim]"
table.add_row(
f"[{command.style}]{command.key}[/]",
", ".join(command.aliases) if command.aliases else "",
@ -445,7 +447,6 @@ class Falyx:
bottom_toolbar=self._get_bottom_bar_render(),
key_bindings=self.key_bindings,
validate_while_typing=False,
interrupt_exception=FlowSignal,
)
return self._prompt_session
@ -608,7 +609,6 @@ class Falyx:
retry: bool = False,
retry_all: bool = False,
retry_policy: RetryPolicy | None = None,
requires_input: bool | None = None,
arg_parser: CommandArgumentParser | None = None,
arguments: list[dict[str, Any]] | None = None,
argument_config: Callable[[CommandArgumentParser], None] | None = None,
@ -660,7 +660,6 @@ class Falyx:
retry=retry,
retry_all=retry_all,
retry_policy=retry_policy or RetryPolicy(),
requires_input=requires_input,
options_manager=self.options,
arg_parser=arg_parser,
arguments=arguments or [],
@ -768,26 +767,27 @@ class Falyx:
choice = choice.upper()
name_map = self._name_map
if choice in name_map:
if name_map.get(choice):
if not from_validate:
logger.info("Command '%s' selected.", choice)
if input_args and name_map[choice].arg_parser:
try:
args, kwargs = name_map[choice].parse_args(input_args, from_validate)
except CommandArgumentError as error:
if not from_validate:
if not name_map[choice].show_help():
self.console.print(
f"[{OneColors.DARK_RED}]❌ Invalid arguments for '{choice}': {error}"
)
else:
name_map[choice].show_help()
raise ValidationError(
message=str(error), cursor_position=len(raw_choices)
if is_preview:
return True, name_map[choice], args, kwargs
try:
args, kwargs = name_map[choice].parse_args(input_args, from_validate)
except CommandArgumentError as error:
if not from_validate:
if not name_map[choice].show_help():
self.console.print(
f"[{OneColors.DARK_RED}]❌ Invalid arguments for '{choice}': {error}"
)
return is_preview, None, args, kwargs
except HelpSignal:
return True, None, args, kwargs
else:
name_map[choice].show_help()
raise ValidationError(
message=str(error), cursor_position=len(raw_choices)
)
return is_preview, None, args, kwargs
except HelpSignal:
return True, None, args, kwargs
return is_preview, name_map[choice], args, kwargs
prefix_matches = [cmd for key, cmd in name_map.items() if key.startswith(choice)]
@ -975,10 +975,11 @@ class Falyx:
self.print_message(self.welcome_message)
try:
while True:
if callable(self.render_menu):
self.render_menu(self)
else:
self.console.print(self.table, justify="center")
if not self.hide_menu_table:
if callable(self.render_menu):
self.render_menu(self)
else:
self.console.print(self.table, justify="center")
try:
task = asyncio.create_task(self.process_command())
should_continue = await task

15
falyx/hook_manager.py
View File

@ -4,7 +4,7 @@ from __future__ import annotations
import inspect
from enum import Enum
from typing import Awaitable, Callable, Dict, List, Optional, Union
from typing import Awaitable, Callable, Union
from falyx.context import ExecutionContext
from falyx.logger import logger
@ -24,7 +24,7 @@ class HookType(Enum):
ON_TEARDOWN = "on_teardown"
@classmethod
def choices(cls) -> List[HookType]:
def choices(cls) -> list[HookType]:
"""Return a list of all hook type choices."""
return list(cls)
@ -37,16 +37,17 @@ class HookManager:
"""HookManager"""
def __init__(self) -> None:
self._hooks: Dict[HookType, List[Hook]] = {
self._hooks: dict[HookType, list[Hook]] = {
hook_type: [] for hook_type in HookType
}
def register(self, hook_type: HookType, hook: Hook):
if hook_type not in HookType:
raise ValueError(f"Unsupported hook type: {hook_type}")
def register(self, hook_type: HookType | str, hook: Hook):
"""Raises ValueError if the hook type is not supported."""
if not isinstance(hook_type, HookType):
hook_type = HookType(hook_type)
self._hooks[hook_type].append(hook)
def clear(self, hook_type: Optional[HookType] = None):
def clear(self, hook_type: HookType | None = None):
if hook_type:
self._hooks[hook_type] = []
else:

28
falyx/menu.py
View File

@ -33,7 +33,7 @@ class MenuOptionMap(CaseInsensitiveDict):
and special signal entries like Quit and Back.
"""
RESERVED_KEYS = {"Q", "B"}
RESERVED_KEYS = {"B", "X"}
def __init__(
self,
@ -49,14 +49,14 @@ class MenuOptionMap(CaseInsensitiveDict):
def _inject_reserved_defaults(self):
from falyx.action import SignalAction
self._add_reserved(
"Q",
MenuOption("Exit", SignalAction("Quit", QuitSignal()), OneColors.DARK_RED),
)
self._add_reserved(
"B",
MenuOption("Back", SignalAction("Back", BackSignal()), OneColors.DARK_YELLOW),
)
self._add_reserved(
"X",
MenuOption("Exit", SignalAction("Quit", QuitSignal()), OneColors.DARK_RED),
)
def _add_reserved(self, key: str, option: MenuOption) -> None:
"""Add a reserved key, bypassing validation."""
@ -78,8 +78,20 @@ class MenuOptionMap(CaseInsensitiveDict):
raise ValueError(f"Cannot delete reserved option '{key}'.")
super().__delitem__(key)
def update(self, other=None, **kwargs):
"""Update the selection options with another dictionary."""
if other:
for key, option in other.items():
if not isinstance(option, MenuOption):
raise TypeError(f"Value for key '{key}' must be a SelectionOption.")
self[key] = option
for key, option in kwargs.items():
if not isinstance(option, MenuOption):
raise TypeError(f"Value for key '{key}' must be a SelectionOption.")
self[key] = option
def items(self, include_reserved: bool = True):
for k, v in super().items():
if not include_reserved and k in self.RESERVED_KEYS:
for key, option in super().items():
if not include_reserved and key in self.RESERVED_KEYS:
continue
yield k, v
yield key, option

11
falyx/parsers/argparse.py
View File

@ -1,4 +1,6 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
from __future__ import annotations
from copy import deepcopy
from dataclasses import dataclass
from enum import Enum
@ -23,6 +25,15 @@ class ArgumentAction(Enum):
COUNT = "count"
HELP = "help"
@classmethod
def choices(cls) -> list[ArgumentAction]:
"""Return a list of all argument actions."""
return list(cls)
def __str__(self) -> str:
"""Return the string representation of the argument action."""
return self.value
@dataclass
class Argument:

3
falyx/parsers/utils.py
View File

@ -13,7 +13,8 @@ def same_argument_definitions(
arg_sets = []
for action in actions:
if isinstance(action, BaseAction):
arg_defs = infer_args_from_func(action.get_infer_target(), arg_metadata)
infer_target, _ = action.get_infer_target()
arg_defs = infer_args_from_func(infer_target, arg_metadata)
elif callable(action):
arg_defs = infer_args_from_func(action, arg_metadata)
else:

58
falyx/selection.py
View File

@ -10,7 +10,7 @@ from rich.markup import escape
from rich.table import Table
from falyx.themes import OneColors
from falyx.utils import chunks
from falyx.utils import CaseInsensitiveDict, chunks
from falyx.validators import int_range_validator, key_validator
@ -32,6 +32,62 @@ class SelectionOption:
return f"[{OneColors.WHITE}]{key}[/] [{self.style}]{self.description}[/]"
class SelectionOptionMap(CaseInsensitiveDict):
"""
Manages selection options including validation and reserved key protection.
"""
RESERVED_KEYS: set[str] = set()
def __init__(
self,
options: dict[str, SelectionOption] | None = None,
allow_reserved: bool = False,
):
super().__init__()
self.allow_reserved = allow_reserved
if options:
self.update(options)
def _add_reserved(self, key: str, option: SelectionOption) -> None:
"""Add a reserved key, bypassing validation."""
norm_key = key.upper()
super().__setitem__(norm_key, option)
def __setitem__(self, key: str, option: SelectionOption) -> None:
if not isinstance(option, SelectionOption):
raise TypeError(f"Value for key '{key}' must be a SelectionOption.")
norm_key = key.upper()
if norm_key in self.RESERVED_KEYS and not self.allow_reserved:
raise ValueError(
f"Key '{key}' is reserved and cannot be used in SelectionOptionMap."
)
super().__setitem__(norm_key, option)
def __delitem__(self, key: str) -> None:
if key.upper() in self.RESERVED_KEYS and not self.allow_reserved:
raise ValueError(f"Cannot delete reserved option '{key}'.")
super().__delitem__(key)
def update(self, other=None, **kwargs):
"""Update the selection options with another dictionary."""
if other:
for key, option in other.items():
if not isinstance(option, SelectionOption):
raise TypeError(f"Value for key '{key}' must be a SelectionOption.")
self[key] = option
for key, option in kwargs.items():
if not isinstance(option, SelectionOption):
raise TypeError(f"Value for key '{key}' must be a SelectionOption.")
self[key] = option
def items(self, include_reserved: bool = True):
for k, v in super().items():
if not include_reserved and k in self.RESERVED_KEYS:
continue
yield k, v
def render_table_base(
title: str,
*,

2
falyx/version.py
View File

@ -1 +1 @@
__version__ = "0.1.30"
__version__ = "0.1.31"

2
pyproject.toml
View File

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

96
tests/test_command.py
View File

@ -56,102 +56,6 @@ def test_command_str():
)
@pytest.mark.parametrize(
"action_factory, expected_requires_input",
[
(lambda: Action(name="normal", action=dummy_action), False),
(lambda: DummyInputAction(name="io"), True),
(
lambda: ChainedAction(name="chain", actions=[DummyInputAction(name="io")]),
True,
),
(
lambda: ActionGroup(name="group", actions=[DummyInputAction(name="io")]),
True,
),
],
)
def test_command_requires_input_detection(action_factory, expected_requires_input):
action = action_factory()
cmd = Command(key="TEST", description="Test Command", action=action)
assert cmd.requires_input == expected_requires_input
if expected_requires_input:
assert cmd.hidden is True
else:
assert cmd.hidden is False
def test_requires_input_flag_detected_for_baseioaction():
"""Command should automatically detect requires_input=True for BaseIOAction."""
cmd = Command(
key="X",
description="Echo input",
action=DummyInputAction(name="dummy"),
)
assert cmd.requires_input is True
assert cmd.hidden is True
def test_requires_input_manual_override():
"""Command manually set requires_input=False should not auto-hide."""
cmd = Command(
key="Y",
description="Custom input command",
action=DummyInputAction(name="dummy"),
requires_input=False,
)
assert cmd.requires_input is False
assert cmd.hidden is False
def test_default_command_does_not_require_input():
"""Normal Command without IO Action should not require input."""
cmd = Command(
key="Z",
description="Simple action",
action=lambda: 42,
)
assert cmd.requires_input is False
assert cmd.hidden is False
def test_chain_requires_input():
"""If first action in a chain requires input, the command should require input."""
chain = ChainedAction(
name="ChainWithInput",
actions=[
DummyInputAction(name="dummy"),
Action(name="action1", action=lambda: 1),
],
)
cmd = Command(
key="A",
description="Chain with input",
action=chain,
)
assert cmd.requires_input is True
assert cmd.hidden is True
def test_group_requires_input():
"""If any action in a group requires input, the command should require input."""
group = ActionGroup(
name="GroupWithInput",
actions=[
Action(name="action1", action=lambda: 1),
DummyInputAction(name="dummy"),
],
)
cmd = Command(
key="B",
description="Group with input",
action=group,
)
assert cmd.requires_input is True
assert cmd.hidden is True
def test_enable_retry():
"""Command should enable retry if action is an Action and retry is set to True."""
cmd = Command(