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

Make auto_args default fallback, integrate io_actions with argument parsing

This commit is contained in:
Roland Thomas Jr 2025-05-19 20:03:04 -04:00
parent 4fa6e3bf1f
commit 3c0a81359c
Signed by: roland
GPG Key ID: 7C3C2B085A4C2872
16 changed files with 125 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
examples/auto_args_group.py
View File

@ -24,7 +24,6 @@ cmd = Command(
key="G",
description="Greet someone with multiple variations.",
action=group,
auto_args=True,
arg_metadata={
"name": {
"help": "The name of the person to greet.",

28
examples/auto_parse_demo.py
View File

@ -1,14 +1,18 @@
import asyncio
from falyx import Action, Falyx
from falyx import Action, ChainedAction, Falyx
from falyx.utils import setup_logging
setup_logging()
async def deploy(service: str, region: str = "us-east-1", verbose: bool = False):
async def deploy(service: str, region: str = "us-east-1", verbose: bool = False) -> str:
if verbose:
print(f"Deploying {service} to {region}...")
await asyncio.sleep(2)
if verbose:
print(f"{service} deployed successfully!")
return f"{service} deployed to {region}"
flx = Falyx("Deployment CLI")
@ -21,7 +25,6 @@ flx.add_command(
name="deploy_service",
action=deploy,
),
auto_args=True,
arg_metadata={
"service": "Service name",
"region": {"help": "Deployment region", "choices": ["us-east-1", "us-west-2"]},
@ -29,4 +32,23 @@ flx.add_command(
},
)
deploy_chain = ChainedAction(
name="DeployChain",
actions=[
Action(name="deploy_service", action=deploy),
Action(
name="notify",
action=lambda last_result: print(f"Notification: {last_result}"),
),
],
auto_inject=True,
)
flx.add_command(
key="N",
aliases=["notify"],
description="Deploy a service and notify.",
action=deploy_chain,
)
asyncio.run(flx.run())

48
falyx/action/action.py
View File

@ -47,6 +47,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.utils import same_argument_definitions
from falyx.retry import RetryHandler, RetryPolicy
from falyx.themes import OneColors
from falyx.utils import ensure_async
@ -101,6 +102,14 @@ class BaseAction(ABC):
async def preview(self, parent: Tree | None = None):
raise NotImplementedError("preview must be implemented by subclasses")
@abstractmethod
def get_infer_target(self) -> Callable[..., Any] | None:
"""
Returns the callable to be used for argument inference.
By default, it returns None.
"""
raise NotImplementedError("get_infer_target must be implemented by subclasses")
def set_options_manager(self, options_manager: OptionsManager) -> None:
self.options_manager = options_manager
@ -246,6 +255,13 @@ class Action(BaseAction):
if policy.enabled:
self.enable_retry()
def get_infer_target(self) -> Callable[..., Any]:
"""
Returns the callable to be used for argument inference.
By default, it returns the action itself.
"""
return self.action
async def _run(self, *args, **kwargs) -> Any:
combined_args = args + self.args
combined_kwargs = self._maybe_inject_last_result({**self.kwargs, **kwargs})
@ -477,6 +493,14 @@ 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:
if self.actions:
return self.actions[0].get_infer_target()
return None
def _clear_args(self):
return (), {}
async def _run(self, *args, **kwargs) -> list[Any]:
if not self.actions:
raise EmptyChainError(f"[{self.name}] No actions to execute.")
@ -505,12 +529,8 @@ class ChainedAction(BaseAction, ActionListMixin):
continue
shared_context.current_index = index
prepared = action.prepare(shared_context, self.options_manager)
last_result = shared_context.last_result()
try:
if self.requires_io_injection() and last_result is not None:
result = await prepared(**{prepared.inject_into: last_result})
else:
result = await prepared(*args, **updated_kwargs)
result = await prepared(*args, **updated_kwargs)
except Exception as error:
if index + 1 < len(self.actions) and isinstance(
self.actions[index + 1], FallbackAction
@ -529,6 +549,7 @@ class ChainedAction(BaseAction, ActionListMixin):
fallback._skip_in_chain = True
else:
raise
args, updated_kwargs = self._clear_args()
shared_context.add_result(result)
context.extra["results"].append(result)
context.extra["rollback_stack"].append(prepared)
@ -669,6 +690,16 @@ 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:
arg_defs = same_argument_definitions(self.actions)
if arg_defs:
return self.actions[0].get_infer_target()
logger.debug(
"[%s] auto_args disabled: mismatched ActionGroup arguments",
self.name,
)
return None
async def _run(self, *args, **kwargs) -> list[tuple[str, Any]]:
shared_context = SharedContext(name=self.name, action=self, is_parallel=True)
if self.shared_context:
@ -787,8 +818,11 @@ class ProcessAction(BaseAction):
self.executor = executor or ProcessPoolExecutor()
self.is_retryable = True
async def _run(self, *args, **kwargs):
if self.inject_last_result:
def get_infer_target(self) -> Callable[..., Any] | None:
return self.action
async def _run(self, *args, **kwargs) -> Any:
if self.inject_last_result and self.shared_context:
last_result = self.shared_context.last_result()
if not self._validate_pickleable(last_result):
raise ValueError(

5
falyx/action/action_factory.py
View File

@ -1,6 +1,6 @@
# Falyx CLI Framework — (c) 2025 rtj.dev LLC — MIT Licensed
"""action_factory.py"""
from typing import Any
from typing import Any, Callable
from rich.tree import Tree
@ -55,6 +55,9 @@ class ActionFactoryAction(BaseAction):
def factory(self, value: ActionFactoryProtocol):
self._factory = ensure_async(value)
def get_infer_target(self) -> Callable[..., Any]:
return self.factory
async def _run(self, *args, **kwargs) -> Any:
updated_kwargs = self._maybe_inject_last_result(kwargs)
context = ExecutionContext(

24
falyx/action/io_action.py
View File

@ -19,7 +19,7 @@ import asyncio
import shlex
import subprocess
import sys
from typing import Any
from typing import Any, Callable
from rich.tree import Tree
@ -81,15 +81,15 @@ class BaseIOAction(BaseAction):
def to_output(self, result: Any) -> str | bytes:
raise NotImplementedError
async def _resolve_input(self, kwargs: dict[str, Any]) -> str | bytes:
last_result = kwargs.pop(self.inject_into, None)
async def _resolve_input(
self, args: tuple[Any], kwargs: dict[str, Any]
) -> str | bytes:
data = await self._read_stdin()
if data:
return self.from_input(data)
if last_result is not None:
return last_result
if len(args) == 1:
return self.from_input(args[0])
if self.inject_last_result and self.shared_context:
return self.shared_context.last_result()
@ -99,6 +99,9 @@ class BaseIOAction(BaseAction):
)
raise FalyxError("No input provided and no last result to inject.")
def get_infer_target(self) -> Callable[..., Any] | None:
return None
async def __call__(self, *args, **kwargs):
context = ExecutionContext(
name=self.name,
@ -117,8 +120,8 @@ class BaseIOAction(BaseAction):
pass
result = getattr(self, "_last_result", None)
else:
parsed_input = await self._resolve_input(kwargs)
result = await self._run(parsed_input, *args, **kwargs)
parsed_input = await self._resolve_input(args, kwargs)
result = await self._run(parsed_input)
output = self.to_output(result)
await self._write_stdout(output)
context.result = result
@ -220,6 +223,11 @@ class ShellAction(BaseIOAction):
)
return raw.strip() if isinstance(raw, str) else raw.decode("utf-8").strip()
def get_infer_target(self) -> Callable[..., Any] | None:
if sys.stdin.isatty():
return self._run
return None
async def _run(self, parsed_input: str) -> str:
# Replace placeholder in template, or use raw input as full command
command = self.command_template.format(parsed_input)

3
falyx/action/menu_action.py
View File

@ -73,6 +73,9 @@ class MenuAction(BaseAction):
table.add_row(*row)
return table
def get_infer_target(self) -> None:
return None
async def _run(self, *args, **kwargs) -> Any:
kwargs = self._maybe_inject_last_result(kwargs)
context = ExecutionContext(

3
falyx/action/select_file_action.py
View File

@ -121,6 +121,9 @@ class SelectFileAction(BaseAction):
logger.warning("[ERROR] Failed to parse %s: %s", file.name, error)
return options
def get_infer_target(self) -> None:
return None
async def _run(self, *args, **kwargs) -> Any:
context = ExecutionContext(name=self.name, args=args, kwargs=kwargs, action=self)
context.start_timer()

3
falyx/action/selection_action.py
View File

@ -85,6 +85,9 @@ class SelectionAction(BaseAction):
f"got {type(value).__name__}"
)
def get_infer_target(self) -> None:
return None
async def _run(self, *args, **kwargs) -> Any:
kwargs = self._maybe_inject_last_result(kwargs)
context = ExecutionContext(

3
falyx/action/user_input_action.py
View File

@ -43,6 +43,9 @@ 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
async def _run(self, *args, **kwargs) -> str:
context = ExecutionContext(
name=self.name,

40
falyx/command.py
View File

@ -27,13 +27,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,
ProcessAction,
)
from falyx.action.action import Action, ActionGroup, BaseAction, ChainedAction
from falyx.action.io_action import BaseIOAction
from falyx.context import ExecutionContext
from falyx.debug import register_debug_hooks
@ -41,11 +35,8 @@ 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.parsers import (
CommandArgumentParser,
infer_args_from_func,
same_argument_definitions,
)
from falyx.parsers.argparse import CommandArgumentParser
from falyx.parsers.signature import infer_args_from_func
from falyx.prompt_utils import confirm_async, should_prompt_user
from falyx.protocols import ArgParserProtocol
from falyx.retry import RetryPolicy
@ -116,7 +107,7 @@ class Command(BaseModel):
key: str
description: str
action: BaseAction | Callable[[Any], Any]
action: BaseAction | Callable[..., Any]
args: tuple = ()
kwargs: dict[str, Any] = Field(default_factory=dict)
hidden: bool = False
@ -145,7 +136,7 @@ class Command(BaseModel):
argument_config: Callable[[CommandArgumentParser], None] | None = None
custom_parser: ArgParserProtocol | None = None
custom_help: Callable[[], str | None] | None = None
auto_args: bool = False
auto_args: bool = True
arg_metadata: dict[str, str | dict[str, Any]] = Field(default_factory=dict)
_context: ExecutionContext | None = PrivateAttr(default=None)
@ -195,24 +186,9 @@ class Command(BaseModel):
elif self.argument_config:
self.argument_config(self.arg_parser)
elif self.auto_args:
if isinstance(self.action, (Action, ProcessAction)):
return infer_args_from_func(self.action.action, self.arg_metadata)
elif isinstance(self.action, ChainedAction):
if self.action.actions:
action = self.action.actions[0]
if isinstance(action, Action):
return infer_args_from_func(action.action, self.arg_metadata)
elif callable(action):
return infer_args_from_func(action, self.arg_metadata)
elif isinstance(self.action, ActionGroup):
arg_defs = same_argument_definitions(
self.action.actions, self.arg_metadata
)
if arg_defs:
return arg_defs
logger.debug(
"[Command:%s] auto_args disabled: mismatched ActionGroup arguments",
self.key,
if isinstance(self.action, BaseAction):
return infer_args_from_func(
self.action.get_infer_target(), self.arg_metadata
)
elif callable(self.action):
return infer_args_from_func(self.action, self.arg_metadata)

25
falyx/falyx.py
View File

@ -63,7 +63,7 @@ from falyx.protocols import ArgParserProtocol
from falyx.retry import RetryPolicy
from falyx.signals import BackSignal, CancelSignal, FlowSignal, HelpSignal, QuitSignal
from falyx.themes import OneColors, get_nord_theme
from falyx.utils import CaseInsensitiveDict, _noop, chunks, get_program_invocation
from falyx.utils import CaseInsensitiveDict, _noop, chunks
from falyx.version import __version__
@ -158,8 +158,8 @@ class Falyx:
force_confirm: bool = False,
cli_args: Namespace | None = None,
options: OptionsManager | None = None,
render_menu: Callable[["Falyx"], None] | None = None,
custom_table: Callable[["Falyx"], Table] | Table | None = None,
render_menu: Callable[[Falyx], None] | None = None,
custom_table: Callable[[Falyx], Table] | Table | None = None,
) -> None:
"""Initializes the Falyx object."""
self.title: str | Markdown = title
@ -183,8 +183,8 @@ class Falyx:
self._never_prompt: bool = never_prompt
self._force_confirm: bool = force_confirm
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.render_menu: Callable[[Falyx], None] | None = render_menu
self.custom_table: Callable[[Falyx], Table] | Table | None = custom_table
self.validate_options(cli_args, options)
self._prompt_session: PromptSession | None = None
self.mode = FalyxMode.MENU
@ -526,7 +526,7 @@ class Falyx:
key: str = "X",
description: str = "Exit",
aliases: list[str] | None = None,
action: Callable[[Any], Any] | None = None,
action: Callable[..., Any] | None = None,
style: str = OneColors.DARK_RED,
confirm: bool = False,
confirm_message: str = "Are you sure?",
@ -580,7 +580,7 @@ class Falyx:
self,
key: str,
description: str,
action: BaseAction | Callable[[Any], Any],
action: BaseAction | Callable[..., Any],
*,
args: tuple = (),
kwargs: dict[str, Any] | None = None,
@ -614,7 +614,7 @@ class Falyx:
argument_config: Callable[[CommandArgumentParser], None] | None = None,
custom_parser: ArgParserProtocol | None = None,
custom_help: Callable[[], str | None] | None = None,
auto_args: bool = False,
auto_args: bool = True,
arg_metadata: dict[str, str | dict[str, Any]] | None = None,
) -> Command:
"""Adds an command to the menu, preventing duplicates."""
@ -844,15 +844,6 @@ class Falyx:
await selected_command.preview()
return True
if selected_command.requires_input:
program = get_program_invocation()
self.console.print(
f"[{OneColors.LIGHT_YELLOW}]⚠️ Command '{selected_command.key}' requires"
f" input and must be run via [{OneColors.MAGENTA}]'{program} run"
f"'[{OneColors.LIGHT_YELLOW}] with proper piping or arguments.[/]"
)
return True
self.last_run_command = selected_command
if selected_command == self.exit_command:

4
falyx/parsers/__init__.py
View File

@ -7,8 +7,6 @@ Licensed under the MIT License. See LICENSE file for details.
from .argparse import Argument, ArgumentAction, CommandArgumentParser
from .parsers import FalyxParsers, get_arg_parsers
from .signature import infer_args_from_func
from .utils import same_argument_definitions
__all__ = [
"Argument",
@ -16,6 +14,4 @@ __all__ = [
"CommandArgumentParser",
"get_arg_parsers",
"FalyxParsers",
"infer_args_from_func",
"same_argument_definitions",
]

7
falyx/parsers/signature.py
View File

@ -1,17 +1,20 @@
import inspect
from typing import Any, Callable
from falyx import logger
from falyx.logger import logger
def infer_args_from_func(
func: Callable[[Any], Any],
func: Callable[[Any], Any] | None,
arg_metadata: dict[str, str | dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
"""
Infer argument definitions from a callable's signature.
Returns a list of kwargs suitable for CommandArgumentParser.add_argument.
"""
if not callable(func):
logger.debug("Provided argument is not callable: %s", func)
return []
arg_metadata = arg_metadata or {}
signature = inspect.signature(func)
arg_defs = []

14
falyx/parsers/utils.py
View File

@ -1,7 +1,6 @@
from typing import Any
from falyx import logger
from falyx.action.action import Action, ChainedAction, ProcessAction
from falyx.parsers.signature import infer_args_from_func
@ -9,17 +8,12 @@ def same_argument_definitions(
actions: list[Any],
arg_metadata: dict[str, str | dict[str, Any]] | None = None,
) -> list[dict[str, Any]] | None:
from falyx.action.action import BaseAction
arg_sets = []
for action in actions:
if isinstance(action, (Action, ProcessAction)):
arg_defs = infer_args_from_func(action.action, arg_metadata)
elif isinstance(action, ChainedAction):
if action.actions:
action = action.actions[0]
if isinstance(action, Action):
arg_defs = infer_args_from_func(action.action, arg_metadata)
elif callable(action):
arg_defs = infer_args_from_func(action, arg_metadata)
if isinstance(action, BaseAction):
arg_defs = infer_args_from_func(action.get_infer_target(), arg_metadata)
elif callable(action):
arg_defs = infer_args_from_func(action, arg_metadata)
else:

2
falyx/version.py
View File

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

2
pyproject.toml
View File

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